edq.testing.httpserver

  1import glob
  2import http.server
  3import logging
  4import os
  5import threading
  6import time
  7import typing
  8
  9import requests
 10
 11import edq.testing.unittest
 12import edq.util.dirent
 13import edq.util.json
 14import edq.util.net
 15
 16SERVER_THREAD_START_WAIT_SEC: float = 0.02
 17SERVER_THREAD_REAP_WAIT_SEC: float = 0.15
 18
 19class HTTPTestServer():
 20    """
 21    An HTTP server meant for testing.
 22    This server is generally meant to already know about all the requests that will be made to it,
 23    and all the responses it should make in reaction to those respective requests.
 24    This allows the server to respond very quickly, and makes it ideal for testing.
 25    This makes it easy to mock external services for testing.
 26
 27    If a request is not found in the predefined requests,
 28    then missing_request() will be called.
 29    If a response is still not available (indicated by a None return from missing_request()),
 30    then an error will be raised.
 31    """
 32
 33    def __init__(self,
 34            port: typing.Union[int, None] = None,
 35            match_options: typing.Union[typing.Dict[str, typing.Any], None] = None,
 36            default_match_options: bool = True,
 37            verbose: bool = False,
 38            raise_on_404: bool = False,
 39            **kwargs: typing.Any) -> None:
 40        self.port: typing.Union[int, None] = port
 41        """
 42        The active port this server is using.
 43        If None, then a random port will be chosen when the server is started and this field will be populated.
 44        """
 45
 46        self._http_server: typing.Union[http.server.HTTPServer, None] = None
 47        """ The HTTP server listening for connections. """
 48
 49        self._thread: typing.Union[threading.Thread, None] = None
 50        """ The thread running the HTTP server. """
 51
 52        self._run_lock: threading.Lock = threading.Lock()
 53        """ A lock that the server holds while running. """
 54
 55        self.verbose: bool = verbose
 56        """ Log more information. """
 57
 58        self.raise_on_404: bool = raise_on_404
 59        """ Raise an exception when no exchange is matched (instead of a 404 error). """
 60
 61        self._exchanges: typing.Dict[str, typing.Dict[typing.Union[str, None], typing.Dict[str, typing.List[edq.util.net.HTTPExchange]]]] = {}
 62        """
 63        The HTTP exchanges (requests+responses) that this server knows about.
 64        Exchanges are stored in layers to help make errors for missing requests easier.
 65        Exchanges are stored as: {url_path: {anchor: {method: [exchange, ...]}, ...}, ...}.
 66        """
 67
 68        if (match_options is None):
 69            match_options = {}
 70
 71        if (default_match_options):
 72            if ('headers_to_skip' not in match_options):
 73                match_options['headers_to_skip'] = []
 74
 75            match_options['headers_to_skip'] += edq.util.net.DEFAULT_EXCHANGE_IGNORE_HEADERS
 76
 77        self.match_options: typing.Dict[str, typing.Any] = match_options.copy()
 78        """ Options to use when matching HTTP exchanges. """
 79
 80    def get_exchanges(self) -> typing.List[edq.util.net.HTTPExchange]:
 81        """
 82        Get a shallow list of all the exchanges in this server.
 83        Ordering is not guaranteed.
 84        """
 85
 86        exchanges = []
 87
 88        for url_exchanges in self._exchanges.values():
 89            for anchor_exchanges in url_exchanges.values():
 90                for method_exchanges in anchor_exchanges.values():
 91                    exchanges += method_exchanges
 92
 93        return exchanges
 94
 95    def start(self) -> None:
 96        """ Start this server in a thread and return the port. """
 97
 98        class NestedHTTPHandler(_TestHTTPHandler):
 99            """ An HTTP handler as a nested class to bind this server object to the handler. """
100
101            _server = self
102            _verbose = self.verbose
103            _raise_on_404 = self.raise_on_404
104            _missing_request_func = self.missing_request
105
106        if (self.port is None):
107            self.port = edq.util.net.find_open_port()
108
109        self._http_server = http.server.HTTPServer(('', self.port), NestedHTTPHandler)
110
111        if (self.verbose):
112            logging.info("Starting test server on port %d.", self.port)
113
114        # Use a barrier to ensure that the server thread has started.
115        server_startup_barrier = threading.Barrier(2)
116
117        def _run_server(server: 'HTTPTestServer', server_startup_barrier: threading.Barrier) -> None:
118            server_startup_barrier.wait()
119
120            if (server._http_server is None):
121                raise ValueError('Server was not initialized.')
122
123            # Run the server within the run lock context.
124            with server._run_lock:
125                server._http_server.serve_forever(poll_interval = 0.01)
126                server._http_server.server_close()
127
128            if (self.verbose):
129                logging.info("Stopping test server.")
130
131        self._thread = threading.Thread(target = _run_server, args = (self, server_startup_barrier))
132        self._thread.start()
133
134        # Wait for the server to startup.
135        server_startup_barrier.wait()
136        time.sleep(SERVER_THREAD_START_WAIT_SEC)
137
138    def wait_for_completion(self) -> None:
139        """
140        Block until the server is not running.
141        If called while the server is not running, this will return immediately.
142        This function will handle keyboard interrupts (Ctrl-C).
143        """
144
145        try:
146            with self._run_lock:
147                pass
148        except KeyboardInterrupt:
149            self.stop()
150            self.wait_for_completion()
151
152    def start_and_wait(self) -> None:
153        """ Start the server and block until it is done. """
154
155        self.start()
156        self.wait_for_completion()
157
158    def stop(self) -> None:
159        """ Stop this server. """
160
161        self.port = None
162
163        if (self._http_server is not None):
164            self._http_server.shutdown()
165            self._http_server = None
166
167        if (self._thread is not None):
168            if (self._thread.is_alive()):
169                self._thread.join(SERVER_THREAD_REAP_WAIT_SEC)
170
171            self._thread = None
172
173    def missing_request(self, query: edq.util.net.HTTPExchange) -> typing.Union[edq.util.net.HTTPExchange, None]:
174        """
175        Provide the server (specifically, child classes) one last chance to resolve an incoming HTTP request
176        before the server raises an exception.
177        Usually exchanges are loaded from disk, but technically a server can resolve all requests with this method.
178
179        Exchanges returned from this method are not cached/saved.
180        """
181
182        return None
183
184    def modify_exchanges(self, exchanges: typing.List[edq.util.net.HTTPExchange]) -> typing.List[edq.util.net.HTTPExchange]:
185        """
186        Modify any exchanges before they are saved into this server's cache.
187        The returned exchanges will be saved in this server's cache.
188
189        This method may be called multiple times with different collections of exchanges.
190        """
191
192        return exchanges
193
194    def lookup_exchange(self,
195            query: edq.util.net.HTTPExchange,
196            match_options: typing.Union[typing.Dict[str, typing.Any], None] = None,
197            ) -> typing.Tuple[typing.Union[edq.util.net.HTTPExchange, None], typing.Union[str, None]]:
198        """
199        Lookup the query exchange to see if it exists in this server.
200        If a match exists, the matching exchange (likely a full version of the query) will be returned along with None.
201        If a match does not exist, a None will be returned along with a message indicating how the query missed
202        (e.g., the URL was matched, but the method was not).
203        """
204
205        if (match_options is None):
206            match_options = {}
207
208        hint_display = query.url_path
209        target: typing.Any = self._exchanges
210
211        if (query.url_path not in target):
212            return None, f"Could not find matching URL path for '{hint_display}'."
213
214        hint_display = query.url_path
215        if (query.url_anchor is not None):
216            hint_display = f"{query.url_path}#{query.url_anchor}"
217
218        target = target[query.url_path]
219
220        if (query.url_anchor not in target):
221            return None, f"Found URL path, but could not find matching anchor for '{hint_display}'."
222
223        hint_display = f"{hint_display} ({query.method})"
224        target = target[query.url_anchor]
225
226        if (query.method not in target):
227            return None, f"Found URL, but could not find matching method for '{hint_display}'."
228
229        params = list(sorted(query.parameters.keys()))
230        hint_display = f"{hint_display}, (param keys = {params})"
231        target = target[query.method]
232
233        full_match_options = self.match_options.copy()
234        full_match_options.update(match_options)
235
236        hints = []
237        matches = []
238
239        for (i, exchange) in enumerate(target):
240            match, hint = exchange.match(query, **full_match_options)
241            if (match):
242                matches.append(exchange)
243                continue
244
245            # Collect hints for non-matches.
246            label = exchange.source_path
247            if (label is None):
248                label = str(i)
249
250            hints.append(f"{label}: {hint}")
251
252        if (len(matches) == 1):
253            # Found exactly one match.
254            return matches[0], None
255
256        if (len(matches) > 1):
257            # Found multiple matches.
258            match_paths = list(sorted([match.source_path for match in matches]))
259            return None, f"Found multiple matching exchanges for '{hint_display}': {match_paths}."
260
261        # Found no matches.
262        return None, f"Found matching URL and method, but could not find matching headers/params for '{hint_display}' (non-matching hints: {hints})."
263
264    def load_exchange(self, exchange: edq.util.net.HTTPExchange) -> None:
265        """ Load an exchange into this server. """
266
267        if (exchange is None):
268            raise ValueError("Cannot load a None exchange.")
269
270        target: typing.Any = self._exchanges
271        if (exchange.url_path not in target):
272            target[exchange.url_path] = {}
273
274        target = target[exchange.url_path]
275        if (exchange.url_anchor not in target):
276            target[exchange.url_anchor] = {}
277
278        target = target[exchange.url_anchor]
279        if (exchange.method not in target):
280            target[exchange.method] = []
281
282        target = target[exchange.method]
283        target.append(exchange)
284
285    def load_exchange_file(self, path: str) -> None:
286        """
287        Load an exchange from a file.
288        This will also handle setting the exchanges source path and resolving the exchange's paths.
289        """
290
291        self.load_exchange(edq.util.net.HTTPExchange.from_path(path))
292
293    def load_exchanges_dir(self, base_dir: str, extension: str = edq.util.net.DEFAULT_HTTP_EXCHANGE_EXTENSION) -> None:
294        """ Load all exchanges found (recursively) within a directory. """
295
296        paths = list(sorted(glob.glob(os.path.join(base_dir, "**", f"*{extension}"), recursive = True)))
297        for path in paths:
298            self.load_exchange_file(path)
299
300@typing.runtime_checkable
301class MissingRequestFunction(typing.Protocol):
302    """
303    A function that can be used to create an exchange when none was found.
304    This is often the last resort when a test server cannot find an exchange matching a request.
305    """
306
307    def __call__(self,
308            query: edq.util.net.HTTPExchange,
309            ) -> typing.Union[edq.util.net.HTTPExchange, None]:
310        """
311        Create an exchange for the given query or return None.
312        """
313
314class _TestHTTPHandler(http.server.BaseHTTPRequestHandler):
315    _server: typing.Union[HTTPTestServer, None] = None
316    """ The test server this handler is being used for. """
317
318    _verbose: bool = False
319    """ Log more interactions. """
320
321    _raise_on_404: bool = True
322    """ Raise an exception when no exchange is matched (instead of a 404 error). """
323
324    _missing_request_func: typing.Union[MissingRequestFunction, None] = None
325    """ A fallback to get an exchange before resulting in a 404. """
326
327    # Quiet logs.
328    def log_message(self, format: str, *args: typing.Any) -> None:  # pylint: disable=redefined-builtin
329        pass
330
331    def do_DELETE(self) -> None:  # pylint: disable=invalid-name
332        """ A handler for DELETE requests. """
333
334        self._do_request('DELETE')
335
336    def do_GET(self) -> None:  # pylint: disable=invalid-name
337        """ A handler for GET requests. """
338
339        self._do_request('GET')
340
341    def do_HEAD(self) -> None:  # pylint: disable=invalid-name
342        """ A handler for HEAD requests. """
343
344        self._do_request('HEAD')
345
346    def do_OPTIONS(self) -> None:  # pylint: disable=invalid-name
347        """ A handler for OPTIONS requests. """
348
349        self._do_request('OPTIONS')
350
351    def do_PATCH(self) -> None:  # pylint: disable=invalid-name
352        """ A handler for PATCH requests. """
353
354        self._do_request('PATCH')
355
356    def do_POST(self) -> None:  # pylint: disable=invalid-name
357        """ A handler for POST requests. """
358
359        self._do_request('POST')
360
361    def do_PUT(self) -> None:  # pylint: disable=invalid-name
362        """ A handler for PUT requests. """
363
364        self._do_request('PUT')
365
366    def _do_request(self, method: str) -> None:
367        """ A common handler for multiple types of requests. """
368
369        if (self._server is None):
370            raise ValueError("Server has not been initialized.")
371
372        if (self._verbose):
373            logging.debug("Incoming %s request: '%s'.", method, self.path)
374
375        # Parse data from the request url and body.
376        request_data, request_files = edq.util.net.parse_request_data(self.path, self.headers, self.rfile)
377
378        # Construct file info objects from the raw files.
379        files = [edq.util.net.FileInfo(name = name, content = content) for (name, content) in request_files.items()]
380
381        exchange, hint = self._get_exchange(method, parameters = request_data, files = files)  # type: ignore[arg-type]
382
383        if (exchange is None):
384            code = http.HTTPStatus.NOT_FOUND.value
385            headers = {}
386            payload = hint
387        else:
388            code = exchange.response_code
389            headers = exchange.response_headers
390            payload = exchange.response_body
391
392        if (payload is None):
393            payload = ''
394
395        self.send_response(code)
396        for (key, value) in headers.items():
397            self.send_header(key, value)
398        self.end_headers()
399
400        self.wfile.write(payload.encode(edq.util.dirent.DEFAULT_ENCODING))
401
402    def _get_exchange(self, method: str,
403            parameters: typing.Union[typing.Dict[str, typing.Any], None] = None,
404            files: typing.Union[typing.List[typing.Union[edq.util.net.FileInfo, typing.Dict[str, str]]], None] = None,
405            ) -> typing.Tuple[typing.Union[edq.util.net.HTTPExchange, None], typing.Union[str, None]]:
406        """ Get the matching exchange or raise an error. """
407
408        if (self._server is None):
409            raise ValueError("Server has not been initialized.")
410
411        query = edq.util.net.HTTPExchange(method = method,
412                url = self.path,
413                url_anchor = self.headers.get(edq.util.net.ANCHOR_HEADER_KEY, None),
414                headers = self.headers,  # type: ignore[arg-type]
415                parameters = parameters, files = files)
416
417        exchange, hint = self._server.lookup_exchange(query)
418
419        if ((exchange is None) and (self._missing_request_func is not None)):
420            exchange = self._missing_request_func(query)  # pylint: disable=not-callable
421
422        if ((exchange is None) and self._raise_on_404):
423            raise ValueError(f"Failed to lookup exchange: '{hint}'.")
424
425        return exchange, hint
426
427class HTTPServerTest(edq.testing.unittest.BaseTest):
428    """
429    A unit test class that requires a testing HTTP server to be running.
430    """
431
432    server_key: str = ''
433    """
434    A key to indicate which test server this test class is using.
435    By default all test classes share the same server,
436    but child classes can set this if they want to control who is using the same server.
437    If `tear_down_server` is true, then the relevant server will be stopped (and removed) on a call to tearDownClass(),
438    which happens after a test class is complete.
439    """
440
441    tear_down_server: bool = True
442    """
443    Tear down the relevant test server in tearDownClass().
444    If set to false then the server will never get torn down,
445    but can be shared between child test classes.
446    """
447
448    skip_test_exchanges_base: bool = False
449    """ Skip test_exchanges_base. """
450
451    _servers: typing.Dict[str, HTTPTestServer] = {}
452    """ The active test servers. """
453
454    _complete_exchange_tests: typing.Set[str] = set()
455    """
456    Keep track of the servers (by key) that have run their test_exchanges_base.
457    This test should only be run once per server.
458    """
459
460    _child_class_setup_called: bool = False
461    """ Keep track if the child class setup was called. """
462
463    @classmethod
464    def setUpClass(cls) -> None:
465        if (not cls._child_class_setup_called):
466            cls.child_class_setup()
467            cls._child_class_setup_called = True
468
469        if (cls.server_key in cls._servers):
470            return
471
472        server = HTTPTestServer()
473        cls._servers[cls.server_key] = server
474
475        cls.setup_server(server)
476        server.start()
477        cls.post_start_server(server)
478
479    @classmethod
480    def tearDownClass(cls) -> None:
481        if (cls.server_key not in cls._servers):
482            return
483
484        server = cls.get_server()
485
486        if (cls.tear_down_server):
487            server.stop()
488            del cls._servers[cls.server_key]
489            cls._complete_exchange_tests.discard(cls.server_key)
490
491    @classmethod
492    def suite_cleanup(cls) -> None:
493        """ Cleanup all test servers. """
494
495        for server in cls._servers.values():
496            server.stop()
497
498        cls._servers.clear()
499
500    @classmethod
501    def get_server(cls) -> HTTPTestServer:
502        """ Get the current HTTP server or raise if there is no server. """
503
504        server = cls._servers.get(cls.server_key, None)
505        if (server is None):
506            raise ValueError("Server has not been initialized.")
507
508        return server
509
510    @classmethod
511    def child_class_setup(cls) -> None:
512        """ This function is the recommended time for child classes to set any configuration. """
513
514    @classmethod
515    def setup_server(cls, server: HTTPTestServer) -> None:
516        """ An opportunity for child classes to configure the test server before starting it. """
517
518    @classmethod
519    def post_start_server(cls, server: HTTPTestServer) -> None:
520        """ An opportunity for child classes to work with the server after it has been started, but before any tests. """
521
522    @classmethod
523    def get_server_url(cls) -> str:
524        """ Get the URL for this test's test server. """
525
526        server = cls.get_server()
527
528        if (server.port is None):
529            raise ValueError("Test server port has not been set.")
530
531        return f"http://127.0.0.1:{server.port}"
532
533    def assert_exchange(self, request: edq.util.net.HTTPExchange, response: edq.util.net.HTTPExchange,
534            base_url: typing.Union[str, None] = None,
535            ) -> requests.Response:
536        """
537        Assert that the result of making the provided request matches the provided response.
538        The same HTTPExchange may be supplied for both the request and response.
539        By default, the server's URL will be used as the base URL.
540        The full response will be returned (if no assertion is raised).
541        """
542
543        server = self.get_server()
544
545        if (base_url is None):
546            base_url = self.get_server_url()
547
548        full_response, body = request.make_request(base_url, raise_for_status = True, **server.match_options)
549
550        match, hint = response.match_response(full_response, override_body = body, **server.match_options)
551        if (not match):
552            raise AssertionError(f"Exchange does not match: '{hint}'.")
553
554        return full_response
555
556    def test_exchanges_base(self) -> None:
557        """ Test making a request with each of the loaded exchanges. """
558
559        # Check if this test has already been run for this server.
560        if (self.server_key in self._complete_exchange_tests):
561            # Don't skip the test (which will show up in the test output).
562            # Instead, just return.
563            return
564
565        if (self.skip_test_exchanges_base):
566            self.skipTest('test_exchanges_base has been manually skipped.')
567
568        self._complete_exchange_tests.add(self.server_key)
569
570        server = self.get_server()
571
572        for (i, exchange) in enumerate(server.get_exchanges()):
573            base_name = exchange.get_url()
574            if (exchange.source_path is not None):
575                base_name = os.path.splitext(os.path.basename(exchange.source_path))[0]
576
577            with self.subTest(msg = f"Case {i} ({base_name}):"):
578                self.assert_exchange(exchange, exchange)