x/tobiko
Code Issues Proposed changes

Fix typing hints for topology connection module

Browse Source 
Change-Id: I67f998e7feae496cd8bdce4259232ca4bbd9f73e
This commit is contained in:
Federico Ressi 2021-06-16 13:28:21 +02:00
parent 770f3dab17
commit b564c4a7c8
1 changed files with 32 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
tobiko/openstack/topology/_connection.py
View File

@ -63,34 +63,43 @@ class SSHConnection(object):
self.ssh_client is not None) self.ssh_client is not None)
SSHConnectionKey = typing.Tuple[netaddr.IPAddress,
typing.Optional[ssh.SSHClientFixture]]
SSHConnectionDict = typing.Dict[SSHConnectionKey, SSHConnection]
class SSHConnectionManager(tobiko.SharedFixture): class SSHConnectionManager(tobiko.SharedFixture):
config = tobiko.required_setup_fixture(_config.OpenStackTopologyConfig) config = tobiko.required_setup_fixture(_config.OpenStackTopologyConfig)
def __init__(self): def __init__(self):
super(SSHConnectionManager, self).__init__() super(SSHConnectionManager, self).__init__()
self._connections: typing.Dict[netaddr.IPAddress, SSHConnection] = ( self._connections: SSHConnectionDict = collections.OrderedDict()
collections.OrderedDict())
def cleanup_fixture(self): def cleanup_fixture(self):
connections = list(self._connections.values()) connections = list(self._connections.values())
self._connections.clear() self._connections.clear()
for connection in connections: for connection in connections:
connection.close() ssh_client = connection.ssh_client
if ssh_client is not None:
ssh_client.close()
def connect(self, def connect(self,
addresses: typing.List[netaddr.IPAddress], addresses: typing.List[netaddr.IPAddress],
proxy_client: typing.Optional[ssh.SSHClientFixture] = None, proxy_client: typing.Optional[ssh.SSHClientFixture] = None,
**connect_parameters) -> ssh.SSHClientFixture: **connect_parameters) \
-> ssh.SSHClientFixture:
if not addresses: if not addresses:
raise ValueError(f"'addresses' list is empty: {addresses}") raise ValueError(f"'addresses' list is empty: {addresses}")
connections = self.list_connections(addresses,
connections = tobiko.select( proxy_client=proxy_client)
self.list_connections(addresses, proxy_client=proxy_client))
try: try:
return connections.with_attributes(is_valid=True).first.ssh_client connection = connections.with_attributes(is_valid=True).first
except tobiko.ObjectNotFound: except tobiko.ObjectNotFound:
pass pass
else:
assert isinstance(connection.ssh_client, ssh.SSHClientFixture)
return connection.ssh_client
for connection in connections.with_attributes(failure=None): for connection in connections.with_attributes(failure=None):
# connection not tried yet # connection not tried yet
@ -119,23 +128,24 @@ class SSHConnectionManager(tobiko.SharedFixture):
failures=failures) failures=failures)
def list_connections( def list_connections(
self, addresses: typing.List[netaddr.IPAddress], self,
proxy_client: typing.Optional[ssh.SSHClientFixture] = None addresses: typing.List[netaddr.IPAddress],
) -> typing.List[SSHConnection]: proxy_client: ssh.SSHClientFixture = None) \
connections = [] -> tobiko.Selection[SSHConnection]:
for address in addresses: # Ensure there is any address duplication
connection = self.get_connection(address, addresses = list(collections.OrderedDict.fromkeys(addresses))
proxy_client=proxy_client) return tobiko.Selection[SSHConnection](
connections.append(connection) self.get_connection(address, proxy_client=proxy_client)
return connections for address in addresses)
def get_connection( def get_connection(
self, address: netaddr.IPAddress, self,
proxy_client: typing.Optional[ssh.SSHClientFixture] = None address: netaddr.IPAddress,
) -> SSHConnection: proxy_client: ssh.SSHClientFixture = None) \
-> SSHConnection:
tobiko.check_valid_type(address, netaddr.IPAddress) tobiko.check_valid_type(address, netaddr.IPAddress)
if proxy_client is not None: tobiko.check_valid_type(proxy_client, ssh.SSHClientFixture,
tobiko.check_valid_type(proxy_client, ssh.SSHClientFixture) type(None))
connection = SSHConnection(address, proxy_client=proxy_client) connection = SSHConnection(address, proxy_client=proxy_client)
return self._connections.setdefault((address, proxy_client), return self._connections.setdefault((address, proxy_client),
connection) connection)