Get high level get_l3_agent_mode topology function

Change-Id: I614d72b7877464a7663b497aecc0e8b5a6e2b6b2

tobiko/openstack/topology/__init__.py

@@ -42,6 +42,7 @@ UnknowOpenStackContainerNameError = _topology.UnknowOpenStackContainerNameError
 UnknowOpenStackServiceNameError = _topology.UnknowOpenStackServiceNameError
 get_agent_service_name = _topology.get_agent_service_name
 get_agent_container_name = _topology.get_agent_container_name
+get_l3_agent_mode = _topology.get_l3_agent_mode
 get_log_file_digger = _topology.get_log_file_digger
 get_openstack_topology = _topology.get_openstack_topology
 get_openstack_node = _topology.get_openstack_node

tobiko/openstack/topology/_address.py

@@ -47,6 +47,8 @@ def list_addresses(obj,
         addresses = tobiko.Selection()
         for item in iter(obj):
             addresses.extend(list_addresses(item))
+    else:
+        raise TypeError(f'{obj} object is not of a valid address(es) type')
 
     if addresses and ip_version is not None:
         addresses = addresses.with_attributes(version=ip_version)

tobiko/openstack/topology/_topology.py

@@ -78,9 +78,26 @@ def find_openstack_node(topology=None, unique=False, **kwargs):
         return nodes.first
 
 
-def get_openstack_node(hostname, address=None, topology=None):
-    topology = topology or get_openstack_topology()
-    return topology.get_node(hostname=hostname, address=address)
+def get_l3_agent_mode(name: str = None,
+                      hostname: str = None,
+                      address=None,
+                      topology: 'OpenStackTopology' = None):
+    return get_openstack_node(name=name,
+                              hostname=hostname,
+                              address=address,
+                              topology=topology).l3_agent_mode
+
+
+def get_openstack_node(name: str = None,
+                       hostname: str = None,
+                       address=None,
+                       topology: 'OpenStackTopology' = None) \
+        -> 'OpenStackTopologyNode':
+    if topology is None:
+        topology = get_openstack_topology()
+    return topology.get_node(name=name,
+                             hostname=hostname,
+                             address=address)
 
 
 def list_openstack_node_groups(topology=None):
@@ -147,6 +164,14 @@ class OpenStackTopologyNode(object):
         self.addresses: typing.List[netaddr.IPAddress] = list(addresses)
         self.hostname: str = hostname
 
+    _connection: typing.Optional[sh.ShellConnection] = None
+
+    @property
+    def connection(self) -> sh.ShellConnection:
+        if self._connection is None:
+            self._connection = sh.shell_connection(self.ssh_client)
+        return self._connection
+
     @property
     def topology(self):
         return self._topology()
@@ -180,6 +205,23 @@ class OpenStackTopologyNode(object):
                 ssh_client=self.ssh_client)
         return podman_client
 
+    l3_agent_conf_path = '/etc/neutron/l3_agent.ini'
+    _l3_agent_mode: typing.Optional[str] = None
+
+    @property
+    def l3_agent_mode(self) -> typing.Optional[str]:
+        if self._l3_agent_mode is None:
+            try:
+                self._l3_agent_mode = neutron.get_l3_agent_mode(
+                    l3_agent_conf_path=self.l3_agent_conf_path,
+                    connection=self.connection)
+            except sh.ShellCommandFailed:
+                LOG.debug('Unable to read L3 agent mode for host '
+                          f'{self.hostname}. Assuming legacy mode.',
+                          exc_info=1)
+                self._l3_agent_mode = 'legacy'
+        return self._l3_agent_mode
+
     def __repr__(self):
         return "{cls!s}<name={name!r}>".format(cls=type(self).__name__,
                                                name=self.name)
@@ -383,17 +425,22 @@ class OpenStackTopology(tobiko.SharedFixture):
                           f"used by node '{address_node.name}'")
         return node
 
-    def get_node(self, name=None, hostname=None, address=None):
-        name = name or (hostname and node_name_from_hostname(hostname))
+    def get_node(self,
+                 name: str = None,
+                 hostname: str = None,
+                 address=None) \
+            -> 'OpenStackTopologyNode':
+        if name is None and hostname is not None:
+            name = node_name_from_hostname(hostname)
         details = {}
-        if name:
+        if name is not None:
             tobiko.check_valid_type(name, str)
             details['name'] = name
             try:
                 return self._names[name]
             except KeyError:
                 pass
-        if address:
+        if address is not None:
             details['address'] = address
             for address in self._list_addresses(address):
                 try:

tobiko/tests/functional/openstack/test_topology.py

@@ -24,6 +24,7 @@ import testtools
 
 import tobiko
 from tobiko.openstack import keystone
+from tobiko.openstack import neutron
 from tobiko.openstack import nova
 from tobiko.openstack import topology
 from tobiko.shell import ping
@@ -169,6 +170,18 @@ class OpenStackTopologyTest(testtools.TestCase):
             if attempt.is_last:
                 self.fail("Process listed after killing it")
 
+    @neutron.skip_unless_is_ovs()
+    def test_l3_agent_mode(self):
+        for node in topology.list_openstack_nodes(
+                group=['controller', 'compute', 'overcloud']):
+            assert isinstance(node, topology.OpenStackTopologyNode)
+            self.assertEqual(
+                neutron.get_l3_agent_mode(
+                    l3_agent_conf_path=node.l3_agent_conf_path,
+                    connection=node.connection),
+                node.l3_agent_mode)
+            self.assertIs(node.l3_agent_mode, node.l3_agent_mode)
+
 
 def node_name_from_hostname(hostname):
     return hostname.split('.', 1)[0].lower()

tobiko/tripleo/_topology.py

@@ -139,6 +139,9 @@ class TripleoTopologyNode(topology.OpenStackTopologyNode):
 
     overcloud_server: typing.Optional[nova.NovaServer] = None
 
+    l3_agent_conf_path = (
+        '/var/lib/config-data/neutron/etc/neutron/l3_agent.ini')
+
     def reboot_overcloud_node(self, reactivate_servers=True):
         """Reboot overcloud node