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 = edq.net.exchangeserver.HTTPExchangeServer()
 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 tearDownClass(cls) -> None:
 70        super().tearDownClass()
 71
 72        if (cls.server_key not in cls._servers):
 73            return
 74
 75        server = cls.get_server()
 76
 77        if (cls.tear_down_server):
 78            server.stop()
 79            del cls._servers[cls.server_key]
 80            cls._complete_exchange_tests.discard(cls.server_key)
 81
 82    @classmethod
 83    def suite_cleanup(cls) -> None:
 84        """ Cleanup all test servers. """
 85
 86        for server in cls._servers.values():
 87            server.stop()
 88
 89        cls._servers.clear()
 90
 91    @classmethod
 92    def get_server(cls) -> edq.net.exchangeserver.HTTPExchangeServer:
 93        """ Get the current HTTP server or raise if there is no server. """
 94
 95        server = cls._servers.get(cls.server_key, None)
 96        if (server is None):
 97            raise ValueError("Server has not been initialized.")
 98
 99        return server
100
101    @classmethod
102    def child_class_setup(cls) -> None:
103        """ This function is the recommended time for child classes to set any configuration. """
104
105    @classmethod
106    def setup_server(cls, server: edq.net.exchangeserver.HTTPExchangeServer) -> None:
107        """ An opportunity for child classes to configure the test server before starting it. """
108
109    @classmethod
110    def post_start_server(cls, server: edq.net.exchangeserver.HTTPExchangeServer) -> None:
111        """ An opportunity for child classes to work with the server after it has been started, but before any tests. """
112
113    @classmethod
114    def get_server_url(cls) -> str:
115        """ Get the URL for this test's test server. """
116
117        if (cls.override_server_url is not None):
118            return cls.override_server_url
119
120        server = cls.get_server()
121
122        if (server.port is None):
123            raise ValueError("Test server port has not been set.")
124
125        return f"http://127.0.0.1:{server.port}"
126
127    def assert_exchange(self, request: edq.net.exchange.HTTPExchange, response: edq.net.exchange.HTTPExchange,
128            base_url: typing.Union[str, None] = None,
129            ) -> requests.Response:
130        """
131        Assert that the result of making the provided request matches the provided response.
132        The same HTTPExchange may be supplied for both the request and response.
133        By default, the server's URL will be used as the base URL.
134        The full response will be returned (if no assertion is raised).
135        """
136
137        server = self.get_server()
138
139        if (base_url is None):
140            base_url = self.get_server_url()
141
142        full_response, body = edq.net.request.make_with_exchange(request, base_url, raise_for_status = True, **server.match_options)
143
144        match, hint = response.match_response(full_response, override_body = body, **server.match_options)
145        if (not match):
146            raise AssertionError(f"Exchange does not match: '{hint}'.")
147
148        return full_response
149
150    def test_exchanges_base(self) -> None:
151        """ Test making a request with each of the loaded exchanges. """
152
153        # Check if this test has already been run for this server.
154        if (self.server_key in self._complete_exchange_tests):
155            # Don't skip the test (which will show up in the test output).
156            # Instead, just return.
157            return
158
159        if (self.skip_test_exchanges_base):
160            self.skipTest('test_exchanges_base has been manually skipped.')
161
162        self._complete_exchange_tests.add(self.server_key)
163
164        server = self.get_server()
165
166        for (i, exchange) in enumerate(server.get_exchanges()):
167            base_name = exchange.get_url()
168            if (exchange.source_path is not None):
169                base_name = os.path.splitext(os.path.basename(exchange.source_path))[0]
170
171            with self.subTest(msg = f"Case {i} ({base_name}):"):
172                self.assert_exchange(exchange, exchange)