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