(self)
| 134 | return self.runtime.provider == "ssh" |
| 135 | |
| 136 | def _connect(self) -> None: |
| 137 | if not self._uses_ssh(): |
| 138 | return |
| 139 | if self.client is not None and self.session is not None: |
| 140 | return |
| 141 | |
| 142 | try: |
| 143 | import paramiko |
| 144 | except ImportError as exc: |
| 145 | raise RuntimeError( |
| 146 | "Paramiko is required for InteractiveShell. " |
| 147 | "Install project dependencies before using the terminal tool." |
| 148 | ) from exc |
| 149 | |
| 150 | client = paramiko.SSHClient() |
| 151 | client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) |
| 152 | client.connect( |
| 153 | self.runtime.ssh.host, |
| 154 | username=self.runtime.ssh.username, |
| 155 | password=self.runtime.ssh.password, |
| 156 | port=self.runtime.ssh.port, |
| 157 | timeout=self.runtime.ssh.timeout, |
| 158 | ) |
| 159 | self.client = client |
| 160 | self.session = client.invoke_shell() |
| 161 | self._drain_pending_output() |
| 162 | |
| 163 | def _drain_pending_output(self, idle_window: float = 0.2) -> None: |
| 164 | if self.session is None: |
no test coverage detected