diff options
Diffstat (limited to 'tests/topotests/munet/mutest/userapi.py')
| -rw-r--r-- | tests/topotests/munet/mutest/userapi.py | 184 |
1 files changed, 129 insertions, 55 deletions
diff --git a/tests/topotests/munet/mutest/userapi.py b/tests/topotests/munet/mutest/userapi.py index 1df8c0d..7967dd0 100644 --- a/tests/topotests/munet/mutest/userapi.py +++ b/tests/topotests/munet/mutest/userapi.py @@ -144,7 +144,6 @@ class TestCase: result_logger: logging.Logger = None, full_summary: bool = False, ): - self.info = TestCaseInfo(tag, name, path) self.__saved_info = [] self.__short_doc_header = not full_summary @@ -248,7 +247,6 @@ class TestCase: self.rlog.info("%s. %s", tag, header) def __exec_script(self, path, print_header, add_newline): - # Below was the original method to avoid the global TestCase # variable; however, we need global functions so we can import them # into test scripts. Without imports pylint will complain about undefined @@ -393,12 +391,12 @@ class TestCase: self, target: str, cmd: str, - ) -> dict: + ) -> Union[list, dict]: """Execute a json ``cmd`` and return json result. Args: target: the target to execute the command on. - cmd: string to execut on the target. + cmd: string to execute on the target. """ out = self.targets[target].cmd_nostatus(cmd, warn=False) self.last = out = out.rstrip() @@ -420,6 +418,7 @@ class TestCase: match: str, expect_fail: bool, flags: int, + exact_match: bool, ) -> (bool, Union[str, list]): """Execute a ``cmd`` and check result. @@ -429,6 +428,8 @@ class TestCase: match: regex to ``re.search()`` for in output. expect_fail: if True then succeed when the regexp doesn't match. flags: python regex flags to modify matching behavior + exact_match: if True then ``match`` must be exactly matched somewhere + in the output of ``cmd`` using ``str.find()``. Returns: (success, matches): if the match fails then "matches" will be None, @@ -436,6 +437,17 @@ class TestCase: ``matches`` otherwise group(0) (i.e., the matching text). """ out = self._command(target, cmd) + if exact_match: + if match not in out: + success = expect_fail + ret = None + else: + success = not expect_fail + ret = match + level = logging.DEBUG if success else logging.WARNING + self.olog.log(level, "exactly matched:%s:", ret) + return success, ret + search = re.search(match, out, flags) self.last_m = search if search is None: @@ -455,17 +467,19 @@ class TestCase: self, target: str, cmd: str, - match: Union[str, dict], + match: Union[str, list, dict], expect_fail: bool, - ) -> Union[str, dict]: + exact_match: bool, + ) -> (bool, Union[list, dict]): """Execute a json ``cmd`` and check result. Args: target: the target to execute the command on. cmd: string to execut on the target. - match: A json ``str`` or object (``dict``) to compare against the json - output from ``cmd``. + match: A json ``str``, object (``dict``), or array (``list``) to + compare against the json output from ``cmd``. expect_fail: if True then succeed when the json doesn't match. + exact_match: if True then the json must exactly match. """ js = self._command_json(target, cmd) try: @@ -476,7 +490,27 @@ class TestCase: "JSON load failed. Check match value is in JSON format: %s", error ) - if json_diff := json_cmp(expect, js): + if exact_match: + deep_diff = json_cmp(expect, js) + # Convert DeepDiff completely into dicts or lists at all levels + json_diff = json.loads(deep_diff.to_json()) + else: + deep_diff = json_cmp(expect, js, ignore_order=True) + # Convert DeepDiff completely into dicts or lists at all levels + json_diff = json.loads(deep_diff.to_json()) + # Remove new fields in json object from diff + if json_diff.get("dictionary_item_added") is not None: + del json_diff["dictionary_item_added"] + # Remove new json objects in json array from diff + if (new_items := json_diff.get("iterable_item_added")) is not None: + new_item_paths = list(new_items.keys()) + for path in new_item_paths: + if type(new_items[path]) is dict: + del new_items[path] + if len(new_items) == 0: + del json_diff["iterable_item_added"] + + if json_diff: success = expect_fail if not success: self.logf("JSON DIFF:%s:" % json_diff) @@ -489,14 +523,24 @@ class TestCase: self, target: str, cmd: str, - match: Union[str, dict], + match: Union[str, list, dict], is_json: bool, timeout: float, interval: float, expect_fail: bool, flags: int, - ) -> Union[str, dict]: - """Execute a command repeatedly waiting for result until timeout.""" + exact_match: bool, + ) -> Union[str, list, dict]: + """Execute a command repeatedly waiting for result until timeout. + + ``match`` is a regular expression to search for in the output of ``cmd`` + when ``is_json`` is False. + + When ``is_json`` is True ``match`` must be a json object, a json array, + or a ``str`` which parses into a json object. Likewise, the ``cmd`` output + is parsed into a json object or array and then a comparison is done between + the two json objects or arrays. + """ startt = time.time() endt = startt + timeout @@ -504,10 +548,12 @@ class TestCase: ret = None while not success and time.time() < endt: if is_json: - success, ret = self._match_command_json(target, cmd, match, expect_fail) + success, ret = self._match_command_json( + target, cmd, match, expect_fail, exact_match + ) else: success, ret = self._match_command( - target, cmd, match, expect_fail, flags + target, cmd, match, expect_fail, flags, exact_match ) if not success: time.sleep(interval) @@ -626,7 +672,7 @@ class TestCase: ) return self._command(target, cmd) - def step_json(self, target: str, cmd: str) -> dict: + def step_json(self, target: str, cmd: str) -> Union[list, dict]: """See :py:func:`~munet.mutest.userapi.step_json`. :meta private: @@ -649,13 +695,14 @@ class TestCase: desc: str = "", expect_fail: bool = False, flags: int = re.DOTALL, + exact_match: bool = False, ) -> (bool, Union[str, list]): """See :py:func:`~munet.mutest.userapi.match_step`. :meta private: """ self.logf( - "#%s.%s:%s:MATCH_STEP:%s:%s:%s:%s:%s:%s", + "#%s.%s:%s:MATCH_STEP:%s:%s:%s:%s:%s:%s:%s", self.tag, self.steps + 1, self.info.path, @@ -665,8 +712,11 @@ class TestCase: desc, expect_fail, flags, + exact_match, + ) + success, ret = self._match_command( + target, cmd, match, expect_fail, flags, exact_match ) - success, ret = self._match_command(target, cmd, match, expect_fail, flags) if desc: self.__post_result(target, success, desc) return success, ret @@ -684,16 +734,17 @@ class TestCase: self, target: str, cmd: str, - match: Union[str, dict], + match: Union[str, list, dict], desc: str = "", expect_fail: bool = False, - ) -> (bool, Union[str, dict]): + exact_match: bool = False, + ) -> (bool, Union[list, dict]): """See :py:func:`~munet.mutest.userapi.match_step_json`. :meta private: """ self.logf( - "#%s.%s:%s:MATCH_STEP_JSON:%s:%s:%s:%s:%s", + "#%s.%s:%s:MATCH_STEP_JSON:%s:%s:%s:%s:%s:%s", self.tag, self.steps + 1, self.info.path, @@ -702,8 +753,11 @@ class TestCase: match, desc, expect_fail, + exact_match, + ) + success, ret = self._match_command_json( + target, cmd, match, expect_fail, exact_match ) - success, ret = self._match_command_json(target, cmd, match, expect_fail) if desc: self.__post_result(target, success, desc) return success, ret @@ -718,6 +772,7 @@ class TestCase: interval=0.5, expect_fail: bool = False, flags: int = re.DOTALL, + exact_match: bool = False, ) -> (bool, Union[str, list]): """See :py:func:`~munet.mutest.userapi.wait_step`. @@ -726,7 +781,7 @@ class TestCase: if interval is None: interval = min(timeout / 20, 0.25) self.logf( - "#%s.%s:%s:WAIT_STEP:%s:%s:%s:%s:%s:%s:%s:%s", + "#%s.%s:%s:WAIT_STEP:%s:%s:%s:%s:%s:%s:%s:%s:%s", self.tag, self.steps + 1, self.info.path, @@ -738,9 +793,18 @@ class TestCase: desc, expect_fail, flags, + exact_match, ) success, ret = self._wait( - target, cmd, match, False, timeout, interval, expect_fail, flags + target, + cmd, + match, + False, + timeout, + interval, + expect_fail, + flags, + exact_match, ) if desc: self.__post_result(target, success, desc) @@ -750,12 +814,13 @@ class TestCase: self, target: str, cmd: str, - match: Union[str, dict], + match: Union[str, list, dict], desc: str = "", timeout=10, interval=None, expect_fail: bool = False, - ) -> (bool, Union[str, dict]): + exact_match: bool = False, + ) -> (bool, Union[list, dict]): """See :py:func:`~munet.mutest.userapi.wait_step_json`. :meta private: @@ -763,7 +828,7 @@ class TestCase: if interval is None: interval = min(timeout / 20, 0.25) self.logf( - "#%s.%s:%s:WAIT_STEP:%s:%s:%s:%s:%s:%s:%s", + "#%s.%s:%s:WAIT_STEP:%s:%s:%s:%s:%s:%s:%s:%s", self.tag, self.steps + 1, self.info.path, @@ -774,9 +839,10 @@ class TestCase: interval, desc, expect_fail, + exact_match, ) success, ret = self._wait( - target, cmd, match, True, timeout, interval, expect_fail, 0 + target, cmd, match, True, timeout, interval, expect_fail, 0, exact_match ) if desc: self.__post_result(target, success, desc) @@ -864,15 +930,15 @@ def step(target: str, cmd: str) -> str: return TestCase.g_tc.step(target, cmd) -def step_json(target: str, cmd: str) -> dict: - """Execute a json ``cmd`` on a ``target`` and return the json object. +def step_json(target: str, cmd: str) -> Union[list, dict]: + """Execute a json ``cmd`` on a ``target`` and return the json object or array. Args: target: the target to execute the ``cmd`` on. cmd: string to execute on the target. Returns: - Returns the json object after parsing the ``cmd`` output. + Returns the json object or array after parsing the ``cmd`` output. If json parse fails, a warning is logged and an empty ``dict`` is used. """ @@ -904,6 +970,7 @@ def match_step( desc: str = "", expect_fail: bool = False, flags: int = re.DOTALL, + exact_match: bool = False, ) -> (bool, Union[str, list]): """Execute a ``cmd`` on a ``target`` check result. @@ -922,44 +989,53 @@ def match_step( desc: description of test, if no description then no result is logged. expect_fail: if True then succeed when the regexp doesn't match. flags: python regex flags to modify matching behavior + exact_match: if True then ``match`` must be exactly matched somewhere + in the output of ``cmd`` using ``str.find()``. Returns: Returns a 2-tuple. The first value is a bool indicating ``success``. The second value will be a list from ``re.Match.groups()`` if non-empty, otherwise ``re.Match.group(0)`` if there was a match otherwise None. """ - return TestCase.g_tc.match_step(target, cmd, match, desc, expect_fail, flags) + return TestCase.g_tc.match_step( + target, cmd, match, desc, expect_fail, flags, exact_match + ) def match_step_json( target: str, cmd: str, - match: Union[str, dict], + match: Union[str, list, dict], desc: str = "", expect_fail: bool = False, -) -> (bool, Union[str, dict]): + exact_match: bool = False, +) -> (bool, Union[list, dict]): """Execute a ``cmd`` on a ``target`` check result. - Execute ``cmd`` on ``target`` and check if the json object in ``match`` + Execute ``cmd`` on ``target`` and check if the json object or array in ``match`` matches or doesn't match (according to the ``expect_fail`` value) the json output from ``cmd``. Args: target: the target to execute the ``cmd`` on. cmd: string to execut on the ``target``. - match: A json ``str`` or object (``dict``) to compare against the json - output from ``cmd``. + match: A json ``str``, object (``dict``), or array (``list``) to compare + against the json output from ``cmd``. desc: description of test, if no description then no result is logged. expect_fail: if True then succeed if the a json doesn't match. + exact_match: if True then the json must exactly match. Returns: Returns a 2-tuple. The first value is a bool indicating ``success``. The - second value is a ``str`` diff if there is a difference found in the json - compare, otherwise the value is the json object (``dict``) from the ``cmd``. + second value is a ``dict`` of the diff if there is a difference found in + the json compare, otherwise the value is the json object (``dict``) or + array (``list``) from the ``cmd``. If json parse fails, a warning is logged and an empty ``dict`` is used. """ - return TestCase.g_tc.match_step_json(target, cmd, match, desc, expect_fail) + return TestCase.g_tc.match_step_json( + target, cmd, match, desc, expect_fail, exact_match + ) def wait_step( @@ -971,6 +1047,7 @@ def wait_step( interval: float = 0.5, expect_fail: bool = False, flags: int = re.DOTALL, + exact_match: bool = False, ) -> (bool, Union[str, list]): """Execute a ``cmd`` on a ``target`` repeatedly, looking for a result. @@ -991,6 +1068,8 @@ def wait_step( desc: description of test, if no description then no result is logged. expect_fail: if True then succeed when the regexp *doesn't* match. flags: python regex flags to modify matching behavior + exact_match: if True then ``match`` must be exactly matched somewhere + in the output of ``cmd`` using ``str.find()``. Returns: Returns a 2-tuple. The first value is a bool indicating ``success``. @@ -998,37 +1077,31 @@ def wait_step( otherwise ``re.Match.group(0)`` if there was a match otherwise None. """ return TestCase.g_tc.wait_step( - target, cmd, match, desc, timeout, interval, expect_fail, flags + target, cmd, match, desc, timeout, interval, expect_fail, flags, exact_match ) def wait_step_json( target: str, cmd: str, - match: Union[str, dict], + match: Union[str, list, dict], desc: str = "", timeout=10, interval=None, expect_fail: bool = False, -) -> (bool, Union[str, dict]): + exact_match: bool = False, +) -> (bool, Union[list, dict]): """Execute a cmd repeatedly and wait for matching result. Execute ``cmd`` on ``target``, every ``interval`` seconds until the output of ``cmd`` matches or doesn't match (according to the ``expect_fail`` value) ``match``, for up to ``timeout`` seconds. - ``match`` is a regular expression to search for in the output of ``cmd`` when - ``is_json`` is False. - - When ``is_json`` is True ``match`` must be a json object or a ``str`` which - parses into a json object. Likewise, the ``cmd`` output is parsed into a json - object and then a comparison is done between the two json objects. - Args: target: the target to execute the ``cmd`` on. cmd: string to execut on the ``target``. - match: A json object or str representation of one to compare against json - output from ``cmd``. + match: A json object, json array, or str representation of json to compare + against json output from ``cmd``. desc: description of test, if no description then no result is logged. timeout: The number of seconds to repeat the ``cmd`` looking for a match (or non-match if ``expect_fail`` is True). @@ -1037,17 +1110,18 @@ def wait_step_json( average the cmd will execute 10 times. The minimum calculated interval is .25s, shorter values can be passed explicitly. expect_fail: if True then succeed if the a json doesn't match. + exact_match: if True then the json must exactly match. Returns: Returns a 2-tuple. The first value is a bool indicating ``success``. - The second value is a ``str`` diff if there is a difference found in the - json compare, otherwise the value is a json object (dict) from the ``cmd`` - output. + The second value is a ``dict`` of the diff if there is a difference + found in the json compare, otherwise the value is a json object (``dict``) + or array (``list``) from the ``cmd`` output. If json parse fails, a warning is logged and an empty ``dict`` is used. """ return TestCase.g_tc.wait_step_json( - target, cmd, match, desc, timeout, interval, expect_fail + target, cmd, match, desc, timeout, interval, expect_fail, exact_match ) |