How to use the respx.models.RequestPattern function in respx

To help you get started, we’ve selected a few respx examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github lundberg / respx / tests / test_api.py View on Github external
        ({"method": RequestPattern("GET", "http://foo.bar/", pass_through=True)}, True),
    ],
)
async def test_pass_through(client, parameters, expected):
    async with MockTransport() as respx_mock:
        request = respx_mock.add(**parameters)

        with asynctest.mock.patch(
            "asyncio.open_connection",
            side_effect=ConnectionRefusedError("test request blocked"),
        ) as open_connection:
            with pytest.raises(NetworkError):
                await client.get("https://example.org/")

        assert open_connection.called is True
        assert request.called is True
        assert request.pass_through is expected
github lundberg / respx / respx / transports.py View on Github external
content_type: Optional[str] = None,
        headers: Optional[HeaderTypes] = None,
        pass_through: bool = False,
        alias: Optional[str] = None,
    ) -> RequestPattern:
        """
        Adds a request pattern with given mocked response details.
        """
        if isinstance(method, RequestPattern):
            pattern = method

        else:
            response = ResponseTemplate(
                status_code, headers, content, content_type=content_type
            )
            pattern = RequestPattern(
                method,
                url,
                response,
                pass_through=pass_through,
                alias=alias,
                base_url=self._base_url,
            )

        self.patterns.append(pattern)
        if pattern.alias:
            self.aliases[pattern.alias] = pattern

        return pattern
github lundberg / respx / respx / transports.py View on Github external
def add(
        self,
        method: Union[str, Callable, RequestPattern],
        url: Optional[Union[str, Pattern]] = None,
        status_code: Optional[int] = None,
        content: Optional[ContentDataTypes] = None,
        content_type: Optional[str] = None,
        headers: Optional[HeaderTypes] = None,
        pass_through: bool = False,
        alias: Optional[str] = None,
    ) -> RequestPattern:
        """
        Adds a request pattern with given mocked response details.
        """
        if isinstance(method, RequestPattern):
            pattern = method

        else:
            response = ResponseTemplate(
                status_code, headers, content, content_type=content_type
            )
            pattern = RequestPattern(
                method,
                url,
                response,
                pass_through=pass_through,
                alias=alias,
                base_url=self._base_url,
            )

        self.patterns.append(pattern)
github lundberg / respx / respx / transports.py View on Github external
def match(
        self,
        method: bytes,
        url: URL,
        headers: Headers = None,
        stream: Union[SyncByteStream, AsyncByteStream] = None,
    ) -> Tuple[Optional[RequestPattern], Request, Optional[ResponseTemplate]]:
        request: Request = (method, url, headers, stream)

        matched_pattern: Optional[RequestPattern] = None
        matched_pattern_index: Optional[int] = None
        response: Optional[ResponseTemplate] = None

        for i, pattern in enumerate(self.patterns):
            match = pattern.match(request)
            if not match:
                continue

            if matched_pattern_index is not None:
                # Multiple matches found, drop and use the first one
                self.patterns.pop(matched_pattern_index)
                break

            matched_pattern = pattern
            matched_pattern_index = i