edq.testing.httpserver

  1import os
  2import typing
  3
  4import requests
  5
  6import edq.net.exchange
  7import edq.net.exchangeserver
  8import edq.net.request
  9import edq.testing.unittest
 10
 11class HTTPServerTest(edq.testing.unittest.BaseTest):
 12    """
 13    A unit test class that requires a testing HTTP server to be running.
 14    """
 15
 16    server_key: str = ''
 17    """
 18    A key to indicate which test server this test class is using.
 19    By default all test classes share the same server,
 20    but child classes can set this if they want to control who is using the same server.
 21    If `tear_down_server` is true, then the relevant server will be stopped (and removed) on a call to tearDownClass(),
 22    which happens after a test class is complete.
 23    """
 24
 25    tear_down_server: bool = True
 26    """
 27    Tear down the relevant test server in tearDownClass().
 28    If set to false then the server will never get torn down,
 29    but can be shared between child test classes.
 30    """
 31
 32    skip_test_exchanges_base: bool = False
 33    """ Skip test_exchanges_base. """
 34
 35    override_server_url: typing.Union[str, None] = None
 36    """ If set, return this URL from get_server_url(). """
 37
 38    _servers: typing.Dict[str, edq.net.exchangeserver.HTTPExchangeServer] = {}
 39    """ The active test servers. """
 40
 41    _complete_exchange_tests: typing.Set[str] = set()
 42    """
 43    Keep track of the servers (by key) that have run their test_exchanges_base.
 44    This test should only be run once per server.
 45    """
 46
 47    _child_class_setup_called: bool = False
 48    """ Keep track if the child class setup was called. """
 49
 50    @classmethod
 51    def setUpClass(cls) -> None:
 52        super().setUpClass()
 53
 54        if (not cls._child_class_setup_called):
 55            cls.child_class_setup()
 56            cls._child_class_setup_called = True
 57
 58        if (cls.server_key in cls._servers):
 59            return
 60
 61        server = cls.create_server()
 62        cls._servers[cls.server_key] = server
 63
 64        cls.setup_server(server)
 65        server.start()
 66        cls.post_start_server(server)
 67
 68    @classmethod
 69    def create_server(cls) -> edq.net.exchangeserver.HTTPExchangeServer:
 70        """ Create the actual exchange server. """
 71
 72        return edq.net.exchangeserver.HTTPExchangeServer()
 73
 74    @classmethod
 75    def tearDownClass(cls) -> None:
 76        super().tearDownClass()
 77
 78        if (cls.server_key not in cls._servers):
 79            return
 80
 81        server = cls.get_server()
 82
 83        if (cls.tear_down_server):
 84            server.stop()
 85            del cls._servers[cls.server_key]
 86            cls._complete_exchange_tests.discard(cls.server_key)
 87
 88    @classmethod
 89    def suite_cleanup(cls) -> None:
 90        """ Cleanup all test servers. """
 91
 92        for server in cls._servers.values():
 93            server.stop()
 94
 95        cls._servers.clear()
 96
 97    @classmethod
 98    def get_server(cls) -> edq.net.exchangeserver.HTTPExchangeServer:
 99        """ Get the current HTTP server or raise if there is no server. """
100
101        server = cls._servers.get(cls.server_key, None)
102        if (server is None):
103            raise ValueError("Server has not been initialized.")
104
105        return server
106
107    @classmethod
108    def child_class_setup(cls) -> None:
109        """ This function is the recommended time for child classes to set any configuration. """
110
111    @classmethod
112    def setup_server(cls, server: edq.net.exchangeserver.HTTPExchangeServer) -> None:
113        """ An opportunity for child classes to configure the test server before starting it. """
114
115    @classmethod
116    def post_start_server(cls, server: edq.net.exchangeserver.HTTPExchangeServer) -> None:
117        """ An opportunity for child classes to work with the server after it has been started, but before any tests. """
118
119    @classmethod
120    def get_server_url(cls) -> str:
121        """ Get the URL for this test's test server. """
122
123        if (cls.override_server_url is not None):
124            return cls.override_server_url
125
126        server = cls.get_server()
127
128        if (server.port is None):
129            raise ValueError("Test server port has not been set.")
130
131        return f"http://127.0.0.1:{server.port}"
132
133    def assert_exchange(self, request: edq.net.exchange.HTTPExchange, response: edq.net.exchange.HTTPExchange,
134            base_url: typing.Union[str, None] = None,
135            ) -> requests.Response:
136        """
137        Assert that the result of making the provided request matches the provided response.
138        The same HTTPExchange may be supplied for both the request and response.
139        By default, the server's URL will be used as the base URL.
140        The full response will be returned (if no assertion is raised).
141        """
142
143        server = self.get_server()
144
145        if (base_url is None):
146            base_url = self.get_server_url()
147
148        full_response, body = edq.net.request.make_with_exchange(request, base_url, raise_for_status = True, **server.match_options)
149
150        match, hint = response.match_response(full_response, override_body = body, **server.match_options)
151        if (not match):
152            raise AssertionError(f"Exchange does not match: '{hint}'.")
153
154        return full_response
155
156    def test_exchanges_base(self) -> None:
157        """ Test making a request with each of the loaded exchanges. """
158
159        # Check if this test has already been run for this server.
160        if (self.server_key in self._complete_exchange_tests):
161            # Don't skip the test (which will show up in the test output).
162            # Instead, just return.
163            return
164
165        if (self.skip_test_exchanges_base):
166            self.skipTest('test_exchanges_base has been manually skipped.')
167
168        self._complete_exchange_tests.add(self.server_key)
169
170        server = self.get_server()
171
172        for (i, exchange) in enumerate(server.get_exchanges()):
173            base_name = exchange.get_url()
174            if (exchange.source_path is not None):
175                base_name = os.path.splitext(os.path.basename(exchange.source_path))[0]
176
177            with self.subTest(msg = f"Case {i} ({base_name}):"):
178                self.assert_exchange(exchange, exchange)