diff options
Diffstat (limited to 'smoketest/scripts/cli/base_vyostest_shim.py')
| -rw-r--r-- | smoketest/scripts/cli/base_vyostest_shim.py | 81 |
1 files changed, 71 insertions, 10 deletions
diff --git a/smoketest/scripts/cli/base_vyostest_shim.py b/smoketest/scripts/cli/base_vyostest_shim.py index 940306ac3..f0674f187 100644 --- a/smoketest/scripts/cli/base_vyostest_shim.py +++ b/smoketest/scripts/cli/base_vyostest_shim.py @@ -29,6 +29,14 @@ from vyos.utils.process import run save_config = '/tmp/vyos-smoketest-save' +# The commit process is not finished until all pending files from +# VYATTA_CHANGES_ONLY_DIR are copied to VYATTA_ACTIVE_CONFIGURATION_DIR. This +# is done inside libvyatta-cfg1 and the FUSE UnionFS part. On large non- +# interactive commits FUSE UnionFS might not replicate the real state in time, +# leading to errors when querying the working and effective configuration. +# TO BE DELETED AFTER SWITCH TO IN MEMORY CONFIG +CSTORE_GUARD_TIME = 4 + # This class acts as shim between individual Smoketests developed for VyOS and # the Python UnitTest framework. Before every test is loaded, we dump the current # system configuration and reload it after the test - despite the test results. @@ -43,7 +51,9 @@ class VyOSUnitTestSHIM: # trigger the certain failure condition. # Use "self.debug = True" in derived classes setUp() method debug = False - + # Time to wait after a commit to ensure the CStore is up to date + # only required for testcases using FRR + _commit_guard_time = 0 @classmethod def setUpClass(cls): cls._session = ConfigSession(os.getpid()) @@ -65,10 +75,11 @@ class VyOSUnitTestSHIM: cls._session.discard() cls.fail(cls) - def cli_set(self, config): + def cli_set(self, path, value=None): if self.debug: - print('set ' + ' '.join(config)) - self._session.set(config) + str = f'set {" ".join(path)} {value}' if value else f'set {" ".join(path)}' + print(str) + self._session.set(path, value) def cli_delete(self, config): if self.debug: @@ -83,10 +94,17 @@ class VyOSUnitTestSHIM: def cli_commit(self): if self.debug: print('commit') - self._session.commit() - # during a commit there is a process opening commit_lock, and run() returns 0 + # During a commit there is a process opening commit_lock, and run() + # returns 0 while run(f'sudo lsof -nP {commit_lock}') == 0: sleep(0.250) + # Return the output of commit + # Necessary for testing Warning cases + out = self._session.commit() + # Wait for CStore completion for fast non-interactive commits + sleep(self._commit_guard_time) + + return out def op_mode(self, path : list) -> None: """ @@ -101,14 +119,36 @@ class VyOSUnitTestSHIM: pprint.pprint(out) return out - def getFRRconfig(self, string=None, end='$', endsection='^!', daemon=''): - """ Retrieve current "running configuration" from FRR """ - command = f'vtysh -c "show run {daemon} no-header"' - if string: command += f' | sed -n "/^{string}{end}/,/{endsection}/p"' + def getFRRconfig(self, string=None, end='$', endsection='^!', + substring=None, endsubsection=None, empty_retry=0): + """ + 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") + """ + 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) 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 @staticmethod @@ -147,6 +187,27 @@ class VyOSUnitTestSHIM: break self.assertTrue(not matched if inverse else matched, msg=search) + def verify_nftables_chain_exists(self, table, chain, inverse=False): + try: + cmd(f'sudo nft list chain {table} {chain}') + if inverse: + self.fail(f'Chain exists: {table} {chain}') + except OSError: + if not inverse: + self.fail(f'Chain does not exist: {table} {chain}') + + # Verify ip rule output + def verify_rules(self, rules_search, inverse=False, addr_family='inet'): + rule_output = cmd(f'ip -family {addr_family} rule show') + + for search in rules_search: + matched = False + for line in rule_output.split("\n"): + if all(item in line for item in search): + matched = True + break + self.assertTrue(not matched if inverse else matched, msg=search) + # standard construction; typing suggestion: https://stackoverflow.com/a/70292317 def ignore_warning(warning: Type[Warning]): import warnings |