Coverage for src/zapy/requests/models.py: 100%
64 statements
« prev ^ index » next coverage.py v7.4.1, created at 2024-02-10 19:35 +0000
« prev ^ index » next coverage.py v7.4.1, created at 2024-02-10 19:35 +0000
1from __future__ import annotations
3import typing
4from pathlib import Path
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
13from zapy.__about__ import __version__
14from zapy.base import Metadata, ZapyCell
15from zapy.store import Store
16from zapy.test import assert_test_result_dict
18Code = list[str] | str
21class KeyValueItem(BaseModel):
22 key: str
23 value: str
24 active: bool = True
27class RequestMetadata(Metadata):
28 cell_type: str = "zapy.ZapyRequest"
29 v: str = __version__
32class RequestArguments(TypedDict, total=False):
33 store: Store | None
34 logger: typing.Callable
35 client: HttpxAsyncClient | None
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
49 @classmethod
50 def from_dict(cls, value: dict[str, typing.Any]) -> ZapyRequest:
51 return cls.model_validate(value)
53 @classmethod
54 def from_path(cls, file_path: str | Path) -> ZapyRequest:
55 import json
57 with open(file_path) as f:
58 loaded_json: dict = json.load(f)
59 return cls.from_dict(loaded_json)
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
71 request_wrapper = await send_request(self, store=store, logger=logger, client=client)
73 if request_wrapper.test_result and raise_assert is True:
74 assert_test_result_dict(request_wrapper.test_result)
76 return request_wrapper.response
79# Copied from httpx
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