edq.procedure.verify_exchanges

 1"""
 2Verify that exchanges sent to a given server have the same response.
 3"""
 4
 5import glob
 6import logging
 7import os
 8import typing
 9import unittest
10
11import edq.testing.unittest
12import edq.util.net
13
14class ExchangeVerification(edq.testing.unittest.BaseTest):
15    """ Verify that exchanges match their content. """
16
17def run(paths: typing.List[str], server: str) -> int:
18    """ Run exchange verification. """
19
20    exchange_paths = _collect_exchange_paths(paths)
21
22    _attach_tests(exchange_paths, server)
23
24    runner = unittest.TextTestRunner(verbosity = 2)
25    tests = unittest.defaultTestLoader.loadTestsFromTestCase(ExchangeVerification)
26    results = runner.run(tests)
27
28    return len(results.errors) + len(results.failures)
29
30def _attach_tests(
31        paths: typing.List[str],
32        server: str,
33        extension: str = edq.util.net.DEFAULT_HTTP_EXCHANGE_EXTENSION,
34        ) -> None:
35    """ Create tests for each path and attach them to the ExchangeVerification class. """
36
37    common_prefix = os.path.commonprefix(paths)
38
39    for path in paths:
40        name = path.replace(common_prefix, '').replace(extension, '')
41        test_name = f"test_verify_exchange__{name}"
42
43        setattr(ExchangeVerification, test_name, _get_test_method(path, server))
44
45def _get_test_method(path: str, server: str,
46        match_options: typing.Union[typing.Dict[str, typing.Any], None] = None,
47        ) -> typing.Callable:
48    """ Create a test method for the given path. """
49
50    if (match_options is None):
51        match_options = {}
52
53    def __method(self: edq.testing.unittest.BaseTest) -> None:
54        exchange = edq.util.net.HTTPExchange.from_path(path)
55        response, body = exchange.make_request(server, raise_for_status = False, **match_options)
56
57        match, hint = exchange.match_response(response, override_body = body, **match_options)
58        if (not match):
59            raise AssertionError(f"Exchange does not match: '{hint}'.")
60
61    return __method
62
63def _collect_exchange_paths(
64        paths: typing.List[str],
65        extension: str = edq.util.net.DEFAULT_HTTP_EXCHANGE_EXTENSION,
66        ) -> typing.List[str]:
67    """ Collect exchange files by matching extensions and descending dirs. """
68
69    final_paths = []
70
71    for path in paths:
72        path = os.path.abspath(path)
73
74        if (os.path.isfile(path)):
75            if (path.endswith(extension)):
76                final_paths.append(path)
77            else:
78                logging.warning("Path does not look like an exchange file: '%s'.", path)
79        else:
80            dirent_paths = glob.glob(os.path.join(path, "**", f"*{extension}"), recursive = True)
81            for dirent_path in dirent_paths:
82                final_paths.append(dirent_path)
83
84    final_paths.sort()
85    return final_paths