frr: T7664: drop BASH/SED implementation in smoketest getFRRconfig()

Rathern then re-inventing the wheel by outself we import the frr-reload.py
script and their read-in function of the current vtysh configuration.

Author: c-po

smoketest/scripts/cli/base_vyostest_shim.py

@@ -12,10 +12,13 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+import importlib.util
 import os
-import unittest
 import paramiko
 import pprint
+import re
+import sys
+import unittest
 
 from time import sleep
 from typing import Type
@@ -36,6 +39,7 @@
 # Using this approach we can not render a live system useless while running any
 # kind of smoketest. In addition it adds debug capabilities like printing the
 # command used to execute the test.
+
 class VyOSUnitTestSHIM:
     class TestCase(unittest.TestCase):
         # if enabled in derived class, print out each and every set/del command
@@ -50,6 +54,17 @@ def debug_on():
 
         @classmethod
         def setUpClass(cls):
+            # Import frr-reload.py functionality
+            file_path = '/usr/lib/frr/frr-reload.py'
+            module_name = 'frr_reload'
+
+            spec = importlib.util.spec_from_file_location(module_name, file_path)
+            module = importlib.util.module_from_spec(spec)
+            sys.modules[module_name] = module
+            spec.loader.exec_module(module)
+            Vtysh = getattr(module, 'Vtysh')
+            cls._vtysh = Vtysh(bindir='/usr/bin', confdir='/etc/frr')
+
             cls._session = ConfigSession(os.getpid())
             cls._session.save_config(save_config)
             cls.debug = cls.debug_on()
@@ -93,8 +108,7 @@ def cli_commit(self):
                 sleep(0.250)
             # Return the output of commit
             # Necessary for testing Warning cases
-            out = self._session.commit()
-            return out
+            return self._session.commit()
 
         def cli_save(self, file):
             if self.debug:
@@ -114,37 +128,52 @@ def op_mode(self, path : list) -> None:
                 pprint.pprint(out)
             return out
 
-        def getFRRconfig(self, string=None, end='$', endsection='^!',
-                         substring=None, endsubsection=None, empty_retry=0):
+        def getFRRconfig(self, start_section:str=None, stop_section='^!',
+                         start_subsection:str=None, stop_subsection='^ exit') -> str:
             """
             Retrieve current "running configuration" from FRR
 
-            string:        search for a specific start string in the configuration
-            end:           end of the section to search for (line ending)
-            endsection:    end of the configuration
-            substring:     search section under the result found by string
-            endsubsection: end of the subsection (usually something with "exit")
+            start_section:    search for a specific start string in the configuration
+            stop_section:     end of the configuration
+            start_subsection: search section under the result found by string
+            stop_subsection:  end of the subsection (usually something with "exit")
             """
-            command = f'vtysh -c "show run no-header"'
-            if string:
-                command += f' | sed -n "/^{string}{end}/,/{endsection}/p"'
-                if substring and endsubsection:
-                    command += f' | sed -n "/^{substring}/,/{endsubsection}/p"'
-            out = cmd(command)
+            frr_config = self._vtysh.mark_show_run()
+            if not start_section:
+                return frr_config
+
+            extracted = []
+            in_section = False
+            for line in frr_config.splitlines():
+                if not in_section:
+                    if re.match(start_section, line):
+                        in_section = True
+                        extracted.append(line)
+                else:
+                    extracted.append(line)
+                    if re.match(stop_section, line):
+                        break
+            output = '\n'.join(extracted)
+
+            # Use extracted list when searching for optional subsection
+            # used by e.g. BGP address-family check
+            if start_subsection:
+                extracted_subsection = []
+                in_subsection = False
+                for line in extracted:
+                    if not in_subsection:
+                        if re.match(start_subsection, line):
+                            in_subsection = True
+                            extracted_subsection.append(line)
+                    else:
+                        extracted_subsection.append(line)
+                        if re.match(stop_subsection, line):
+                            break
+                output = '\n'.join(extracted_subsection)
+
             if self.debug:
-                print(f'\n\ncommand "{command}" returned:\n')
-                pprint.pprint(out)
-            if empty_retry > 0:
-                retry_count = 0
-                while not out and retry_count < empty_retry:
-                    if self.debug and retry_count % 10 == 0:
-                        print(f"Attempt {retry_count}: FRR config is still empty. Retrying...")
-                    retry_count += 1
-                    sleep(1)
-                    out = cmd(command)
-                if not out:
-                    print(f'FRR configuration still empty after {empty_retry} retires!')
-            return out
+                print(output)
+            return output
 
         def getFRRopmode(self, command : str, json : bool=False):
             from json import loads

smoketest/scripts/cli/test_interfaces_bonding.py

@@ -311,7 +311,7 @@ def test_bonding_evpn_multihoming(self):
 
         id = '5'
         for interface in self._interfaces:
-            frrconfig = self.getFRRconfig(f'interface {interface}', endsection='^exit')
+            frrconfig = self.getFRRconfig(f'interface {interface}', stop_section='^exit')
 
             self.assertIn(f' evpn mh es-id {id}', frrconfig)
             self.assertIn(f' evpn mh es-df-pref {id}', frrconfig)
@@ -328,7 +328,7 @@ def test_bonding_evpn_multihoming(self):
 
         id = '5'
         for interface in self._interfaces:
-            frrconfig = self.getFRRconfig(f'interface {interface}', endsection='^exit')
+            frrconfig = self.getFRRconfig(f'interface {interface}', stop_section='^exit')
             self.assertIn(f' evpn mh es-sys-mac 00:12:34:56:78:0{id}', frrconfig)
             self.assertIn(f' evpn mh uplink', frrconfig)
 

smoketest/scripts/cli/test_interfaces_ethernet.py

@@ -229,7 +229,7 @@ def test_ethtool_evpn_uplink_tracking(self):
         self.cli_commit()
 
         for interface in self._interfaces:
-            frrconfig = self.getFRRconfig(f'interface {interface}', endsection='^exit')
+            frrconfig = self.getFRRconfig(f'interface {interface}', stop_section='^exit')
             self.assertIn(' evpn mh uplink', frrconfig)
 
     def test_switchdev(self):

smoketest/scripts/cli/test_interfaces_pppoe.py

@@ -305,7 +305,7 @@ def test_pppoe_dhcpv6pd(self):
                 # Get corresponding PD assigned prefix for this site/connection
                 pd_prefix = IPv6Network(f"{ipv6}/56", strict=False)
                 # Prefix must be within the PD pool
-                self.assertIn(pd_prefix, IPv6Network(ipv6_pool_pd))
+                self.assertTrue(pd_prefix.subnet_of(IPv6Network(ipv6_pool_pd)))
                 # Calculate the assigned IP address from the prefix delegation
                 network_addr = str(pd_prefix.network_address) # 2001:db8:8003::
                 generated_sla_addr = network_addr[:-1] + f'{sla_id}::{address}'

smoketest/scripts/cli/test_policy.py

@@ -121,7 +121,7 @@ def test_access_list(self):
 
         self.cli_commit()
 
-        config = self.getFRRconfig('access-list', end='')
+        config = self.getFRRconfig('access-list')
         for acl, acl_config in acls.items():
             for rule, rule_config in acl_config['rule'].items():
                 tmp = f'access-list {acl} seq {rule}'
@@ -212,7 +212,7 @@ def test_access_list6(self):
 
         self.cli_commit()
 
-        config = self.getFRRconfig('ipv6 access-list', end='')
+        config = self.getFRRconfig('ipv6 access-list')
         for acl, acl_config in acls.items():
             for rule, rule_config in acl_config['rule'].items():
                 tmp = f'ipv6 access-list {acl} seq {rule}'
@@ -310,7 +310,7 @@ def test_as_path_list(self):
 
         self.cli_commit()
 
-        config = self.getFRRconfig('bgp as-path access-list', end='')
+        config = self.getFRRconfig('bgp as-path access-list')
         for as_path, as_path_config in test_data.items():
             if 'rule' not in as_path_config:
                 continue
@@ -368,7 +368,7 @@ def test_community_list(self):
 
         self.cli_commit()
 
-        config = self.getFRRconfig('bgp community-list', end='')
+        config = self.getFRRconfig('bgp community-list')
         for comm_list, comm_list_config in test_data.items():
             if 'rule' not in comm_list_config:
                 continue
@@ -426,7 +426,7 @@ def test_extended_community_list(self):
 
         self.cli_commit()
 
-        config = self.getFRRconfig('bgp extcommunity-list', end='')
+        config = self.getFRRconfig('bgp extcommunity-list')
         for comm_list, comm_list_config in test_data.items():
             if 'rule' not in comm_list_config:
                 continue
@@ -491,7 +491,7 @@ def test_large_community_list(self):
 
         self.cli_commit()
 
-        config = self.getFRRconfig('bgp large-community-list', end='')
+        config = self.getFRRconfig('bgp large-community-list')
         for comm_list, comm_list_config in test_data.items():
             if 'rule' not in comm_list_config:
                 continue
@@ -569,7 +569,7 @@ def test_prefix_list(self):
 
         self.cli_commit()
 
-        config = self.getFRRconfig('ip prefix-list', end='')
+        config = self.getFRRconfig('ip prefix-list')
         for prefix_list, prefix_list_config in test_data.items():
             if 'rule' not in prefix_list_config:
                 continue
@@ -652,7 +652,7 @@ def test_prefix_list6(self):
 
         self.cli_commit()
 
-        config = self.getFRRconfig('ipv6 prefix-list', end='')
+        config = self.getFRRconfig('ipv6 prefix-list')
         for prefix_list, prefix_list_config in test_data.items():
             if 'rule' not in prefix_list_config:
                 continue
@@ -703,7 +703,7 @@ def test_prefix_list_duplicates(self):
 
         self.cli_commit()
 
-        config = self.getFRRconfig('ip prefix-list', end='')
+        config = self.getFRRconfig('ip prefix-list')
         for rule in test_range:
             tmp = f'ip prefix-list {prefix_list} seq {rule} permit {prefix} le {rule}'
             self.assertIn(tmp, config)
@@ -2002,7 +2002,7 @@ def test_frr_individual_remove_T6283_T6250(self):
         local_preference = base_local_preference
         table = base_table
         for route_map in route_maps:
-            config = self.getFRRconfig(f'route-map {route_map} permit {seq}', end='', endsection='^exit')
+            config = self.getFRRconfig(f'route-map {route_map} permit {seq}', stop_section='^exit')
             self.assertIn(f' set local-preference {local_preference}', config)
             self.assertIn(f' set table {table}', config)
             local_preference += 20
@@ -2015,7 +2015,7 @@ def test_frr_individual_remove_T6283_T6250(self):
 
         local_preference = base_local_preference
         for route_map in route_maps:
-            config = self.getFRRconfig(f'route-map {route_map} permit {seq}', end='', endsection='^exit')
+            config = self.getFRRconfig(f'route-map {route_map} permit {seq}', stop_section='^exit')
             self.assertIn(f' set local-preference {local_preference}', config)
             local_preference += 20
 
@@ -2029,7 +2029,7 @@ def test_frr_individual_remove_T6283_T6250(self):
         self.cli_commit()
 
         for route_map in route_maps:
-            config = self.getFRRconfig(f'route-map {route_map} permit {seq}', end='', endsection='^exit')
+            config = self.getFRRconfig(f'route-map {route_map} permit {seq}', stop_section='^exit')
             self.assertIn(f' set as-path prepend {prepend}', config)
 
         for route_map in route_maps:
@@ -2038,7 +2038,7 @@ def test_frr_individual_remove_T6283_T6250(self):
             self.cli_commit()
 
         for route_map in route_maps:
-            config = self.getFRRconfig(f'route-map {route_map} permit {seq}', end='', endsection='^exit')
+            config = self.getFRRconfig(f'route-map {route_map} permit {seq}', stop_section='^exit')
             self.assertNotIn(f' set', config)
 
 def sort_ip(output):

smoketest/scripts/cli/test_protocols_babel.py

@@ -62,7 +62,7 @@ def test_01_basic(self):
 
         self.cli_commit()
 
-        frrconfig = self.getFRRconfig('router babel', endsection='^exit')
+        frrconfig = self.getFRRconfig('router babel', stop_section='^exit')
         self.assertIn(f' babel diversity', frrconfig)
         self.assertIn(f' babel diversity-factor {diversity_factor}', frrconfig)
         self.assertIn(f' babel resend-delay {resend_delay}', frrconfig)
@@ -81,7 +81,7 @@ def test_02_redistribute(self):
 
         self.cli_commit()
 
-        frrconfig = self.getFRRconfig('router babel', endsection='^exit', empty_retry=5)
+        frrconfig = self.getFRRconfig('router babel', stop_section='^exit')
         for protocol in ipv4_protos:
             self.assertIn(f' redistribute ipv4 {protocol}', frrconfig)
         for protocol in ipv6_protos:
@@ -150,7 +150,7 @@ def test_03_distribute_list(self):
 
         self.cli_commit()
 
-        frrconfig = self.getFRRconfig('router babel', endsection='^exit')
+        frrconfig = self.getFRRconfig('router babel', stop_section='^exit')
         self.assertIn(f' distribute-list {access_list_in4} in', frrconfig)
         self.assertIn(f' distribute-list {access_list_out4} out', frrconfig)
         self.assertIn(f' ipv6 distribute-list {access_list_in6} in', frrconfig)
@@ -198,11 +198,11 @@ def test_04_interfaces(self):
 
         self.cli_commit()
 
-        frrconfig = self.getFRRconfig('router babel', endsection='^exit')
+        frrconfig = self.getFRRconfig('router babel', stop_section='^exit')
         for interface in self._interfaces:
             self.assertIn(f' network {interface}', frrconfig)
 
-            iface_config = self.getFRRconfig(f'interface {interface}', endsection='^exit')
+            iface_config = self.getFRRconfig(f'interface {interface}', stop_section='^exit')
             self.assertIn(f' babel channel {channel}', iface_config)
             self.assertIn(f' babel enable-timestamps', iface_config)
             self.assertIn(f' babel update-interval {def_update_interval}', iface_config)

smoketest/scripts/cli/test_protocols_bfd.py

@@ -131,7 +131,7 @@ def test_bfd_peer(self):
         self.cli_commit()
 
         # Verify FRR bgpd configuration
-        frrconfig = self.getFRRconfig('bfd', endsection='^exit')
+        frrconfig = self.getFRRconfig('bfd', stop_section='^exit')
         for peer, peer_config in peers.items():
             tmp = f'peer {peer}'
             if 'multihop' in peer_config:
@@ -144,8 +144,9 @@ def test_bfd_peer(self):
                 tmp += f' vrf {peer_config["vrf"]}'
 
             self.assertIn(tmp, frrconfig)
-            peerconfig = self.getFRRconfig('bfd', endsection='^exit', substring=f' peer {peer}',
-                                           endsubsection='^ exit')
+            peerconfig = self.getFRRconfig('bfd', stop_section='^exit',
+                                           start_subsection=f' peer {peer}',
+                                           stop_subsection='^ exit')
             if 'echo_mode' in peer_config:
                 self.assertIn(f'echo-mode', peerconfig)
             if 'intv_echo' in peer_config:
@@ -207,8 +208,8 @@ def test_bfd_profile(self):
 
         # Verify FRR bgpd configuration
         for profile, profile_config in profiles.items():
-            config = self.getFRRconfig('bfd', endsection='^exit',
-                                       substring=f' profile {profile}', endsubsection='^ exit',)
+            config = self.getFRRconfig('bfd', stop_section='^exit',
+                                       start_subsection=f' profile {profile}', stop_subsection='^ exit',)
             if 'echo_mode' in profile_config:
                 self.assertIn(f' echo-mode', config)
             if 'intv_echo' in profile_config:
@@ -230,8 +231,8 @@ def test_bfd_profile(self):
                 self.assertNotIn(f'shutdown', config)
 
         for peer, peer_config in peers.items():
-            peerconfig = self.getFRRconfig('bfd', endsection='^exit',
-                                           substring=f' peer {peer}', endsubsection='^ exit')
+            peerconfig = self.getFRRconfig('bfd', stop_section='^exit',
+                                           start_subsection=f' peer {peer}', stop_subsection='^ exit')
             if 'profile' in peer_config:
                 self.assertIn(f' profile {peer_config["profile"]}', peerconfig)