File: __init__.py

package info (click to toggle)
pytest-httpx 0.35.0-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 424 kB
  • sloc: python: 4,386; makefile: 7; sh: 5
file content (76 lines) | stat: -rw-r--r-- 2,277 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
from collections.abc import Generator
from operator import methodcaller

import httpx
import pytest
from pytest import Config, FixtureRequest, MonkeyPatch

from pytest_httpx._httpx_mock import HTTPXMock
from pytest_httpx._httpx_internals import IteratorStream
from pytest_httpx._options import _HTTPXMockOptions
from pytest_httpx.version import __version__

__all__ = (
    "HTTPXMock",
    "IteratorStream",
    "__version__",
)


@pytest.fixture
def httpx_mock(
    monkeypatch: MonkeyPatch,
    request: FixtureRequest,
) -> Generator[HTTPXMock, None, None]:
    options = {}
    for marker in request.node.iter_markers("httpx_mock"):
        options = marker.kwargs | options
    __tracebackhide__ = methodcaller("errisinstance", TypeError)
    options = _HTTPXMockOptions(**options)

    mock = HTTPXMock(options)

    # Mock synchronous requests
    real_handle_request = httpx.HTTPTransport.handle_request

    def mocked_handle_request(
        transport: httpx.HTTPTransport, request: httpx.Request
    ) -> httpx.Response:
        if options.should_mock(request):
            return mock._handle_request(transport, request)
        return real_handle_request(transport, request)

    monkeypatch.setattr(
        httpx.HTTPTransport,
        "handle_request",
        mocked_handle_request,
    )

    # Mock asynchronous requests
    real_handle_async_request = httpx.AsyncHTTPTransport.handle_async_request

    async def mocked_handle_async_request(
        transport: httpx.AsyncHTTPTransport, request: httpx.Request
    ) -> httpx.Response:
        if options.should_mock(request):
            return await mock._handle_async_request(transport, request)
        return await real_handle_async_request(transport, request)

    monkeypatch.setattr(
        httpx.AsyncHTTPTransport,
        "handle_async_request",
        mocked_handle_async_request,
    )

    yield mock
    try:
        mock._assert_options()
    finally:
        mock.reset()


def pytest_configure(config: Config) -> None:
    config.addinivalue_line(
        "markers",
        "httpx_mock(*, assert_all_responses_were_requested=True, assert_all_requests_were_expected=True, can_send_already_matched_responses=False, should_mock=lambda request: True): Configure httpx_mock fixture.",
    )