diff --git a/validate_response/__init__.py b/validate_response/__init__.py index 1536bd2dd52bd6cac64fea1d2e393c0ff168f015..fa57145212248d66d58e5bc4a5a9bd512c43a481 100644 --- a/validate_response/__init__.py +++ b/validate_response/__init__.py @@ -1,7 +1,9 @@ import os import csv from difflib import SequenceMatcher +from pathlib import Path from time import perf_counter as perf +from typing import Any import deepdiff import re @@ -15,7 +17,7 @@ class NoResultException(Exception): # of result json when comparing with pattern - only one per test (ignore_tags should # be a string denoting predefined situation), exclusive with regular (not predefined) # tags (in normal case ignore_tags must be a list of tag specifiers) -predefined_ignore_tags = { +predefined_ignore_tags: dict[str, re.Pattern] = { '<bridge post>' : re.compile(r"root\['post_id'\]"), '<bridge posts>' : re.compile(r"root\[\d+\]\['post_id'\]"), '<bridge discussion>' : re.compile(r"root\[.+\]\['post_id'\]"), @@ -31,12 +33,13 @@ predefined_ignore_tags = { '<database votes>' : re.compile(r"root\['votes'\]\[\d+\]\['id'\]"), '<follow blog>' : re.compile(r"root\[\d+\]\['comment'\]\['post_id'\]"), # follow_api.get_blog '<tags posts>' : re.compile(r"root\[\d+\]\['post_id'\]"), - '<tags post>' : re.compile(r"root\['post_id'\]") # tags_api.get_discussion + '<tags post>' : re.compile(r"root\['post_id'\]"), # tags_api.get_discussion + '<hafbe cache_update>' : re.compile(r"root\['votes_updated_at'\]") # witness api in haf_block_explorer } def get_overlap(s1, s2): s = SequenceMatcher(None, s1, s2) - pos_a, pos_b, size = s.find_longest_match(0, len(s1), 0, len(s2)) + pos_a, pos_b, size = s.find_longest_match(0, len(s1), 0, len(s2)) return s1[pos_a:pos_a+size] if pos_b == 0 else "" def json_pretty_string(json_obj): @@ -90,15 +93,15 @@ def get_time(test_id): def compare_response_with_pattern(response, method=None, directory=None, ignore_tags=None, error_response=False, benchmark_time_threshold=None, allow_null_response=False): """ This method will compare response with pattern file """ test_fname, _ = os.getenv('PYTEST_CURRENT_TEST').split("::") - + test_dir = os.getenv("TAVERN_DIR", "") overlap = get_overlap(test_dir, test_fname) test_fname = test_dir + "/" + test_fname.replace(overlap, "") test_fname = test_fname.replace(TEST_FILE_EXT, "") - + response_fname = test_fname + RESPONSE_FILE_EXT pattern_fname = test_fname + PATTERN_FILE_EXT - + tavern_disable_comparator = bool(os.getenv('TAVERN_DISABLE_COMPARATOR', False)) if os.path.exists(response_fname) and not tavern_disable_comparator: @@ -119,7 +122,7 @@ def compare_response_with_pattern(response, method=None, directory=None, ignore_ if ignore_tags is not None: assert isinstance(ignore_tags, list), "ignore_tags should be list of tags" - # disable comparison with pattern on demand and save + # disable comparison with pattern on demand and save if tavern_disable_comparator: if error is not None: save_json(response_fname, error) @@ -168,16 +171,16 @@ def compare_response_with_pattern(response, method=None, directory=None, ignore_ def has_valid_response(response, method=None, directory=None, error_response=False, response_fname=None, benchmark_time_threshold=None): test_fname, _ = os.getenv('PYTEST_CURRENT_TEST').split("::") - + test_dir = os.getenv("TAVERN_DIR", "") overlap = get_overlap(test_dir, test_fname) test_fname = test_dir + "/" + test_fname.replace(overlap, "") test_fname = test_fname.replace(TEST_FILE_EXT, "") - + response_fname = test_fname + RESPONSE_FILE_EXT tavern_disable_comparator = bool(os.getenv('TAVERN_DISABLE_COMPARATOR', False)) - + if os.path.exists(response_fname) and not tavern_disable_comparator: os.remove(response_fname) @@ -191,7 +194,7 @@ def has_valid_response(response, method=None, directory=None, error_response=Fal correct_response = result # disable coparison with pattern on demand - # and save + # and save if tavern_disable_comparator: test_id = response_json.get("id", None) if error is not None: @@ -207,3 +210,31 @@ def has_valid_response(response, method=None, directory=None, error_response=Fal if correct_response is None: msg = "Error detected in response: result is null, json object was expected" raise NoResultException(msg) + + +def compare_rest_response_with_pattern(response, method=None, directory=None, error_response: bool = False, ignore_tags: str | list[str] | list[re.Pattern] | None = None): + pytest_current_test = os.getenv('PYTEST_CURRENT_TEST') + assert pytest_current_test is not None, "Environment variable not set: PYTEST_CURRENT_TEST" + test_fname, _ = pytest_current_test.split("::") + + test_dir = os.getenv("TAVERN_DIR", "") + overlap = get_overlap(test_dir, test_fname) + test_fname = test_dir + "/" + test_fname.replace(overlap, "") + test_fname = test_fname.replace(TEST_FILE_EXT, "") + + response_fname = test_fname + RESPONSE_FILE_EXT + + json_response: dict[str, Any] = response.json() + save_json(response_fname, json_response) + + if error_response: + for required_key in ["code", "details", "hint", "message"]: + assert required_key in json_response, f"Response, marked as error, does not contain {required_key} key" + + if isinstance(ignore_tags, str): + ignore_tags = [predefined_ignore_tags[ignore_tags]] + + pattern = load_pattern(test_fname + PATTERN_FILE_EXT) + pattern_resp_diff = deepdiff.DeepDiff(pattern, json_response, exclude_regex_paths=ignore_tags) + if pattern_resp_diff: + raise PatternDiffException("Differences detected between response and pattern.")