Coverage for tests/json_body_serializer.py: 82.43%

1# pyright: reportUnknownMemberType=false, reportUnknownVariableType=false 

2import json (empty)

3import urllib.parse (empty)

4from typing import TYPE_CHECKING, Any (empty)

5 

6import yaml (empty)

7 

8if TYPE_CHECKING: (empty)

9 from yaml import Dumper, Loader 

10else: 

11 try: (empty)

12 from yaml import CDumper as Dumper, CLoader as Loader (empty)

13 except ImportError: 

14 from yaml import Dumper, Loader 

15 

16FILTERED_HEADER_PREFIXES = ['anthropic-', 'cf-', 'x-'] (empty)

17FILTERED_HEADERS = {'authorization', 'date', 'request-id', 'server', 'user-agent', 'via', 'set-cookie', 'api-key'} (empty)

18 

19 

20class LiteralDumper(Dumper): (empty)

21 """ 

22 A custom dumper that will represent multi-line strings using literal style. 

23 """ 

24 

25 

26def str_presenter(dumper: Dumper, data: str): (empty)

27 """If the string contains newlines, represent it as a literal block.""" 

28 if '\n' in data: (empty)

29 return dumper.represent_scalar('tag:yaml.org,2002:str', data, style='|') (empty)

30 return dumper.represent_scalar('tag:yaml.org,2002:str', data) (empty)

31 

32 

33# Register the custom presenter on our dumper 

34LiteralDumper.add_representer(str, str_presenter) (empty)

35 

36 

37def deserialize(cassette_string: str): (empty)

38 cassette_dict = yaml.load(cassette_string, Loader=Loader) (empty)

39 for interaction in cassette_dict['interactions']: (empty)

40 for kind, data in interaction.items(): (empty)

41 parsed_body = data.pop('parsed_body', None) (empty)

42 if parsed_body is not None: (empty)

43 dumped_body = json.dumps(parsed_body) (empty)

44 data['body'] = {'string': dumped_body} if kind == 'response' else dumped_body (empty)

45 return cassette_dict (empty)

46 

47 

48def serialize(cassette_dict: Any): (empty)

49 for interaction in cassette_dict['interactions']: (empty)

50 for _kind, data in interaction.items(): (empty)

51 headers: dict[str, list[str]] = data.get('headers', {}) (empty)

52 # make headers lowercase 

53 headers = {k.lower(): v for k, v in headers.items()} (empty)

54 # filter headers by name 

55 headers = {k: v for k, v in headers.items() if k not in FILTERED_HEADERS} (empty)

56 # filter headers by prefix 

57 headers = { (empty)

58 k: v for k, v in headers.items() if not any(k.startswith(prefix) for prefix in FILTERED_HEADER_PREFIXES) 

59 } 

60 # update headers on source object 

61 data['headers'] = headers (empty)

62 

63 content_type = headers.get('content-type', []) (empty)

64 if any(isinstance(header, str) and header.startswith('application/json') for header in content_type): 64 ↛ 76line 64 didn't jump to line 76 because the condition on line 64 was always true(empty)

65 # Parse the body as JSON 

66 body: Any = data.get('body', None) (empty)

67 assert body is not None, data (empty)

68 if isinstance(body, dict): (empty)

69 # Responses will have the body under a field called 'string' 

70 body = body.get('string') (empty)

71 if body is not None: 71 ↛ 76line 71 didn't jump to line 76 because the condition on line 71 was always true(empty)

72 data['parsed_body'] = json.loads(body) (empty)

73 if 'access_token' in data['parsed_body']: 73 ↛ 74line 73 didn't jump to line 74 because the condition on line 73 was never true(empty)

74 data['parsed_body']['access_token'] = 'scrubbed' 

75 del data['body'] (empty)

76 if content_type == ['application/x-www-form-urlencoded']: 76 ↛ 77line 76 didn't jump to line 77 because the condition on line 76 was never true(empty)

77 query_params = urllib.parse.parse_qs(data['body']) 

78 if 'client_secret' in query_params: 

79 query_params['client_secret'] = ['scrubbed'] 

80 data['body'] = urllib.parse.urlencode(query_params) 

81 

82 # Use our custom dumper 

83 return yaml.dump(cassette_dict, Dumper=LiteralDumper, allow_unicode=True, width=120) (empty)