12095558571

Committed 30 Nov 2024 10:11AM UTC coverage: 97.727% (-1.0%) from 98.68%

Build # 12095558571

Build Type

Pull #276

github

Committed by

web-flow

Commit Message

Merge a69d07eba into 9cfad4cdb

Pull Request Pull Request #276: refactor: rework Entry, Request and Response classes

Run Details

732 of 752 new or added lines in 30 files covered. (97.34%)

1 existing line in 1 file now uncovered.

1204 of 1232 relevant lines covered (97.73%)

6.8 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.04

/mocket/compat/mockhttp.py

from __future__ import annotations

from io import BufferedReader
from typing import Any

from mocket.http import (
    MocketHttpEntry,
    MocketHttpMethod,
    MocketHttpRequest,
    MocketHttpResponse,
)
from mocket.mocket import Mocket


class Response(MocketHttpResponse):
    def __init__(
        self,
        body: str | bytes | BufferedReader = b"",
        status: int = 200,
        headers: dict[str, str] | None = None,
    ) -> None:
        super().__init__(
            status_code=status,
            headers=headers,
            body=body,
        )

    @property
    def status(self) -> int:
        return self.status_code


class Request(MocketHttpRequest):
    @property
    def body(self) -> str | None:  # type: ignore
        body = super().body
        if body is None:
            return None
        return body.decode()


class Entry(MocketHttpEntry):
    request_cls = Request
    response_cls = Response  # type: ignore[assignment]

    CONNECT = MocketHttpMethod.CONNECT
    DELETE = MocketHttpMethod.DELETE
    GET = MocketHttpMethod.GET
    HEAD = MocketHttpMethod.HEAD
    OPTIONS = MocketHttpMethod.OPTIONS
    PATCH = MocketHttpMethod.PATCH
    POST = MocketHttpMethod.POST
    PUT = MocketHttpMethod.PUT
    TRACE = MocketHttpMethod.TRACE

    METHODS = list(MocketHttpMethod)

    def __init__(
        self,
        uri: str,
        method: MocketHttpMethod,
        responses: list[Exception | Response],
        match_querystring: bool = True,
        add_trailing_slash: bool = True,
    ) -> None:
        super().__init__(
            method=method,
            uri=uri,
            responses=responses,
            match_querystring=match_querystring,
            add_trailing_slash=add_trailing_slash,
        )

    def __repr__(self) -> str:
        return (
            f"{self.__class__.__name__}("
            f"method='{self.method.name}', "
            f"schema='{self.schema}', "
            f"location={self.address}, "
            f"path='{self.path}', "
            f"query='{self.query}'"
            ")"
        )

    @property
    def schema(self) -> str:
        return self.scheme

    @classmethod
    def register(
        cls,
        method: MocketHttpMethod,
        uri: str,
        *responses: Exception | Response,
        **config: Any,
    ) -> None:
        if "body" in config or "status" in config:
            raise AttributeError("Did you mean `Entry.single_register(...)`?")

        if isinstance(config, dict):
            match_querystring = config.get("match_querystring", True)
            add_trailing_slash = config.get("add_trailing_slash", True)

        entry = cls(
            method=method,
            uri=uri,
            responses=list(responses),
            match_querystring=match_querystring,
            add_trailing_slash=add_trailing_slash,
        )
        Mocket.register(entry)

    @classmethod
    def single_register(
        cls,
        method: MocketHttpMethod,
        uri: str,
        body: str | bytes | BufferedReader = b"",
        status: int = 200,
        headers: dict[str, str] | None = None,
        match_querystring: bool = True,
        exception: Exception | None = None,
    ) -> None:
        response: Response | Exception
        if exception is not None:
            response = exception
        else:
            response = Response(
                body=body,
                status=status,
                headers=headers,
            )

        cls.register(
            method,
            uri,
            response,
            match_querystring=match_querystring,
        )

1	from __future__ import annotations	7✔
2
3	from io import BufferedReader	7✔
4	from typing import Any	7✔
5
6	from mocket.http import (	7✔
7	MocketHttpEntry,
8	MocketHttpMethod,
9	MocketHttpRequest,
10	MocketHttpResponse,
11	)
12	from mocket.mocket import Mocket	7✔
13
14
15	class Response(MocketHttpResponse):	7✔
16	def __init__(	7✔
17	self,
18	body: str \| bytes \| BufferedReader = b"",
19	status: int = 200,
20	headers: dict[str, str] \| None = None,
21	) -> None:
22	super().__init__(	7✔
23	status_code=status,
24	headers=headers,
25	body=body,
26	)
27
28	@property	7✔
29	def status(self) -> int:	7✔
30	return self.status_code	7✔
31
32
33	class Request(MocketHttpRequest):	7✔
34	@property	7✔
35	def body(self) -> str \| None: # type: ignore	7✔
36	body = super().body	7✔
37	if body is None:	7✔
NEW 38	return None	×
39	return body.decode()	7✔
40
41
42	class Entry(MocketHttpEntry):	7✔
43	request_cls = Request	7✔
44	response_cls = Response # type: ignore[assignment]	7✔
45
46	CONNECT = MocketHttpMethod.CONNECT	7✔
47	DELETE = MocketHttpMethod.DELETE	7✔
48	GET = MocketHttpMethod.GET	7✔
49	HEAD = MocketHttpMethod.HEAD	7✔
50	OPTIONS = MocketHttpMethod.OPTIONS	7✔
51	PATCH = MocketHttpMethod.PATCH	7✔
52	POST = MocketHttpMethod.POST	7✔
53	PUT = MocketHttpMethod.PUT	7✔
54	TRACE = MocketHttpMethod.TRACE	7✔
55
56	METHODS = list(MocketHttpMethod)	7✔
57
58	def __init__(	7✔
59	self,
60	uri: str,
61	method: MocketHttpMethod,
62	responses: list[Exception \| Response],
63	match_querystring: bool = True,
64	add_trailing_slash: bool = True,
65	) -> None:
66	super().__init__(	7✔
67	method=method,
68	uri=uri,
69	responses=responses,
70	match_querystring=match_querystring,
71	add_trailing_slash=add_trailing_slash,
72	)
73
74	def __repr__(self) -> str:
75	return (
76	f"{self.__class__.__name__}("
77	f"method='{self.method.name}', "
78	f"schema='{self.schema}', "
79	f"location={self.address}, "
80	f"path='{self.path}', "
81	f"query='{self.query}'"
82	")"
83	)
84
85	@property	7✔
86	def schema(self) -> str:	7✔
87	return self.scheme	7✔
88
89	@classmethod	7✔
90	def register(	7✔
91	cls,
92	method: MocketHttpMethod,
93	uri: str,
94	*responses: Exception \| Response,
95	**config: Any,
96	) -> None:
97	if "body" in config or "status" in config:	7✔
98	raise AttributeError("Did you mean `Entry.single_register(...)`?")	7✔
99
100	if isinstance(config, dict):	7✔
101	match_querystring = config.get("match_querystring", True)	7✔
102	add_trailing_slash = config.get("add_trailing_slash", True)	7✔
103
104	entry = cls(	7✔
105	method=method,
106	uri=uri,
107	responses=list(responses),
108	match_querystring=match_querystring,
109	add_trailing_slash=add_trailing_slash,
110	)
111	Mocket.register(entry)	7✔
112
113	@classmethod	7✔
114	def single_register(	7✔
115	cls,
116	method: MocketHttpMethod,
117	uri: str,
118	body: str \| bytes \| BufferedReader = b"",
119	status: int = 200,
120	headers: dict[str, str] \| None = None,
121	match_querystring: bool = True,
122	exception: Exception \| None = None,
123	) -> None:
124	response: Response \| Exception
125	if exception is not None:	7✔
126	response = exception	7✔
127	else:
128	response = Response(	7✔
129	body=body,
130	status=status,
131	headers=headers,
132	)
133
134	cls.register(	7✔
135	method,
136	uri,
137	response,
138	match_querystring=match_querystring,
139	)

mindflayer / python-mocket / 12095558571

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous