Merge "[Podified] Run test_list_nodes_processes only on compute nodes"

tobiko/openstack/topology/__init__.py

@@ -65,3 +65,5 @@ verify_osp_version = _topology.verify_osp_version
 get_config_setting = _topology.get_config_setting
 node_name_from_hostname = _topology.node_name_from_hostname
 remove_duplications = _topology.remove_duplications
+
+OpenstackGroupNamesType = _topology.OpenstackGroupNamesType

tobiko/openstack/topology/_topology.py

@@ -45,8 +45,8 @@ LOG = log.getLogger(__name__)
 
 PatternType = type(re.compile(r'.*'))
 OpenstackGroupNameType = typing.Union[str, typing.Pattern]
-OpenstackGroupNamesType = typing.Union[OpenstackGroupNameType, typing.Iterable[
-    OpenstackGroupNameType]]
+OpenstackGroupNamesType = typing.Optional[typing.Union[
+    OpenstackGroupNameType, typing.Iterable[OpenstackGroupNameType]]]
 
 
 HostAddressType = typing.Union[str, netaddr.IPAddress]

tobiko/tests/functional/openstack/test_topology.py

@@ -37,6 +37,8 @@ PatternType = type(re.compile(r''))
 @keystone.skip_unless_has_keystone_credentials()
 class OpenStackTopologyTest(testtools.TestCase):
 
+    expected_group: topology.OpenstackGroupNamesType = None
+
     @property
     def topology(self) -> topology.OpenStackTopology:
         return topology.get_openstack_topology()
@@ -134,7 +136,8 @@ class OpenStackTopologyTest(testtools.TestCase):
         self.assertEqual(expected_nodes, actual_nodes)
 
     def test_list_nodes_processes(self):
-        node = random.choice(topology.list_openstack_nodes())
+        node = random.choice(topology.list_openstack_nodes(
+            group=self.expected_group))
         filename = sh.execute(
             'mktemp', ssh_client=node.ssh_client).stdout.strip()
         self.addCleanup(sh.execute, f"rm -f '{filename}'",

tobiko/tests/functional/podified/test_topology.py

@@ -16,6 +16,7 @@
 from __future__ import absolute_import
 
 import tobiko
+from tobiko.openstack import topology
 from tobiko.shell import ping
 from tobiko.shell import sh
 from tobiko.tests.functional.openstack import test_topology
@@ -25,6 +26,8 @@ from tobiko import podified
 @podified.skip_if_not_podified
 class PodifiedTopologyTest(test_topology.OpenStackTopologyTest):
 
+    expected_group: topology.OpenstackGroupNamesType = 'compute'
+
     @property
     def topology(self) -> podified.PodifiedTopology:
         return tobiko.setup_fixture(podified.PodifiedTopology)