Tests#

Provides utilities for testing purposes like mocks of aiohttp, requests, urllib3 and decorators to apply those mocks to test functions.

Decorators#

Test decorators for HTTP library mocking.

This module provides decorators to mock HTTP requests for popular Python HTTP libraries:
  • aiohttp (async HTTP client).

  • requests (synchronous HTTP library).

  • urllib3 (low-level HTTP client).

These decorators simplify test setup by automatically patching the appropriate HTTP methods and returning configurable mock responses.

Example

Basic usage with requests:

@patch_requests(json_response={"id": 123}, status_code=201)
def test_api_call(self):
    response = requests.get("https://api.example.com")
    assert response.status_code == 201
    assert response.json()["id"] == 123

See also

  • BaseAiohttpTestCases: Base class for aiohttp test utilities.

  • BaseRequestsTestCases: Base class for requests test utilities.

  • BaseUrllib3TestCases: Base class for urllib3 test utilities.

core_https.tests.decorators.patch_aiohttp(url: str = 'https://example.com', method: str = 'GET', json_response: Dict[str, Any] | None = None, status: int = 200, headers: Dict[str, str] | None = None, text_response: str | None = None, content: bytes | None = None, content_type: str = 'application/json', charset: str = 'utf-8', raise_for_status_exception: Exception | None = None)[source]#

Decorator that patches aiohttp.ClientSession._request with an AsyncMock. This decorator works with both async test functions and sync functions using asyncio.run(). It automatically creates a mock response with the specified parameters and patches the aiohttp client session.

Parameters:
  • url – The URL that the mock should respond to. Defaults to “https://example.com”.

  • method – HTTP method for the mock request. Defaults to “GET”.

  • json_response – JSON data to return in the response body. Mutually exclusive with text_response and content.

  • status – HTTP status code for the mock response. Defaults to 200.

  • headers – Dictionary of response headers.

  • text_response – Plain text data to return. Mutually exclusive with json_response and content.

  • content – Raw bytes to return as response content. Mutually exclusive with json_response and text_response.

  • content_type – MIME type of the response. Defaults to “application/json”.

  • charset – Character encoding for the response. Defaults to “utf-8”.

  • raise_for_status_exception – Exception to raise when response.raise_for_status() is called.

Returns:

Decorated function that runs with mocked aiohttp requests.

Example

Basic JSON response mocking:

@patch_aiohttp(json_response={"id": 123}, status=201)
def test_aiohttp_mock_json(self):
    async def test():
        async with aiohttp.ClientSession() as session:
            async with session.get("https://example.com") as resp:
                self.assertEqual(resp.status, 201)
                self.assertEqual(await resp.json(), {"id": 123})

    asyncio.run(test())

Error simulation:

@patch_aiohttp(status=404, raise_for_status_exception=aiohttp.ClientResponseError(...))
def test_error_handling(self):
    # Test error handling logic
    pass

Note

This decorator patches the private method _request which may change between aiohttp versions. Consider pinning aiohttp version in your requirements.

core_https.tests.decorators.patch_requests(url: str = 'https://example.com', encoding: str = 'utf-8', headers: Dict[str, str] | None = None, json_response: Dict[str, Any] | None = None, text_response: str | None = None, status_code: int = 200, content: bytes | None = None, raise_for_status_exception: Exception | None = None)[source]#

Decorator that patches requests.sessions.Session.request to return mock responses. This decorator simplifies testing code that uses the requests library by automatically patching the HTTP requests and returning configurable mock responses.

Parameters:
  • url – The URL that the mock should respond to. Defaults to “https://example.com”.

  • encoding – Character encoding for the response. Defaults to “utf-8”.

  • headers – Dictionary of response headers.

  • json_response – JSON data to return in the response body. Mutually exclusive with text_response and content.

  • text_response – Plain text data to return. Mutually exclusive with json_response and content.

  • status_code – HTTP status code for the mock response. Defaults to 200.

  • content – Raw bytes to return as response content. Mutually exclusive with json_response and text_response.

  • raise_for_status_exception – Exception to raise when response.raise_for_status() is called.

Returns:

Decorated function that runs with mocked requests.

Example

Basic JSON response:

@patch_requests(json_response={"id": 1})
def test_get_request_mock_json(self):
    response = requests.get("https://example.com")
    self.assertEqual(response.json(), {"id": 1})

Custom status code and headers:

@patch_requests(
    status_code=201,
    headers={"Location": "/api/users/123"},
    json_response={"created": True}
)
def test_post_creation(self):
    response = requests.post("https://api.example.com/users")
    self.assertEqual(response.status_code, 201)
    self.assertEqual(response.headers["Location"], "/api/users/123")

Error handling:

@patch_requests(
    status_code=400,
    raise_for_status_exception=requests.HTTPError("Bad Request")
)
def test_error_handling(self):
    response = requests.get("https://example.com")
    with self.assertRaises(requests.HTTPError):
        response.raise_for_status()
core_https.tests.decorators.patch_urllib3(url: str = 'https://example.com', method: str = 'GET', status: int = 200, data: bytes = b'{"message": "success"}', headers: Dict[str, str] | None = None, reason: str | None = None, version: int = 11, preload_content: bool = True, decode_content: bool = True, version_string: str = 'HTTP/1.1', original_response: Mock | None = None, pool: Mock | None = None, connection: Mock | None = None, msg: Mock | None = None, retries: Mock | None = None, enforce_content_length: bool = False, with_json_attr: bool = True)[source]#

Decorator that patches urllib3 requests to return mock responses. This decorator provides comprehensive mocking for urllib3, the low-level HTTP library used by requests. It offers fine-grained control over response properties including connection details and HTTP protocol specifics.

Parameters:
  • url – The URL that the mock should respond to. Defaults to “https://example.com”.

  • method – HTTP method for the mock request. Defaults to “GET”.

  • status – HTTP status code for the mock response. Defaults to 200.

  • data – Raw bytes to return as response data. Defaults to b’{“message”: “success”}’.

  • headers – Dictionary of response headers.

  • reason – HTTP status reason phrase (e.g., “OK”, “Not Found”).

  • version – HTTP version as integer (11 for HTTP/1.1, 20 for HTTP/2.0). Defaults to 11.

  • preload_content – Whether response content is preloaded. Defaults to True.

  • decode_content – Whether to decode content based on Content-Encoding. Defaults to True.

  • version_string – HTTP version string representation. Defaults to “HTTP/1.1”.

  • original_response – Mock of the original response object.

  • pool – Mock of the connection pool.

  • connection – Mock of the HTTP connection.

  • msg – Mock of the HTTP message object.

  • retries – Mock of the retry configuration.

  • enforce_content_length – Whether to enforce Content-Length header. Defaults to False.

  • with_json_attr – Whether to add a json() method to the response. Defaults to True.

Returns:

Decorated function that runs with mocked urllib3 requests.

Example

Basic usage:

@patch_urllib3(status=404, data=b'{"error": "not found"}')
def test_urllib3_mock_404(self):
    http = urllib3.PoolManager()
    response = http.request("GET", "https://example.com")
    self.assertEqual(response.status, 404)
    self.assertEqual(response.data, b'{"error": "not found"}')

Advanced configuration:

@patch_urllib3(
    status=200,
    data=b'{"users": []}',
    headers={"Content-Type": "application/json", "X-Total-Count": "0"},
    version=20,  # HTTP/2.0
    version_string="HTTP/2.0"
)
def test_http2_response(self):
    http = urllib3.PoolManager()
    response = http.request("GET", "https://api.example.com/users")
    self.assertEqual(response.status, 200)
    self.assertEqual(response.headers["X-Total-Count"], "0")

JSON response with convenience method:

@patch_urllib3(data=b'{"id": 123}', with_json_attr=True)
def test_json_response(self):
    http = urllib3.PoolManager()
    response = http.request("GET", "https://example.com")
    self.assertEqual(response.json(), {"id": 123})

Note

This decorator patches internal urllib3 methods which may change between versions. The extensive parameter list provides maximum flexibility for testing edge cases and low-level HTTP behavior.

Base classes for Test Cases#

Base test case classes for HTTP request testing.

This module provides the foundational base class for HTTP-related testing in the core_https library. It serves as the parent class for specialized test utilities that mock different HTTP libraries (requests, aiohttp, urllib3).

The base class provides common functionality and utilities that are shared across all HTTP library testing implementations.

Example

Creating a custom HTTP test class:

from core_https.tests.base import BaseHttpTestCases

class MyHttpTests(BaseHttpTestCases):
    def test_status_code_mapping(self):
        # Access HTTP status code mappings
        self.assertEqual(self.code_mapper[200], "OK")
        self.assertEqual(self.code_mapper[404], "Not Found")

See also

  • BaseRequestsTestCases: Specialized test utilities for requests library

  • BaseAiohttpTestCases: Specialized test utilities for aiohttp library

  • BaseUrllib3TestCases: Specialized test utilities for urllib3 library

class core_https.tests.base.BaseHttpTestCases(methodName='runTest')[source]#

Bases: TestCase

Base class for HTTP request test cases.

This class serves as the foundation for all HTTP library testing utilities in the core_https package. It extends unittest.TestCase and provides common functionality shared across different HTTP client testing implementations.

The class primarily provides access to HTTP status code mappings through the code_mapper attribute, which can be used in test assertions and mock response creation.

Example

Basic usage in test methods:

class MyHttpTest(BaseHttpTestCases):
    def test_successful_response(self):
        # Use the inherited code_mapper
        self.assertEqual(self.code_mapper[200], "OK")
        self.assertEqual(self.code_mapper[201], "Created")

    def test_error_responses(self):
        # Access various error status descriptions
        self.assertEqual(self.code_mapper[400], "Bad Request")
        self.assertEqual(self.code_mapper[404], "Not Found")
        self.assertEqual(self.code_mapper[500], "Internal Server Error")

Using with mock responses:

def create_mock_response(self, status_code: int):
    return {
        "status_code": status_code,
        "reason": self.code_mapper.get(status_code, "Unknown Status"),
        "content": b'{"message": "test response"}'
    }

Note

This class is designed to be extended by specialized test utilities for specific HTTP libraries. Direct instantiation is possible but typically not necessary as subclasses provide more specific functionality.

See also

  • BaseRequestsTestCases: For testing code that uses the requests library

  • BaseAiohttpTestCases: For testing code that uses the aiohttp library

  • BaseUrllib3TestCases: For testing code that uses the urllib3 library

  • HTTPStatus: The enum class providing status code definitions

code_mapper: Dict[int, str] = {100: 'Continue', 101: 'Switching Protocols', 102: 'Processing', 103: 'Early Hints', 200: 'OK', 201: 'Created', 202: 'Accepted', 203: 'Non-Authoritative Information', 204: 'No Content', 205: 'Reset Content', 206: 'Partial Content', 207: 'Multi-Status', 208: 'Already Reported', 226: 'IM Used', 300: 'Multiple Choices', 301: 'Moved Permanently', 302: 'Found', 303: 'See Other', 304: 'Not Modified', 305: 'Use Proxy', 307: 'Temporary Redirect', 308: 'Permanent Redirect', 400: 'Bad Request', 401: 'Unauthorized', 402: 'Payment Required', 403: 'Forbidden', 404: 'Not Found', 405: 'Method Not Allowed', 406: 'Not Acceptable', 407: 'Proxy Authentication Required', 408: 'Request Timeout', 409: 'Conflict', 410: 'Gone', 411: 'Length Required', 412: 'Precondition Failed', 413: 'Payload Too Large', 414: 'URI Too Long', 415: 'Unsupported Media Type', 416: 'Range Not Satisfiable', 417: 'Expectation Failed', 418: "I'm a teapot", 421: 'Misdirected Request', 422: 'Unprocessable Entity', 423: 'Locked', 424: 'Failed Dependency', 425: 'Too Early', 426: 'Upgrade Required', 428: 'Precondition Required', 429: 'Too Many Requests', 431: 'Request Header Fields Too Large', 451: 'Unavailable For Legal Reasons', 500: 'Internal Server Error', 501: 'Not Implemented', 502: 'Bad Gateway', 503: 'Service Unavailable', 504: 'Gateway Timeout', 505: 'HTTP Version Not Supported', 506: 'Variant Also Negotiates', 507: 'Insufficient Storage', 508: 'Loop Detected', 510: 'Not Extended', 511: 'Network Authentication Required'}#
_classSetupFailed = False#
_class_cleanups = []#

Test utilities for aiohttp HTTP client library.

This module provides specialized test case utilities for testing code that uses the aiohttp library. It extends the base HTTP test functionality with aiohttp-specific mock objects and utilities.

Key features: - Mock ClientResponse objects with async method support - Error simulation with ClientResponseError creation - Case-insensitive header handling using CIMultiDict - Async context manager support for response objects

Example

Basic aiohttp response mocking:

from core_https.tests.aiohttp_ import BaseAiohttpTestCases

class MyAiohttpTest(BaseAiohttpTestCases):
    def test_async_response(self):
        mock_response = self.get_aiohttp_mock(
            json_response={"id": 123},
            status=200
        )

        # Test async methods
        async def test():
            json_data = await mock_response.json()
            assert json_data["id"] == 123

        asyncio.run(test())

See also

  • BaseHttpTestCases: Base class for HTTP testing utilities

  • BaseRequestsTestCases: Test utilities for requests library

  • BaseUrllib3TestCases: Test utilities for urllib3 library

class core_https.tests.aiohttp_.BaseAiohttpTestCases(methodName='runTest')[source]#

Bases: BaseHttpTestCases

Test utilities for aiohttp HTTP client library.

This class provides specialized methods for creating mock aiohttp.ClientResponse objects and error instances. It extends BaseHttpTestCases with aiohttp-specific functionality including async method mocking and case-insensitive headers.

The class is designed to simplify testing of code that uses aiohttp for HTTP requests by providing pre-configured mock objects that behave like real aiohttp responses.

Example

Creating mock responses for testing:

class MyAiohttpTest(BaseAiohttpTestCases):
    def test_json_response(self):
        mock = self.get_aiohttp_mock(
            json_response={"users": []},
            status=200
        )

        async def test():
            data = await mock.json()
            assert data == {"users": []}

        asyncio.run(test())

Testing error conditions:

def test_error_handling(self):
    error = self.create_client_response_error(
        status=404,
        message="Resource not found"
    )

    mock = self.get_aiohttp_mock(
        status=404,
        raise_for_status_exception=error
    )

    # Test error handling in your code
    pass

Note

This class reuses some functionality from BaseRequestsTestCases for efficiency, adapting the mock objects for aiohttp-specific behavior.

classmethod get_aiohttp_mock(url: str = 'https://example.com', method: str = 'GET', json_response: Dict[str, Any] | None = None, status: int = 200, headers: Dict[str, str] | None = None, text_response: str | None = None, content: bytes | None = None, content_type: str = 'application/json', charset: str = 'utf-8', raise_for_status_exception: Exception | None = None) Mock[source]#

Create a mock aiohttp.ClientResponse object for testing.

This method creates a comprehensive mock of aiohttp’s ClientResponse that supports async operations, case-insensitive headers, and context manager usage. The mock is based on a requests mock but adapted for aiohttp-specific behavior.

Parameters:
  • url – The URL that the mock response represents. Defaults to “https://example.com”.

  • method – HTTP method for the request. Defaults to “GET”.

  • json_response – JSON data to return from the async json() method.

  • status – HTTP status code for the response. Defaults to 200.

  • headers – Dictionary of response headers (case-insensitive).

  • text_response – Plain text content to return from the async text() method.

  • content – Raw bytes to return from the async read() method.

  • content_type – MIME type for the Content-Type header. Defaults to “application/json”.

  • charset – Character encoding for the response. Defaults to “utf-8”.

  • raise_for_status_exception – Exception to raise when raise_for_status() is called.

Returns:

Mock object simulating aiohttp.ClientResponse with async method support.

Example

Basic JSON response mock:

mock_response = BaseAiohttpTestCases.get_aiohttp_mock(
    json_response={"users": ["alice", "bob"]},
    status=200
)

async def test():
    async with mock_response as resp:
        data = await resp.json()
        assert len(data["users"]) == 2

Error response simulation:

error = ClientResponseError(...)
mock_response = BaseAiohttpTestCases.get_aiohttp_mock(
    status=500,
    raise_for_status_exception=error
)

Note

The mock supports aiohttp-specific features like: - Async context manager protocol (__aenter__, __aexit__) - Case-insensitive headers using CIMultiDict - Async methods (json(), text(), read()) - Standard aiohttp response attributes (status, method, url, etc.)

static create_client_response_error(url: str = 'http://example.com', status: int = 404, message: str = 'Not Found', history: List | None = None) ClientResponseError[source]#

Create a ClientResponseError for testing error handling.

This utility method simplifies the creation of aiohttp.ClientResponseError instances for testing error conditions. It creates the necessary request_info mock and formats the error with proper history tracking.

Parameters:
  • url – The URL where the error occurred. Defaults to “http://example.com”.

  • status – HTTP status code for the error. Defaults to 404.

  • message – Error message/reason phrase. Defaults to “Not Found”.

  • history – List of previous responses in redirect chain.

Returns:

ClientResponseError instance ready for use in tests.

Example

Testing 404 error handling:

error = BaseAiohttpTestCases.create_client_response_error(
    url="https://api.example.com/users/999",
    status=404,
    message="User not found"
)

mock_response = self.get_aiohttp_mock(
    status=404,
    raise_for_status_exception=error
)

# Test your error handling code
with self.assertRaises(ClientResponseError):
    mock_response.raise_for_status()

Testing server errors:

server_error = BaseAiohttpTestCases.create_client_response_error(
    status=500,
    message="Internal Server Error"
)

Note

The created error includes a mock request_info object with the specified URL and basic request properties. The history parameter can be used to simulate redirect chains that led to the error.

_classSetupFailed = False#
_class_cleanups = []#

Test utilities for requests HTTP client library.

This module provides specialized test case utilities for testing code that uses the popular requests library. It extends the base HTTP test functionality with requests-specific mock objects and response simulation.

Key features: - Mock requests.Response objects with realistic behavior - Support for JSON, text, and binary content responses - Status code validation and error simulation - Streaming response support with iter_content and iter_lines - Automatic content encoding and type handling

Example

Basic requests response mocking:

from core_https.tests.requests_ import BaseRequestsTestCases

class MyRequestsTest(BaseRequestsTestCases):
    def test_json_response(self):
        mock_response = self.get_requests_mock(
            json_response={"users": ["alice", "bob"]},
            status_code=200
        )

        # Test synchronous methods
        self.assertEqual(mock_response.status_code, 200)
        self.assertEqual(mock_response.json()["users"], ["alice", "bob"])
        self.assertTrue(mock_response.ok)

See also

  • BaseHttpTestCases: Base class for HTTP testing utilities

  • BaseAiohttpTestCases: Test utilities for aiohttp library

  • BaseUrllib3TestCases: Test utilities for urllib3 library

class core_https.tests.requests_.BaseRequestsTestCases(methodName='runTest')[source]#

Bases: BaseHttpTestCases

Test utilities for requests HTTP client library.

This class provides specialized methods for creating mock requests.Response objects that behave like real HTTP responses. It extends BaseHttpTestCases with requests-specific functionality including JSON handling, content encoding, and status code validation.

The class is designed to simplify testing of code that uses the requests library by providing pre-configured mock objects that match requests’ response interface and behavior.

Example

Creating mock responses for different scenarios:

class MyRequestsTest(BaseRequestsTestCases):
    def test_successful_api_call(self):
        mock = self.get_requests_mock(
            json_response={"id": 123, "name": "Alice"},
            status_code=201
        )

        self.assertEqual(mock.status_code, 201)
        self.assertTrue(mock.ok)
        self.assertEqual(mock.json()["name"], "Alice")

    def test_error_handling(self):
        mock = self.get_requests_mock(
            status_code=404,
            raise_for_status_exception=requests.HTTPError("Not Found")
        )

        self.assertFalse(mock.ok)
        with self.assertRaises(requests.HTTPError):
            mock.raise_for_status()

Testing streaming responses:

def test_streaming_content(self):
    content = b"line1\nline2\nline3"
    mock = self.get_requests_mock(content=content)

    # Test streaming methods
    for chunk in mock.iter_content():
        self.assertIsInstance(chunk, bytes)

    lines = list(mock.iter_lines())
    self.assertEqual(len(lines), 3)

Note

The mock objects created by this class include all standard requests.Response attributes and methods, making them suitable for comprehensive testing of requests-based code.

classmethod get_requests_mock(url: str = 'https://example.com', encoding: str = 'utf-8', headers: Dict[str, str] | None = None, json_response: Dict[str, Any] | None = None, text_response: str | None = None, status_code: int = 200, content: bytes | None = None, raise_for_status_exception: Exception | None = None) Mock[source]#

Create a mock requests.Response object for testing.

This method creates a comprehensive mock of requests.Response with all standard attributes and methods. It automatically handles content type conversions and provides realistic response behavior for testing.

Parameters:
  • url – The URL that the mock response represents. Defaults to “https://example.com”.

  • encoding – Character encoding for text content. Defaults to “utf-8”.

  • headers – Dictionary of response headers. Auto-generated if not provided.

  • json_response – JSON data to return from the .json() method. Mutually exclusive with text_response.

  • text_response – Plain text content. Auto-generated from json_response if not provided.

  • status_code – HTTP status code for the response. Defaults to 200.

  • content – Raw bytes content. Auto-generated from text_response if not provided.

  • raise_for_status_exception – Exception to raise when .raise_for_status() is called.

Returns:

Mock object simulating requests.Response with all standard methods and attributes.

Example

Creating a JSON response mock:

mock_response = BaseRequestsTestCases.get_requests_mock(
    json_response={"id": 123, "name": "Alice"},
    status_code=201,
    headers={"Location": "/users/123"}
)

# Test standard response properties
self.assertEqual(mock_response.status_code, 201)
self.assertTrue(mock_response.ok)
self.assertEqual(mock_response.json()["name"], "Alice")

Creating an error response:

import requests
error_mock = BaseRequestsTestCases.get_requests_mock(
    status_code=404,
    text_response="Not Found",
    raise_for_status_exception=requests.HTTPError("404 Not Found")
)

# Test error handling
self.assertFalse(error_mock.ok)
with self.assertRaises(requests.HTTPError):
    error_mock.raise_for_status()

Testing streaming content:

stream_mock = BaseRequestsTestCases.get_requests_mock(
    content=b"chunk1\nchunk2\nchunk3"
)

# Test streaming methods
chunks = list(stream_mock.iter_content())
lines = list(stream_mock.iter_lines())
self.assertEqual(len(lines), 3)

Note

  • If json_response is provided, text_response is auto-generated as JSON string

  • If text_response is provided, content is auto-generated as encoded bytes

  • The mock includes iter_content and iter_lines for streaming support

  • Status codes < 400 automatically set the ‘ok’ property to True

_classSetupFailed = False#
_class_cleanups = []#

Test utilities for urllib3 HTTP client library.

This module provides specialized test case utilities for testing code that uses urllib3, the low-level HTTP library that powers requests. It extends the base HTTP test functionality with urllib3-specific mock objects and response simulation.

Key features: - Comprehensive HTTPResponse mock objects with full urllib3 API support - Streaming and chunked response simulation - HTTP/1.1 and HTTP/2.0 version support - Connection pool and retry mechanism mocking - Low-level socket-like reading methods (read, read1, readinto, etc.) - HTTPHeaderDict integration for case-insensitive headers

Example

Basic urllib3 response mocking:

from core_https.tests.urllib3_ import BaseUrllib3TestCases

class MyUrllib3Test(BaseUrllib3TestCases):
    def test_http_response(self):
        mock_response = self.get_urllib3_mock(
            data=b'{"users": ["alice", "bob"]}',
            status=200,
            headers={"Content-Type": "application/json"}
        )

        # Test urllib3-specific methods
        self.assertEqual(mock_response.status, 200)
        self.assertEqual(mock_response.json(), {"users": ["alice", "bob"]})
        self.assertIsInstance(mock_response.headers, HTTPHeaderDict)

Note

This module creates very comprehensive mocks that simulate most urllib3.HTTPResponse features. The complexity is necessary to support urllib3’s extensive low-level API.

See also

  • BaseHttpTestCases: Base class for HTTP testing utilities

  • BaseRequestsTestCases: Test utilities for requests library

  • BaseAiohttpTestCases: Test utilities for aiohttp library

class core_https.tests.urllib3_.BaseUrllib3TestCases(methodName='runTest')[source]#

Bases: BaseHttpTestCases

Test utilities for urllib3 HTTP client library.

This class provides specialized methods for creating mock urllib3.HTTPResponse objects that closely simulate the behavior of real urllib3 responses. It extends BaseHttpTestCases with urllib3-specific functionality including streaming, connection management, and low-level socket operations.

The class creates comprehensive mocks with extensive parameter control to support testing of code that relies on urllib3’s low-level HTTP functionality.

Example

Creating comprehensive urllib3 response mocks:

class MyUrllib3Test(BaseUrllib3TestCases):
    def test_streaming_response(self):
        mock = self.get_urllib3_mock(
            data=b"chunk1\nchunk2\nchunk3",
            status=200,
            preload_content=False
        )

        # Test streaming methods
        chunks = list(mock.stream(chunk_size=10))
        self.assertGreater(len(chunks), 0)

    def test_http2_response(self):
        mock = self.get_urllib3_mock(
            status=200,
            version=20,
            version_string="HTTP/2.0"
        )

        self.assertEqual(mock.version, 20)
        self.assertEqual(mock.version_string, "HTTP/2.0")

Testing low-level socket operations:

def test_socket_methods(self):
    data = b"Hello, World!"
    mock = self.get_urllib3_mock(data=data)

    # Test socket-like reading methods
    buffer = bytearray(5)
    bytes_read = mock.readinto(buffer)
    self.assertEqual(bytes_read, 5)
    self.assertEqual(buffer, b"Hello")

Note

The urllib3 mocks are the most comprehensive in this test suite due to urllib3’s extensive low-level API. They include socket-like methods, streaming support, and connection management features.

classmethod get_urllib3_mock(url: str = 'https://example.com', method: str = 'GET', status: int = 200, data: bytes = b'{"message": "success"}', headers: Dict[str, str] | None = None, reason: str | None = None, version: int = 11, preload_content: bool = True, decode_content: bool = True, version_string: str = 'HTTP/1.1', original_response: Mock | None = None, pool: Mock | None = None, connection: Mock | None = None, msg: Mock | None = None, retries: Mock | None = None, enforce_content_length: bool = False, with_json_attr: bool = True) Mock[source]#

Create a comprehensive mock urllib3.HTTPResponse object for testing.

This method creates a highly detailed mock that simulates urllib3’s HTTPResponse behavior including streaming, connection management, and low-level socket operations. It supports both HTTP/1.1 and HTTP/2.0 protocols with extensive customization options.

Parameters:
  • url – The URL that the mock response represents. Defaults to “https://example.com”.

  • method – HTTP method for the request. Defaults to “GET”.

  • status – HTTP status code for the response. Defaults to 200.

  • data – Raw bytes content of the response body. Defaults to b’{“message”: “success”}’.

  • headers – Dictionary of response headers (case-insensitive via HTTPHeaderDict).

  • reason – HTTP status reason phrase. Auto-generated from status code if not provided.

  • version – HTTP version as integer (11 for HTTP/1.1, 20 for HTTP/2.0). Defaults to 11.

  • preload_content – Whether response content is preloaded. Affects streaming behavior.

  • decode_content – Whether to decode content based on Content-Encoding headers.

  • version_string – HTTP version string representation. Defaults to “HTTP/1.1”.

  • original_response – Mock of the original raw response object.

  • pool – Mock of the urllib3 connection pool.

  • connection – Mock of the underlying HTTP connection.

  • msg – Mock of the HTTP message object (httplib.HTTPMessage).

  • retries – Mock of the retry configuration object.

  • enforce_content_length – Whether to enforce Content-Length header validation.

  • with_json_attr – Whether to include a json() method for convenience. Defaults to True.

Returns:

Mock object simulating urllib3.HTTPResponse with comprehensive API support.

Example

Basic HTTP response mock:

mock_response = BaseUrllib3TestCases.get_urllib3_mock(
    data=b'{"users": ["alice", "bob"]}',
    status=200,
    headers={"Content-Type": "application/json"}
)

# Test standard response properties
self.assertEqual(mock_response.status, 200)
self.assertEqual(mock_response.json()["users"], ["alice", "bob"])
self.assertIsInstance(mock_response.headers, HTTPHeaderDict)

Streaming response simulation:

stream_mock = BaseUrllib3TestCases.get_urllib3_mock(
    data=b"chunk1\nchunk2\nchunk3",
    preload_content=False
)

# Test streaming methods
chunks = list(stream_mock.stream(chunk_size=10))
lines = stream_mock.readlines()
self.assertGreater(len(chunks), 0)

HTTP/2.0 response:

http2_mock = BaseUrllib3TestCases.get_urllib3_mock(
    status=200,
    version=20,
    version_string="HTTP/2.0",
    headers={"server": "nginx/1.20"}
)

self.assertEqual(http2_mock.version, 20)
self.assertEqual(http2_mock.version_string, "HTTP/2.0")

Socket-like operations:

socket_mock = BaseUrllib3TestCases.get_urllib3_mock(
    data=b"Hello, World!"
)

# Test low-level reading methods
buffer = bytearray(5)
bytes_read = socket_mock.readinto(buffer)
self.assertEqual(bytes_read, 5)
self.assertEqual(bytes(buffer), b"Hello")

# Test line-by-line reading
line = socket_mock.readline()
self.assertIsInstance(line, bytes)

Note

This mock provides extensive urllib3.HTTPResponse API compatibility including: - Standard attributes (status, headers, data, etc.) - Socket-like reading methods (read, read1, readinto, readline, readlines) - Streaming support (stream method with chunking) - Context manager protocol (__enter__, __exit__) - Iterator protocol for line-by-line reading - Header access methods (getheader, getheaders) - Connection management (close, release_conn) - JSON convenience method (if with_json_attr=True)

The complexity of this mock reflects urllib3’s comprehensive low-level HTTP API.

_classSetupFailed = False#
_class_cleanups = []#