(
self,
vnc_port: Optional[int] = None,
port: Optional[int] = None,
require_auth: bool = False,
window_id: Optional[str] = None,
)
| 142 | return self._novnc_password |
| 143 | |
| 144 | def start( |
| 145 | self, |
| 146 | vnc_port: Optional[int] = None, |
| 147 | port: Optional[int] = None, |
| 148 | require_auth: bool = False, |
| 149 | window_id: Optional[str] = None, |
| 150 | ) -> None: |
| 151 | # If stream is already running, throw an error |
| 152 | if self._check_vnc_running(): |
| 153 | raise RuntimeError("Stream is already running") |
| 154 | |
| 155 | # Update parameters if provided |
| 156 | self._vnc_port = vnc_port or self._vnc_port |
| 157 | self._port = port or self._port |
| 158 | self._novnc_auth_enabled = require_auth or self._novnc_auth_enabled |
| 159 | self._novnc_password = self._generate_password() if require_auth else None |
| 160 | |
| 161 | # Update URL with new port |
| 162 | self._url = f"https://{self.__desktop.get_host(self._port)}/vnc.html" |
| 163 | |
| 164 | # Set up VNC command |
| 165 | pwd_flag = "-nopw" |
| 166 | if self._novnc_auth_enabled: |
| 167 | self.__desktop.commands.run("mkdir -p ~/.vnc") |
| 168 | self.__desktop.commands.run( |
| 169 | f"x11vnc -storepasswd {self._novnc_password} ~/.vnc/passwd" |
| 170 | ) |
| 171 | pwd_flag = "-usepw" |
| 172 | |
| 173 | window_id_flag = "" |
| 174 | if window_id: |
| 175 | window_id_flag = f"-id {window_id}" |
| 176 | |
| 177 | vnc_command = ( |
| 178 | f"x11vnc -bg -display {self.__desktop._display} -forever -wait 50 -shared " |
| 179 | f"-rfbport {self._vnc_port} {pwd_flag} 2>/tmp/x11vnc_stderr.log {window_id_flag}" |
| 180 | ) |
| 181 | |
| 182 | novnc_command = ( |
| 183 | f"cd /opt/noVNC/utils && ./novnc_proxy --vnc localhost:{self._vnc_port} " |
| 184 | f"--listen {self._port} --web /opt/noVNC > /tmp/novnc.log 2>&1" |
| 185 | ) |
| 186 | |
| 187 | self.__desktop.commands.run(vnc_command) |
| 188 | |
| 189 | self.__novnc_handle = self.__desktop.commands.run( |
| 190 | novnc_command, background=True, timeout=0 |
| 191 | ) |
| 192 | if not self._wait_for_port(self._port): |
| 193 | raise TimeoutException("Could not start noVNC server") |
| 194 | |
| 195 | def stop(self) -> None: |
| 196 | if self._check_vnc_running(): |
no test coverage detected