diff --git a/tobiko/openstack/stacks/__init__.py b/tobiko/openstack/stacks/__init__.py index 758d35007..43b4661bf 100644 --- a/tobiko/openstack/stacks/__init__.py +++ b/tobiko/openstack/stacks/__init__.py @@ -79,6 +79,7 @@ UbuntuImageFixture = _ubuntu.UbuntuImageFixture UbuntuMinimalImageFixture = _ubuntu.UbuntuMinimalImageFixture UbuntuServerStackFixture = _ubuntu.UbuntuServerStackFixture UbuntuMinimalServerStackFixture = _ubuntu.UbuntuMinimalServerStackFixture +UbuntuExternalServerStackFixture = _ubuntu.UbuntuExternalServerStackFixture OctaviaLoadbalancerStackFixture = _octavia.OctaviaLoadbalancerStackFixture OctaviaListenerStackFixture = _octavia.OctaviaListenerStackFixture diff --git a/tobiko/openstack/stacks/_nova.py b/tobiko/openstack/stacks/_nova.py index d83094689..418ab0052 100644 --- a/tobiko/openstack/stacks/_nova.py +++ b/tobiko/openstack/stacks/_nova.py @@ -98,7 +98,12 @@ class ServerStackFixture(heat.HeatStackFixture, abc.ABC): network_stack = tobiko.required_setup_fixture(_neutron.NetworkStackFixture) #: whenever the server relies only on DHCP for address assignation - need_dhcp = False + @property + def need_dhcp(self) -> bool: + return not self.config_drive + + #: whenever the server will use config-drive to get metadata + config_drive = False def create_stack(self, retry=None): self.ensure_quota_limits() @@ -172,6 +177,13 @@ class ServerStackFixture(heat.HeatStackFixture, abc.ABC): password=self.password, connection_timeout=self.connection_timeout) + @property + def peer_ssh_client(self) -> typing.Optional[ssh.SSHClientFixture]: + """Nearest SSH client to an host that can see server fixed IPs ports + + """ + return self.ssh_client + @property def ssh_command(self) -> sh.ShellCommand: return ssh.ssh_command(host=self.ip_address, @@ -386,6 +398,8 @@ class ExternalServerStackFixture(ServerStackFixture, abc.ABC): config_drive = True + peer_ssh_client = None + @property def floating_network(self): return self.network_stack.network_id @@ -408,7 +422,11 @@ class PeerServerStackFixture(ServerStackFixture, abc.ABC): username=self.username, password=self.password, connection_timeout=self.connection_timeout, - proxy_jump=self.peer_stack.ssh_client) + proxy_jump=self.peer_ssh_client) + + @property + def peer_ssh_client(self) -> ssh.SSHClientFixture: + return self.peer_stack.ssh_client @property def ssh_command(self) -> sh.ShellCommand: diff --git a/tobiko/openstack/stacks/_ubuntu.py b/tobiko/openstack/stacks/_ubuntu.py index 98e8bb809..8e59b8a00 100644 --- a/tobiko/openstack/stacks/_ubuntu.py +++ b/tobiko/openstack/stacks/_ubuntu.py @@ -80,3 +80,8 @@ class UbuntuMinimalServerStackFixture(UbuntuServerStackFixture): #: Setup SWAP file in bytes swap_maxsize = 512 * 1024 * 1024 # 500 MB + + +class UbuntuExternalServerStackFixture(UbuntuMinimalServerStackFixture, + _nova.ExternalServerStackFixture): + pass diff --git a/tobiko/tests/scenario/neutron/test_port.py b/tobiko/tests/scenario/neutron/test_port.py index eb98a640e..dd2596cd6 100644 --- a/tobiko/tests/scenario/neutron/test_port.py +++ b/tobiko/tests/scenario/neutron/test_port.py @@ -64,9 +64,10 @@ class PortTest(testtools.TestCase): except self.failureException: attempt.check_limits() elif ip_version: - self.skipTest(f"No port IPv{ip_version} addresses found") + self.skipTest(f"Server has any port IPv{ip_version} address to be" + " tested") else: - self.skipTest("No port IP addresses found") + self.skipTest("Server has any port IP address to be tested") def test_port_network(self): self.assertEqual(self.stack.network_stack.network_id, @@ -93,21 +94,19 @@ class PortTest(testtools.TestCase): def test_ping_port(self, network_id=None, device_id=None, ip_version=None): """Checks server can ping its own port""" - device_ips = neutron.list_device_ip_addresses( + port_ips = neutron.list_device_ip_addresses( device_id=device_id or self.stack.server_id, network_id=network_id or self.stack.network_stack.network_id, - enable_dhcp=True, ip_version=ip_version) - server_ips = ip.list_ip_addresses(scope='global', - ssh_client=self.stack.ssh_client) - # Remove IPs that hasn't been assigned to server - port_ips = tobiko.Selection(set(device_ips) & set(server_ips)) + need_dhcp=self.stack.need_dhcp, ip_version=ip_version) if port_ips: ping.assert_reachable_hosts(port_ips, - ssh_client=self.stack.ssh_client) + timeout=600., + ssh_client=self.stack.peer_ssh_client) elif ip_version: - self.skipTest(f"No port IPv{ip_version} addresses found") + self.skipTest(f"Server has any port IPv{ip_version} address to be" + " tested") else: - self.skipTest("No port IP addresses found") + self.skipTest("Server has any port IP address to be tested") # --- Test opening ports on external network ---------------------------------- @@ -130,6 +129,15 @@ class CentosExternalPortTest(PortTest): stacks.CentosExternalServerStackFixture) +@stacks.skip_unless_has_external_network +class UbuntuExternalPortTest(PortTest): + """Test Neutron ports""" + + #: Resources stack with Nova server to send messages to + stack = tobiko.required_setup_fixture( + stacks.UbuntuExternalServerStackFixture) + + # --- Test la-h3 extension ---------------------------------------------------- @neutron.skip_if_missing_networking_extensions('l3-ha')