edq.net.exchange

  1import copy
  2import http
  3import os
  4import pathlib
  5import typing
  6import urllib.parse
  7
  8import requests
  9
 10import edq.net.settings
 11import edq.net.util
 12import edq.util.dirent
 13import edq.util.encoding
 14import edq.util.hash
 15import edq.util.json
 16import edq.util.parse
 17import edq.util.pyimport
 18import edq.util.serial
 19
 20DEFAULT_HTTP_EXCHANGE_EXTENSION: str= '.httpex.json'
 21
 22QUERY_CLIP_LENGTH: int = 100
 23""" If the filename of an HTTPExhange being saved is longer than this, then clip it. """
 24
 25ALLOWED_METHODS: typing.List[str] = [
 26    'DELETE',
 27    'GET',
 28    'HEAD',
 29    'OPTIONS',
 30    'PATCH',
 31    'POST',
 32    'PUT',
 33]
 34""" Allowed HTTP methods for an HTTPExchange. """
 35
 36class FileInfo(edq.util.serial.DictConverter):
 37    """ Store info about files used in HTTP exchanges. """
 38
 39    def __init__(self,
 40            path: typing.Union[str, None] = None,
 41            name: typing.Union[str, None] = None,
 42            content: typing.Union[str, bytes, None] = None,
 43            b64_encoded: bool = False,
 44            **kwargs: typing.Any) -> None:
 45        # Normalize the path from POSIX-style to the system's style.
 46        if (path is not None):
 47            path = str(pathlib.PurePath(pathlib.PurePosixPath(path)))
 48
 49        self.path: typing.Union[str, None] = path
 50        """ The on-disk path to a file. """
 51
 52        if ((name is None) and (self.path is not None)):
 53            name = os.path.basename(self.path)
 54
 55        if (name is None):
 56            raise ValueError("No name was provided for file.")
 57
 58        self.name: str = name
 59        """ The name for this file used in an HTTP request. """
 60
 61        self.content: typing.Union[str, bytes, None] = content
 62        """ The contents of this file. """
 63
 64        self.b64_encoded: bool = b64_encoded
 65        """ Whether the content is a string encoded in Base64. """
 66
 67        if ((self.path is None) and (self.content is None)):
 68            raise ValueError("File must have either path or content specified.")
 69
 70    def resolve_path(self, base_dir: str, load_file: bool = True) -> None:
 71        """ Resolve this path relative to the given base dir. """
 72
 73        if ((self.path is not None) and (not os.path.isabs(self.path))):
 74            self.path = os.path.abspath(os.path.join(base_dir, self.path))
 75
 76        if ((self.path is not None) and (self.content is None) and load_file):
 77            self.content = edq.util.dirent.read_file_bytes(self.path)
 78
 79    def hash_content(self) -> str:
 80        """
 81        Compute a hash for the content present.
 82        If no content is provided, use the path.
 83        """
 84
 85        hash_content = self.content
 86
 87        if (self.b64_encoded and isinstance(hash_content, str)):
 88            hash_content = edq.util.encoding.from_base64(hash_content)
 89
 90        if (hash_content is None):
 91            hash_content = self.path
 92
 93        return edq.util.hash.sha256_hex(hash_content)
 94
 95    def to_dict(self,
 96            context: typing.Union[edq.util.serial.SerializationContext, None] = None,
 97            ) -> typing.Dict[str, typing.Any]:
 98        data = vars(self).copy()
 99
100        # JSON does not support raw bytes, so we will need to base64 encode any binary content.
101        if (isinstance(self.content, bytes)):
102            data['content'] = edq.util.encoding.to_base64(self.content)
103            data['b64_encoded'] = True
104
105        return data
106
107    @classmethod
108    def from_dict(cls,
109            data: typing.Dict[str, typing.Any],
110            context: typing.Union[edq.util.serial.SerializationContext, None] = None,
111            ) -> typing.Any:
112        return FileInfo(**data)
113
114class HTTPExchange(edq.util.serial.DictConverter):
115    """
116    The request and response making up a full HTTP exchange.
117    """
118
119    def __init__(self,
120            method: str = 'GET',
121            url: typing.Union[str, None] = None,
122            url_path: typing.Union[str, None] = None,
123            url_anchor: typing.Union[str, None] = None,
124            parameters: typing.Union[typing.Dict[str, typing.Any], None] = None,
125            files: typing.Union[typing.List[typing.Union[FileInfo, typing.Dict[str, typing.Any]]], None] = None,
126            headers: typing.Union[typing.Dict[str, typing.Any], None] = None,
127            allow_redirects: typing.Union[bool, None] = None,
128            response_code: int = http.HTTPStatus.OK,
129            response_headers: typing.Union[typing.Dict[str, typing.Any], None] = None,
130            json_body: typing.Union[bool, None] = None,
131            response_body: typing.Union[str, dict, list, None] = None,
132            source_path: typing.Union[str, None] = None,
133            response_modifier: typing.Union[str, None] = None,
134            finalize: typing.Union[str, None] = None,
135            extra_options: typing.Union[typing.Dict[str, typing.Any], None] = None,
136            **kwargs: typing.Any) -> None:
137        method = str(method).upper()
138        if (method not in ALLOWED_METHODS):
139            raise ValueError(f"Got unknown/disallowed method: '{method}'.")
140
141        self.method: str = method
142        """ The HTTP method for this exchange. """
143
144        url_path, url_anchor, parameters = self._parse_url_components(url, url_path, url_anchor, parameters)
145
146        self.url_path: str = url_path
147        """
148        The path portion of the request URL.
149        Only the path (not domain, port, params, anchor, etc) should be included.
150        """
151
152        self.url_anchor: typing.Union[str, None] = url_anchor
153        """
154        The anchor portion of the request URL (if it exists).
155        """
156
157        self.parameters: typing.Dict[str, typing.Any] = parameters
158        """
159        The parameters/arguments for this request.
160        Parameters should be provided here and not encoded into URLs,
161        regardless of the request method.
162        With the exception of files, all parameters should be placed here.
163        """
164
165        if (files is None):
166            files = []
167
168        parsed_files = []
169        for file in files:
170            if (isinstance(file, FileInfo)):
171                parsed_files.append(file)
172            else:
173                parsed_files.append(FileInfo(**file))
174
175        self.files: typing.List[FileInfo] = parsed_files
176        """
177        A list of files to include in the request.
178        The files are represented as dicts with a
179        "path" (path to the file on disk) and "name" (the filename to send in the request) field.
180        These paths must be POSIX-style paths,
181        they will be converted to system-specific paths.
182        Once this exchange is ready for use, these paths should be resolved (and probably absolute).
183        However, when serialized these paths should probably be relative.
184        To reconcile this, resolve_paths() should be called before using this exchange.
185        """
186
187        if (headers is None):
188            headers = {}
189
190        self.headers: typing.Dict[str, typing.Any] = {key.lower().strip(): value for (key, value) in headers.items()}
191        """
192        Headers in the request.
193        All header keys are stored as lower case.
194        """
195
196        if (allow_redirects is None):
197            allow_redirects = True
198
199        self.allow_redirects: bool = allow_redirects
200        """ Follow redirects. """
201
202        self.response_code: int = response_code
203        """ The HTTP status code of the response. """
204
205        if (response_headers is None):
206            response_headers = {}
207
208        self.response_headers: typing.Dict[str, typing.Any] = {key.lower().strip(): value for (key, value) in response_headers.items()}
209        """
210        Headers in the response.
211        All header keys are stored as lower case.
212        """
213
214        if (json_body is None):
215            json_body = isinstance(response_body, (dict, list))
216
217        self.json_body: bool = json_body
218        """
219        Indicates that the response is JSON and should be converted to/from a string.
220        If the response body is passed in a dict/list and this is passed as None,
221        then this will be set as true.
222        """
223
224        if (self.json_body and isinstance(response_body, (dict, list))):
225            response_body = edq.util.json.dumps(response_body)
226
227        self.response_body: typing.Union[str, None] = response_body  # type: ignore[assignment]
228        """
229        The response that should be sent in this exchange.
230        """
231
232        self.response_modifier: typing.Union[str, None] = response_modifier
233        """
234        This function reference will be used to modify responses (in HTTPExchange.make_request() and HTTPExchange.from_response())
235        before sent back to the caller.
236        This reference must be importable via edq.util.pyimport.fetch().
237        """
238
239        self.finalize: typing.Union[str, None] = finalize
240        """
241        This function reference will be used to finalize echanges when created using from_response() before sent back to the caller.
242        This reference must be importable via edq.util.pyimport.fetch().
243        """
244
245        self.source_path: typing.Union[str, None] = source_path
246        """
247        The path that this exchange was loaded from (if it was loaded from a file).
248        This value should never be seriald, but can be useful for testing.
249        """
250
251        if (extra_options is None):
252            extra_options = {}
253
254        self.extra_options: typing.Dict[str, typing.Any] = extra_options.copy()
255        """
256        Additional options for this exchange.
257        This library will not use these options, but other's may.
258        kwargs will also be added to this.
259        """
260
261        self.extra_options.update(kwargs)
262
263    def _parse_url_components(self,
264            url: typing.Union[str, None] = None,
265            url_path: typing.Union[str, None] = None,
266            url_anchor: typing.Union[str, None] = None,
267            parameters: typing.Union[typing.Dict[str, typing.Any], None] = None,
268            ) -> typing.Tuple[str, typing.Union[str, None], typing.Dict[str, typing.Any]]:
269        """
270        Parse out all URL-based components from raw inputs.
271        The URL's path and anchor can either be supplied separately, or as part of the full given URL.
272        If content is present in both places, they much match (or an error will be raised).
273        Query parameters may be provided in the full URL,
274        but will be overwritten by any that are provided separately.
275        Any information from the URL aside from the path, anchor/fragment, and query will be ignored.
276        Note that path parameters (not query parameters) will be ignored.
277        The final url path, url anchor, and parameters will be returned.
278        """
279
280        # Do base initialization and cleanup.
281
282        if (url_path is not None):
283            url_path = url_path.strip()
284            if (url_path == ''):
285                url_path = ''
286            else:
287                url_path = url_path.lstrip('/')
288
289        if (url_anchor is not None):
290            url_anchor = url_anchor.strip()
291            if (url_anchor == ''):
292                url_anchor = None
293            else:
294                url_anchor = url_anchor.lstrip('#')
295
296        if (parameters is None):
297            parameters = {}
298
299        # Parse the URL (if present).
300
301        if ((url is not None) and (url.strip() != '')):
302            parts = urllib.parse.urlparse(url)
303
304            # Handle the path.
305
306            path = parts.path.lstrip('/')
307
308            if ((url_path is not None) and (url_path != path)):
309                raise ValueError(f"Mismatched URL paths where supplied implicitly ('{path}') and explicitly ('{url_path}').")
310
311            url_path = path
312
313            # Check the optional anchor/fragment.
314
315            if (parts.fragment != ''):
316                fragment = parts.fragment.lstrip('#')
317
318                if ((url_anchor is not None) and (url_anchor != fragment)):
319                    raise ValueError(f"Mismatched URL anchors where supplied implicitly ('{fragment}') and explicitly ('{url_anchor}').")
320
321                url_anchor = fragment
322
323            # Check for any parameters.
324
325            url_params = edq.net.util.parse_query_string(parts.query)
326            for (key, value) in url_params.items():
327                if (key not in parameters):
328                    parameters[key] = value
329
330        if (url_path is None):
331            raise ValueError('URL path cannot be empty, it must be explicitly set via `url_path`, or indirectly via `url`.')
332
333        # Sort parameter keys for consistency.
334        parameters = {key: parameters[key] for key in sorted(parameters.keys())}
335
336        return url_path, url_anchor, parameters
337
338    def resolve_paths(self, base_dir: str) -> None:
339        """ Resolve any paths relative to the given base dir. """
340
341        for file_info in self.files:
342            file_info.resolve_path(base_dir)
343
344    def match(self, query: 'HTTPExchange',
345            match_headers: bool = True,
346            headers_to_skip: typing.Union[typing.List[str], None] = None,
347            params_to_skip: typing.Union[typing.List[str], None] = None,
348            **kwargs: typing.Any) -> typing.Tuple[bool, typing.Union[str, None]]:
349        """
350        Check if this exchange matches the query exchange.
351        If they match, `(True, None)` will be returned.
352        If they do not match, `(False, <hint>)` will be returned, where `<hint>` points to where the mismatch is.
353
354        Note that this is not an equality check,
355        as a query exchange is often missing the response components.
356        This method is often invoked the see if an incoming HTTP request (the query) matches an existing exchange.
357        """
358
359        if (query.method != self.method):
360            return False, f"HTTP method does not match (query = {query.method}, target = {self.method})."
361
362        if (query.url_path != self.url_path):
363            return False, f"URL path does not match (query = {query.url_path}, target = {self.url_path})."
364
365        if (query.url_anchor != self.url_anchor):
366            return False, f"URL anchor does not match (query = {query.url_anchor}, target = {self.url_anchor})."
367
368        if (headers_to_skip is None):
369            headers_to_skip = edq.net.settings.get_exchanges_ignore_headers()
370
371        if (params_to_skip is None):
372            params_to_skip = []
373
374        if (match_headers):
375            match, hint = self._match_dict('header', query.headers, self.headers, headers_to_skip)
376            if (not match):
377                return False, hint
378
379        match, hint = self._match_dict('parameter', query.parameters, self.parameters, params_to_skip)
380        if (not match):
381            return False, hint
382
383        # Check file names and hash contents.
384        query_filenames = {(file.name, file.hash_content()) for file in query.files}
385        target_filenames = {(file.name, file.hash_content()) for file in self.files}
386        if (query_filenames != target_filenames):
387            return False, f"File names do not match (query = {query_filenames}, target = {target_filenames})."
388
389        return True, None
390
391    def _match_dict(self,
392            label: str,
393            query_dict: typing.Dict[str, typing.Any],
394            target_dict: typing.Dict[str, typing.Any],
395            keys_to_skip: typing.Union[typing.List[str], None] = None,
396            query_label: str = 'query',
397            target_label: str = 'target',
398            normalize_key_case: bool = True,
399            ) -> typing.Tuple[bool, typing.Union[str, None]]:
400        """ A subcheck in match(), specifically for a dictionary. """
401
402        if (keys_to_skip is None):
403            keys_to_skip = []
404
405        if (normalize_key_case):
406            keys_to_skip = [key.lower() for key in keys_to_skip]
407            query_dict = {key.lower(): value for (key, value) in query_dict.items()}
408            target_dict = {key.lower(): value for (key, value) in target_dict.items()}
409
410        query_keys = set(query_dict.keys()) - set(keys_to_skip)
411        target_keys = set(target_dict.keys()) - set(keys_to_skip)
412
413        if (query_keys != target_keys):
414            return False, f"{label.title()} keys do not match ({query_label} = {query_keys}, {target_label} = {target_keys})."
415
416        for key in sorted(query_keys):
417            query_value = query_dict[key]
418            target_value = target_dict[key]
419
420            if (query_value != target_value):
421                comparison = f"{query_label} = '{query_value}', {target_label} = '{target_value}'"
422                return False, f"{label.title()} '{key}' has a non-matching value ({comparison})."
423
424        return True, None
425
426    def get_url(self) -> str:
427        """ Get the URL path and anchor combined. """
428
429        url = self.url_path
430
431        if (self.url_anchor is not None):
432            url += ('#' + self.url_anchor)
433
434        return url
435
436    def match_response(self,
437            response: requests.Response,
438            override_body: typing.Union[str, None] = None,
439            headers_to_skip: typing.Union[typing.List[str], None] = None,
440            **kwargs: typing.Any) -> typing.Tuple[bool, typing.Union[str, None]]:
441        """
442        Check if this exchange matches the given response.
443        If they match, `(True, None)` will be returned.
444        If they do not match, `(False, <hint>)` will be returned, where `<hint>` points to where the mismatch is.
445        """
446
447        if (headers_to_skip is None):
448            headers_to_skip = edq.net.settings.get_exchanges_ignore_headers()
449
450        response_body = override_body
451        if (response_body is None):
452            response_body = response.text
453
454        if (self.response_code != response.status_code):
455            return False, f"http status code does match (expected: {self.response_code}, actual: {response.status_code})"
456
457        expected_body = self.response_body
458        actual_body = None
459
460        if (self.json_body):
461            actual_body = response.json()
462
463            # Normalize the actual and expected bodies.
464
465            actual_body = edq.util.json.dumps(actual_body)
466
467            if (isinstance(expected_body, str)):
468                expected_body = edq.util.json.loads(expected_body)
469
470            expected_body = edq.util.json.dumps(expected_body)
471        else:
472            actual_body = response_body
473
474        if (self.response_body != actual_body):
475            body_hint = f"expected: '{self.response_body}', actual: '{actual_body}'"
476            return False, f"body does not match ({body_hint})"
477
478        response_headers = {key.lower().strip(): value for (key, value) in response.headers.items()}
479        match, hint = self._match_dict('header', response_headers, self.response_headers,
480                keys_to_skip = headers_to_skip,
481                query_label = 'response', target_label = 'exchange')
482
483        if (not match):
484            return False, hint
485
486        return True, None
487
488    def compute_relpath(self, http_exchange_extension: str = DEFAULT_HTTP_EXCHANGE_EXTENSION) -> str:
489        """ Create a consistent, semi-unique, and relative path for this exchange. """
490
491        url = self.get_url().strip()
492        parts = url.split('/')
493
494        if (url in ['', '/']):
495            filename = '_index_'
496            dirname = ''
497        else:
498            filename = parts[-1]
499
500            if (len(parts) > 1):
501                dirname = os.path.join(*parts[0:-1])
502            else:
503                dirname = ''
504
505        parameters = {}
506        for key in sorted(self.parameters.keys()):
507            parameters[key] = self.parameters[key]
508
509        # Treat files as params as well.
510        for file_info in self.files:
511            parameters[f"file-{file_info.name}"] = file_info.hash_content()
512
513        query = urllib.parse.urlencode(parameters)
514        if (query != ''):
515            # The query can get very long, so we may have to clip it.
516            query_text = edq.util.hash.clip_text(query, QUERY_CLIP_LENGTH)
517
518            # Note that the '?' is URL encoded.
519            filename += f"%3F{query_text}"
520
521        filename += f"_{self.method}{http_exchange_extension}"
522
523        return os.path.join(dirname, filename)
524
525    def to_dict(self,
526            context: typing.Union[edq.util.serial.SerializationContext, None] = None,
527            ) -> typing.Dict[str, typing.Any]:
528        return vars(self)
529
530    @classmethod
531    def from_dict(cls,
532            data: typing.Dict[str, typing.Any],
533            context: typing.Union[edq.util.serial.SerializationContext, None] = None,
534            ) -> typing.Any:
535        return HTTPExchange(**data)
536
537    @classmethod
538    def from_path(cls,
539            path: str,
540            context: typing.Union[edq.util.serial.SerializationContext, None] = None,
541            ) -> 'HTTPExchange':
542        """
543        Load an exchange from a file.
544        This will also handle setting the exchanges source path (if specified) and resolving the exchange's paths.
545        """
546
547        if (context is None):
548            context = edq.util.serial.SerializationContext()
549
550        exchange = super().from_path(path, context)
551
552        set_source_path = edq.util.parse.soft_boolean(context.extra.get('set_source_path', True))
553        if (set_source_path is True):
554            exchange.source_path = os.path.abspath(path)
555
556        exchange.resolve_paths(os.path.abspath(os.path.dirname(path)))  # pylint: disable=no-member
557
558        return exchange
559
560    @classmethod
561    def from_response(cls,
562            response: requests.Response,
563            headers_to_skip: typing.Union[typing.List[str], None] = None,
564            params_to_skip: typing.Union[typing.List[str], None] = None,
565            allow_redirects: typing.Union[bool, None] = None,
566            clean_response_func: typing.Union[str, None] = None,
567            finalize_func: typing.Union[str, None] = None,
568            ) -> 'HTTPExchange':
569        """ Create a full exchange from a response. """
570
571        if (headers_to_skip is None):
572            headers_to_skip = edq.net.settings.get_exchanges_ignore_headers()
573
574        if (params_to_skip is None):
575            params_to_skip = []
576
577        if (clean_response_func is None):
578            clean_response_func = edq.net.settings.get_exchanges_clean_response_func()
579
580        if (finalize_func is None):
581            finalize_func = edq.net.settings.get_exchanges_finalize_func()
582
583        body = response.text
584
585        # Use a clean function (if one exists).
586        if (clean_response_func is not None):
587            # Make a copy of the response to avoid cleaning functions modifying it.
588            # Note that this is not a very complete solution, since we can't rely on the deep copy getting everything right.
589            response = copy.deepcopy(response)
590
591            modify_func = edq.util.pyimport.fetch(clean_response_func)
592            body = modify_func(response, body)
593
594        request_headers = {key.lower().strip(): value for (key, value) in response.request.headers.items()}
595        response_headers = {key.lower().strip(): value for (key, value) in response.headers.items()}
596
597        request_data, request_files = edq.net.util.parse_request_data(
598                response.request.url,
599                request_headers,
600                response.request.body)  # type: ignore[arg-type]
601
602        # Clean headers.
603        for key in headers_to_skip:
604            key = key.lower()
605
606            # Keep the location header if this is an allowed redirect.
607            if (allow_redirects and (key == 'location') and (300 <= response.status_code < 400)):
608                continue
609
610            request_headers.pop(key, None)
611            response_headers.pop(key, None)
612
613        # Clean parameters.
614        for key in params_to_skip:
615            request_data.pop(key, None)
616
617        files = [FileInfo(name = name, content = content) for (name, content) in request_files.items()]
618
619        data: typing.Dict[str, typing.Any] = {
620            'method': response.request.method,
621            'url': response.request.url,
622            'url_anchor': response.request.headers.get(edq.net.settings.ANCHOR_HEADER_KEY, None),
623            'parameters': request_data,
624            'files': files,
625            'headers': request_headers,
626            'response_code': response.status_code,
627            'response_headers': response_headers,
628            'response_body': body,
629            'response_modifier': clean_response_func,
630            'allow_redirects': allow_redirects,
631        }
632
633        exchange = HTTPExchange(**data)
634
635        # Use a finalize function (if one exists).
636        if (finalize_func is not None):
637            finalize_func_object = edq.util.pyimport.fetch(finalize_func)
638
639            exchange = finalize_func_object(exchange)
640            exchange.finalize = finalize_func
641
642        return exchange
643
644@typing.runtime_checkable
645class HTTPExchangeResponseCleanFunc(typing.Protocol):
646    """
647    A function that can be used to clean a response before creating an exchange.
648    """
649
650    def __call__(self,
651            response: requests.Response,
652            body: str,
653            ) -> str:
654        """
655        Take in a response and response body.
656        Clean the response in-place and return the new body.
657        """
658
659@typing.runtime_checkable
660class HTTPExchangeFinalizeFunc(typing.Protocol):
661    """
662    A function that can be used to finalize an exchange.
663    """
664
665    def __call__(self,
666            exchange: HTTPExchange
667            ) -> HTTPExchange:
668        """
669        Take in an exchange, finalize it, and pass back the finalized exchange (which may be different than the input one).
670        """
671
672@typing.runtime_checkable
673class HTTPExchangeComplete(typing.Protocol):
674    """
675    A function that can be called after a request has been made (and exchange constructed).
676    """
677
678    def __call__(self,
679            exchange: HTTPExchange
680            ) -> str:
681        """
682        Called after an HTTP exchange has been completed.
683        """