Coverage for src/zapy/requests/models.py: 100%

1from __future__ import annotations 

2 

3import typing 

4from pathlib import Path 

5 

6from httpx import _types as httpx_types 

7from httpx._client import AsyncClient as HttpxAsyncClient 

8from httpx._client import Response as HttpxResponse 

9from httpx._client import UseClientDefault 

10from pydantic import BaseModel, Field 

11from typing_extensions import TypedDict 

12 

13from zapy.__about__ import __version__ 

14from zapy.base import Metadata, ZapyCell 

15from zapy.store import Store 

16from zapy.test import assert_test_result_dict 

17 

18Code = list[str] | str 

19 

20 

21class KeyValueItem(BaseModel): 

22 key: str 

23 value: str 

24 active: bool = True 

25 

26 

27class RequestMetadata(Metadata): 

28 cell_type: str = "zapy.ZapyRequest" 

29 v: str = __version__ 

30 

31 

32class RequestArguments(TypedDict, total=False): 

33 store: Store | None 

34 logger: typing.Callable 

35 client: HttpxAsyncClient | None 

36 

37 

38class ZapyRequest(BaseModel, ZapyCell): 

39 metadata: RequestMetadata = Field(default_factory=RequestMetadata) 

40 endpoint: str 

41 method: str 

42 params: list[KeyValueItem] = Field(default_factory=list) 

43 headers: list[KeyValueItem] = Field(default_factory=list) 

44 variables: list[KeyValueItem] = Field(default_factory=list) 

45 script: Code = "" 

46 body_type: str = "None" 

47 body: Code | list[KeyValueItem] | None = None 

48 

49 @classmethod 

50 def from_dict(cls, value: dict[str, typing.Any]) -> ZapyRequest: 

51 return cls.model_validate(value) 

52 

53 @classmethod 

54 def from_path(cls, file_path: str | Path) -> ZapyRequest: 

55 import json 

56 

57 with open(file_path) as f: 

58 loaded_json: dict = json.load(f) 

59 return cls.from_dict(loaded_json) 

60 

61 async def send( 

62 self, 

63 *, 

64 raise_assert: bool = True, 

65 store: Store | None = None, 

66 logger: typing.Callable = print, 

67 client: HttpxAsyncClient | None = None, 

68 ) -> HttpxResponse: 

69 from .requester import send_request 

70 

71 request_wrapper = await send_request(self, store=store, logger=logger, client=client) 

72 

73 if request_wrapper.test_result and raise_assert is True: 

74 assert_test_result_dict(request_wrapper.test_result) 

75 

76 return request_wrapper.response 

77 

78 

79# Copied from httpx 

80 

81 

82class HttpxArguments(TypedDict, total=False): 

83 method: typing.Required[str] 

84 url: typing.Required[httpx_types.URLTypes] 

85 content: httpx_types.RequestContent | None 

86 data: httpx_types.RequestData | None 

87 files: httpx_types.RequestFiles | None 

88 json: typing.Any | None 

89 params: httpx_types.QueryParamTypes 

90 headers: httpx_types.HeaderTypes 

91 cookies: httpx_types.CookieTypes 

92 auth: httpx_types.AuthTypes | UseClientDefault 

93 follow_redirects: bool | UseClientDefault 

94 timeout: httpx_types.TimeoutTypes | UseClientDefault 

95 extensions: httpx_types.RequestExtensions