(self, url, proxies=None)
| 218 | self.ssh_client.connect(**self.ssh_params) |
| 219 | |
| 220 | def get_connection(self, url, proxies=None): |
| 221 | if not self.ssh_client: |
| 222 | return SSHConnectionPool( |
| 223 | ssh_client=self.ssh_client, |
| 224 | timeout=self.timeout, |
| 225 | maxsize=self.max_pool_size, |
| 226 | host=self.ssh_host |
| 227 | ) |
| 228 | with self.pools.lock: |
| 229 | pool = self.pools.get(url) |
| 230 | if pool: |
| 231 | return pool |
| 232 | |
| 233 | # Connection is closed try a reconnect |
| 234 | if self.ssh_client and not self.ssh_client.get_transport(): |
| 235 | self._connect() |
| 236 | |
| 237 | pool = SSHConnectionPool( |
| 238 | ssh_client=self.ssh_client, |
| 239 | timeout=self.timeout, |
| 240 | maxsize=self.max_pool_size, |
| 241 | host=self.ssh_host |
| 242 | ) |
| 243 | self.pools[url] = pool |
| 244 | |
| 245 | return pool |
| 246 | |
| 247 | def close(self): |
| 248 | super().close() |
no test coverage detected