edq.net.exchangeserver

  1import glob
  2import http.server
  3import logging
  4import os
  5import threading
  6import time
  7import typing
  8
  9import edq.net.exchange
 10import edq.util.dirent
 11import edq.util.serial
 12
 13_logger = logging.getLogger(__name__)
 14
 15SERVER_THREAD_START_WAIT_SEC: float = 0.02
 16SERVER_THREAD_REAP_WAIT_SEC: float = 0.15
 17
 18class HTTPExchangeServer():
 19    """
 20    An HTTP server meant for testing.
 21    This server is generally meant to already know about all the requests that will be made to it,
 22    and all the responses it should make in reaction to those respective requests.
 23    This allows the server to respond very quickly, and makes it ideal for testing.
 24    This makes it easy to mock external services for testing.
 25
 26    If a request is not found in the predefined requests,
 27    then missing_request() will be called.
 28    If a response is still not available (indicated by a None return from missing_request()),
 29    then an error will be raised.
 30    """
 31
 32    def __init__(self,
 33            port: typing.Union[int, None] = None,
 34            match_options: typing.Union[typing.Dict[str, typing.Any], None] = None,
 35            default_match_options: bool = True,
 36            verbose: bool = False,
 37            raise_on_404: bool = False,
 38            **kwargs: typing.Any) -> None:
 39        self.port: typing.Union[int, None] = port
 40        """
 41        The active port this server is using.
 42        If None, then a random port will be chosen when the server is started and this field will be populated.
 43        """
 44
 45        self._http_server: typing.Union[http.server.HTTPServer, None] = None
 46        """ The HTTP server listening for connections. """
 47
 48        self._thread: typing.Union[threading.Thread, None] = None
 49        """ The thread running the HTTP server. """
 50
 51        self._run_lock: threading.Lock = threading.Lock()
 52        """ A lock that the server holds while running. """
 53
 54        self.verbose: bool = verbose
 55        """ Log more information. """
 56
 57        self.raise_on_404: bool = raise_on_404
 58        """ Raise an exception when no exchange is matched (instead of a 404 error). """
 59
 60        self._exchanges: typing.Dict[str, typing.Dict[typing.Union[str, None], typing.Dict[str, typing.List[edq.net.exchange.HTTPExchange]]]] = {}
 61        """
 62        The HTTP exchanges (requests+responses) that this server knows about.
 63        Exchanges are stored in layers to help make errors for missing requests easier.
 64        Exchanges are stored as: {url_path: {anchor: {method: [exchange, ...]}, ...}, ...}.
 65        """
 66
 67        if (match_options is None):
 68            match_options = {}
 69
 70        if (default_match_options):
 71            if ('headers_to_skip' not in match_options):
 72                match_options['headers_to_skip'] = []
 73
 74            match_options['headers_to_skip'] += edq.net.settings.get_exchanges_ignore_headers()
 75
 76        self.match_options: typing.Dict[str, typing.Any] = match_options.copy()
 77        """ Options to use when matching HTTP exchanges. """
 78
 79    def get_exchanges(self) -> typing.List[edq.net.exchange.HTTPExchange]:
 80        """
 81        Get a shallow list of all the exchanges in this server.
 82        Ordering is not guaranteed.
 83        """
 84
 85        exchanges = []
 86
 87        for url_exchanges in self._exchanges.values():
 88            for anchor_exchanges in url_exchanges.values():
 89                for method_exchanges in anchor_exchanges.values():
 90                    exchanges += method_exchanges
 91
 92        return exchanges
 93
 94    def start(self) -> None:
 95        """ Start this server in a thread and set the active port. """
 96
 97        class NestedHTTPHandler(_TestHTTPHandler):
 98            """ An HTTP handler as a nested class to bind this server object to the handler. """
 99
100            _server = self
101            _verbose = self.verbose
102            _raise_on_404 = self.raise_on_404
103            _missing_request_func = self.missing_request
104
105        if (self.port is None):
106            self.port = edq.net.util.find_open_port()
107
108        self._http_server = http.server.HTTPServer(('', self.port), NestedHTTPHandler)
109
110        if (self.verbose):
111            _logger.info("Starting test server on port %d.", self.port)
112
113        # Use a barrier to ensure that the server thread has started.
114        server_startup_barrier = threading.Barrier(2)
115
116        def _run_server(server: 'HTTPExchangeServer', server_startup_barrier: threading.Barrier) -> None:
117            server_startup_barrier.wait()
118
119            if (server._http_server is None):
120                raise ValueError('Server was not initialized.')
121
122            # Run the server within the run lock context.
123            with server._run_lock:
124                server._http_server.serve_forever(poll_interval = 0.01)
125                server._http_server.server_close()
126
127            if (self.verbose):
128                _logger.info("Stopping test server.")
129
130        self._thread = threading.Thread(target = _run_server, args = (self, server_startup_barrier))
131        self._thread.start()
132
133        # Wait for the server to startup.
134        server_startup_barrier.wait()
135        time.sleep(SERVER_THREAD_START_WAIT_SEC)
136
137    def wait_for_completion(self) -> None:
138        """
139        Block until the server is not running.
140        If called while the server is not running, this will return immediately.
141        This function will handle keyboard interrupts (Ctrl-C).
142        """
143
144        try:
145            with self._run_lock:
146                pass
147        except KeyboardInterrupt:
148            self.stop()
149            self.wait_for_completion()
150
151    def start_and_wait(self) -> None:
152        """ Start the server and block until it is done. """
153
154        self.start()
155        self.wait_for_completion()
156
157    def stop(self) -> None:
158        """ Stop this server. """
159
160        self.port = None
161
162        if (self._http_server is not None):
163            self._http_server.shutdown()
164            self._http_server = None
165
166        if (self._thread is not None):
167            if (self._thread.is_alive()):
168                self._thread.join(SERVER_THREAD_REAP_WAIT_SEC)
169
170            self._thread = None
171
172    def missing_request(self, query: edq.net.exchange.HTTPExchange) -> typing.Union[edq.net.exchange.HTTPExchange, None]:
173        """
174        Provide the server (specifically, child classes) one last chance to resolve an incoming HTTP request
175        before the server raises an exception.
176        Usually exchanges are loaded from disk, but technically a server can resolve all requests with this method.
177
178        Exchanges returned from this method are not cached/saved.
179        """
180
181        return None
182
183    def modify_exchanges(self, exchanges: typing.List[edq.net.exchange.HTTPExchange]) -> typing.List[edq.net.exchange.HTTPExchange]:
184        """
185        Modify any exchanges before they are saved into this server's cache.
186        The returned exchanges will be saved in this server's cache.
187
188        This method may be called multiple times with different collections of exchanges.
189        """
190
191        return exchanges
192
193    def lookup_exchange(self,
194            query: edq.net.exchange.HTTPExchange,
195            match_options: typing.Union[typing.Dict[str, typing.Any], None] = None,
196            ) -> typing.Tuple[typing.Union[edq.net.exchange.HTTPExchange, None], typing.Union[str, None]]:
197        """
198        Lookup the query exchange to see if it exists in this server.
199        If a match exists, the matching exchange (likely a full version of the query) will be returned along with None.
200        If a match does not exist, a None will be returned along with a message indicating how the query missed
201        (e.g., the URL was matched, but the method was not).
202        """
203
204        if (match_options is None):
205            match_options = {}
206
207        hint_display = query.url_path
208        target: typing.Any = self._exchanges
209
210        if (query.url_path not in target):
211            return None, f"Could not find matching URL path for '{hint_display}'."
212
213        hint_display = query.url_path
214        if (query.url_anchor is not None):
215            hint_display = f"{query.url_path}#{query.url_anchor}"
216
217        target = target[query.url_path]
218
219        if (query.url_anchor not in target):
220            return None, f"Found URL path, but could not find matching anchor for '{hint_display}'."
221
222        hint_display = f"{hint_display} ({query.method})"
223        target = target[query.url_anchor]
224
225        if (query.method not in target):
226            return None, f"Found URL, but could not find matching method for '{hint_display}'."
227
228        params = list(sorted(query.parameters.keys()))
229        hint_display = f"{hint_display}, (param keys = {params})"
230        target = target[query.method]
231
232        full_match_options = self.match_options.copy()
233        full_match_options.update(match_options)
234
235        hints = []
236        matches = []
237
238        for (i, exchange) in enumerate(target):
239            match, hint = exchange.match(query, **full_match_options)
240            if (match):
241                matches.append(exchange)
242                continue
243
244            # Collect hints for non-matches.
245            label = exchange.source_path
246            if (label is None):
247                label = str(i)
248
249            hints.append(f"{label}: {hint}")
250
251        if (len(matches) == 1):
252            # Found exactly one match.
253            return matches[0], None
254
255        if (len(matches) > 1):
256            # Found multiple matches.
257            match_paths = list(sorted([match.source_path for match in matches]))
258            return None, f"Found multiple matching exchanges for '{hint_display}': {match_paths}."
259
260        # Found no matches.
261        return None, f"Found matching URL and method, but could not find matching headers/params for '{hint_display}' (non-matching hints: {hints})."
262
263    def load_exchange(self, exchange: edq.net.exchange.HTTPExchange) -> None:
264        """ Load an exchange into this server. """
265
266        if (exchange is None):
267            raise ValueError("Cannot load a None exchange.")
268
269        target: typing.Any = self._exchanges
270        if (exchange.url_path not in target):
271            target[exchange.url_path] = {}
272
273        target = target[exchange.url_path]
274        if (exchange.url_anchor not in target):
275            target[exchange.url_anchor] = {}
276
277        target = target[exchange.url_anchor]
278        if (exchange.method not in target):
279            target[exchange.method] = []
280
281        target = target[exchange.method]
282        target.append(exchange)
283
284    def load_exchange_file(self,
285            path: str,
286            context: typing.Union[edq.util.serial.SerializationContext, None] = None,
287            finalize_func: typing.Union[edq.net.exchange.HTTPExchangeFinalizeFunc, None] = None,
288            ) -> None:
289        """
290        Load an exchange from a file.
291        This will also handle setting the exchanges source path and resolving the exchange's paths.
292        """
293
294        exchange = edq.net.exchange.HTTPExchange.from_path(path, context = context)
295
296        if (finalize_func is not None):
297            exchange = finalize_func(exchange)
298
299        self.load_exchange(exchange)
300
301    def load_exchanges_dir(self,
302            base_dir: str,
303            extension: str = edq.net.exchange.DEFAULT_HTTP_EXCHANGE_EXTENSION,
304            context: typing.Union[edq.util.serial.SerializationContext, None] = None,
305            finalize_func: typing.Union[edq.net.exchange.HTTPExchangeFinalizeFunc, None] = None,
306            ) -> None:
307        """ Load all exchanges found (recursively) within a directory. """
308
309        paths = list(sorted(glob.glob(os.path.join(base_dir, "**", f"*{extension}"), recursive = True)))
310        for path in paths:
311            self.load_exchange_file(path, context = context, finalize_func = finalize_func)
312
313@typing.runtime_checkable
314class MissingRequestFunction(typing.Protocol):
315    """
316    A function that can be used to create an exchange when none was found.
317    This is often the last resort when a test server cannot find an exchange matching a request.
318    """
319
320    def __call__(self,
321            query: edq.net.exchange.HTTPExchange,
322            ) -> typing.Union[edq.net.exchange.HTTPExchange, None]:
323        """
324        Create an exchange for the given query or return None.
325        """
326
327class _TestHTTPHandler(http.server.BaseHTTPRequestHandler):
328    _server: typing.Union[HTTPExchangeServer, None] = None
329    """ The test server this handler is being used for. """
330
331    _verbose: bool = False
332    """ Log more interactions. """
333
334    _raise_on_404: bool = True
335    """ Raise an exception when no exchange is matched (instead of a 404 error). """
336
337    _missing_request_func: typing.Union[MissingRequestFunction, None] = None
338    """ A fallback to get an exchange before resulting in a 404. """
339
340    # Quiet logs.
341    def log_message(self, format: str, *args: typing.Any) -> None:  # pylint: disable=redefined-builtin
342        pass
343
344    def do_DELETE(self) -> None:  # pylint: disable=invalid-name
345        """ A handler for DELETE requests. """
346
347        self._do_request('DELETE')
348
349    def do_GET(self) -> None:  # pylint: disable=invalid-name
350        """ A handler for GET requests. """
351
352        self._do_request('GET')
353
354    def do_HEAD(self) -> None:  # pylint: disable=invalid-name
355        """ A handler for HEAD requests. """
356
357        self._do_request('HEAD')
358
359    def do_OPTIONS(self) -> None:  # pylint: disable=invalid-name
360        """ A handler for OPTIONS requests. """
361
362        self._do_request('OPTIONS')
363
364    def do_PATCH(self) -> None:  # pylint: disable=invalid-name
365        """ A handler for PATCH requests. """
366
367        self._do_request('PATCH')
368
369    def do_POST(self) -> None:  # pylint: disable=invalid-name
370        """ A handler for POST requests. """
371
372        self._do_request('POST')
373
374    def do_PUT(self) -> None:  # pylint: disable=invalid-name
375        """ A handler for PUT requests. """
376
377        self._do_request('PUT')
378
379    def _do_request(self, method: str) -> None:
380        """ A common handler for multiple types of requests. """
381
382        if (self._server is None):
383            raise ValueError("Server has not been initialized.")
384
385        if (self._verbose):
386            _logger.debug("Incoming %s request: '%s'.", method, self.path)
387
388        # Parse data from the request url and body.
389        request_data, request_files = edq.net.util.parse_request_data(self.path, self.headers, self.rfile)
390
391        # Construct file info objects from the raw files.
392        files = [edq.net.exchange.FileInfo(name = name, content = content) for (name, content) in request_files.items()]
393
394        exchange, hint = self._get_exchange(method, parameters = request_data, files = files)  # type: ignore[arg-type]
395
396        if (exchange is None):
397            code = http.HTTPStatus.NOT_FOUND.value
398            headers = {}
399            payload = hint
400        else:
401            code = exchange.response_code
402            headers = exchange.response_headers
403            payload = exchange.response_body
404
405        if (payload is None):
406            payload = ''
407
408        self.send_response(code)
409        for (key, value) in headers.items():
410            self.send_header(key, value)
411        self.end_headers()
412
413        self.wfile.write(payload.encode(edq.util.dirent.DEFAULT_ENCODING))
414
415    def _get_exchange(self, method: str,
416            parameters: typing.Union[typing.Dict[str, typing.Any], None] = None,
417            files: typing.Union[typing.List[typing.Union[edq.net.exchange.FileInfo, typing.Dict[str, str]]], None] = None,
418            ) -> typing.Tuple[typing.Union[edq.net.exchange.HTTPExchange, None], typing.Union[str, None]]:
419        """ Get the matching exchange or raise an error. """
420
421        if (self._server is None):
422            raise ValueError("Server has not been initialized.")
423
424        query = edq.net.exchange.HTTPExchange(method = method,
425                url = self.path,
426                url_anchor = self.headers.get(edq.net.settings.ANCHOR_HEADER_KEY, None),
427                headers = self.headers,  # type: ignore[arg-type]
428                parameters = parameters, files = files)
429
430        exchange, hint = self._server.lookup_exchange(query)
431
432        if ((exchange is None) and (self._missing_request_func is not None)):
433            exchange = self._missing_request_func(query)  # pylint: disable=not-callable
434
435        if ((exchange is None) and self._raise_on_404):
436            raise ValueError(f"Failed to lookup exchange: '{hint}'.")
437
438        return exchange, hint