Mocket: A Comprehensive Socket Mocking Framework for Python

Introduction

Mocket is a versatile socket mock framework for Python, designed to simplify testing of applications that interact with network services. By monkey-patching the socket and ssl modules, Mocket allows developers to intercept and control network communication, making it ideal for unit and integration tests. It supports a wide range of scenarios, from mocking simple HTTP/HTTPS requests to complex custom protocols, and integrates seamlessly with asyncio, gevent, and even MicroPython's urequests.

Installation

Installing Mocket is straightforward using pip:

pip install mocket

For enhanced performance, Mocket can utilize xxhash instead of hashlib.md5 for creating hashes. Install with speedups:

pip install mocket[speedups]

Examples

Mocket provides flexible ways to mock network interactions, whether through decorators, context managers, or direct API calls.

Mocking HTTP[S] Calls

Here's how to mock an HTTP[S] request using Mocket's decorator or context manager:

import json

from mocket import mocketize, Mocketizer
from mocket.mocks.mockhttp import Entry
import requests
import pytest


@pytest.fixture
def response():
    return {
        "integer": 1,
        "string": "asd",
        "boolean": False,
    }


@mocketize  # Use its decorator
def test_json(response):
    url_to_mock = 'https://testme.org/json'

    Entry.single_register(
        Entry.GET,
        url_to_mock,
        body=json.dumps(response),
        headers={'content-type': 'application/json'}
    )

    mocked_response = requests.get(url_to_mock).json()

    assert response == mocked_response

# OR use its context manager
def test_json_with_context_manager(response):
    url_to_mock = 'https://testme.org/json'

    Entry.single_register(
        Entry.GET,
        url_to_mock,
        body=json.dumps(response),
        headers={'content-type': 'application/json'}
    )

    with Mocketizer():
        mocked_response = requests.get(url_to_mock).json()

    assert response == mocked_response

Preventing Real Network Access (Strict Mode)

To ensure your tests do not accidentally hit the real network, Mocket offers a strict mode:

from mocket import Mocketizer, mocketize
import requests
import pytest
from mocket.exceptions import StrictMocketException

with Mocketizer(strict_mode=True):
    with pytest.raises(StrictMocketException):
        requests.get("https://duckduckgo.com/")

# OR

@mocketize(strict_mode=True)
def test_get():
    with pytest.raises(StrictMocketException):
        requests.get("https://duckduckgo.com/")

You can also specify allowed hosts in strict mode:

from mocket import Mocketizer

with Mocketizer(strict_mode=True, strict_mode_allowed=["localhost", ("intake.ourmetrics.net", 443)]):
    # Your test code here
    pass

Faking Socket Errors

Testing error paths is crucial. Mocket allows you to simulate socket errors:

import socket
import requests
from unittest import TestCase
from mocket import mocketize
from mocket.mocks.mockhttp import Entry

class ErrorHandlingTestCase(TestCase):
    @mocketize
    def test_raise_exception(self):
        url = "http://github.com/fluidicon.png"
        Entry.single_register(Entry.GET, url, exception=socket.error())
        with self.assertRaises(requests.exceptions.ConnectionError):
            requests.get(url)

Custom Request Matching Logic

For complex scenarios, can_handle_fun allows defining custom logic for matching requests:

import json
import re

from mocket import mocketize
from mocket.mocks.mockhttp import Entry
import requests


@mocketize
def test_can_handle():
    url = "https://httpbin.org"

    Entry.single_register(
        Entry.GET,
        url,
        body=json.dumps({"message": "Nope... not this time!"}),
        headers={"content-type": "application/json"},
        can_handle_fun=lambda path, qs_dict: path == "/ip" and qs_dict,
    )
    Entry.single_register(
        Entry.GET,
        url,
        body=json.dumps({"message": "There you go!"}),
        headers={"content-type": "application/json"},
        can_handle_fun=lambda path, qs_dict: path == "/ip" and not qs_dict,
    )

    resp = requests.get("https://httpbin.org/ip")
    assert resp.status_code == 200
    assert resp.json() == {"message": "There you go!"}

# Example of regex path matching
Entry.single_register(
    Entry.GET,
    "https://api.example.com",
    body="ok",
    can_handle_fun=lambda path, qs_dict: bool(re.match(r"^/users/\\d+$", path)),
)

Recording Real Socket Traffic

Mocket can also record real socket traffic, similar to VCRpy, for later playback or analysis:

import json
import os
import tempfile
import io
import requests

from mocket import mocketize, Mocket

@mocketize(truesocket_recording_dir=tempfile.mkdtemp())
def test_truesendall_with_recording_https():
    url = 'https://httpbin.org/ip'

    requests.get(url, headers={"Accept": "application/json"})
    resp = requests.get(url, headers={"Accept": "application/json"})
    assert resp.status_code == 200

    dump_filename = os.path.join(
        Mocket.get_truesocket_recording_dir(),
        Mocket.get_namespace() + '.json',
    )
    with io.open(dump_filename) as f:
        response = json.load(f)

    assert len(response['httpbin.org']['443'].keys()) == 1

HTTPretty Compatibility

Mocket offers a compatibility layer for HTTPretty, allowing for an easier migration:

import json
import aiohttp
import asyncio
from unittest import TestCase

from mocket.plugins.httpretty import httpretty, httprettified


class AioHttpEntryTestCase(TestCase):
    @httprettified
    def test_https_session(self):
        url = 'https://httpbin.org/ip'
        httpretty.register_uri(
            httpretty.GET,
            url,
            body=json.dumps(dict(origin='127.0.0.1')),
        )

        async def main(l):
            async with aiohttp.ClientSession(
                loop=l, timeout=aiohttp.ClientTimeout(total=3)
            ) as session:
                async with session.get(url) as get_response:
                    assert get_response.status == 200
                    assert await get_response.text() == '{"origin": "127.0.0.1"}'

            loop = asyncio.new_event_loop()
            loop.set_debug(True)
            loop.run_until_complete(main(loop))

Asyncio Integration

Mocket works seamlessly with asyncio-based clients like aiohttp:

import json
import aiohttp
import pytest

from mocket import async_mocketize
from mocket.mocks.mockhttp import Entry
from mocket.plugins.aiohttp_connector import MocketTCPConnector


@pytest.mark.asyncio
@async_mocketize
async def test_aiohttp():
    """
    The alternative to using the custom `connector` would be importing
    `aiohttp` when Mocket is already in control (inside the decorated test).
    """

    url = "https://bar.foo/"
    data = {"message": "Hello"}

    Entry.single_register(
        Entry.GET,
        url,
        body=json.dumps(data),
        headers={"content-type": "application/json"},
    )

    async with aiohttp.ClientSession(
        timeout=aiohttp.ClientTimeout(total=3), connector=MocketTCPConnector()
    ) as session, session.get(url) as response:
        response = await response.json()
        assert response == data

Pook Integration

Mocket can also be used as the mocking engine for pook:

import pook
from mocket.plugins.pook_mock_engine import MocketEngine
import requests

pook.set_mock_engine(MocketEngine)
pook.on()

url = 'http://twitter.com/api/1/foobar'
status = 404
response_json = {'error': 'foo'}

mock = pook.get(
    url,
    headers={'content-type': 'application/json'},
    reply=status,
    response_json=response_json,
)
mock.persist()

requests.get(url)
assert mock.calls == 1

resp = requests.get(url)
assert resp.status_code == status
assert resp.json() == response_json
assert mock.calls == 2

Why Use Mocket?

Mocket stands out as a robust solution for testing network-dependent Python applications due to its comprehensive features:

• Versatile Mocking: It can mock any socket communication, not just HTTP, making it suitable for a wide array of protocols and clients.
• Flexibility: With features like can_handle_fun, you have fine-grained control over request matching, allowing for complex testing scenarios.
• Strict Mode: Prevents accidental external network calls, ensuring true isolation for your tests.
• Compatibility: Offers compatibility layers for popular libraries like HTTPretty and integrates well with asyncio, gevent, and pook.
• Error Simulation: Easily simulate network errors to test the robustness of your application's error handling.
• Recording: Ability to record and replay real network traffic for advanced testing and debugging.

Links

Explore Mocket further through these resources:

• GitHub Repository: https://github.com/mindflayer/python-mocket
• PyPI: https://pypi.org/project/mocket/
• openSUSE: https://software.opensuse.org/search?baseproject=ALL&q=mocket
• NixOS: https://search.nixos.org/packages?query=mocket
• ALT Linux: https://packages.altlinux.org/en/sisyphus/srpms/python3-module-mocket/
• NetBSD: https://cdn.netbsd.org/pub/pkgsrc/current/pkgsrc/devel/py-mocket/index.html
• AUR Arch Linux: https://aur.archlinux.org/packages/python-mocket
• Mocketoy (Custom Mock Example): https://github.com/mindflayer/mocketoy
• EuroPython 2013 Video: https://www.youtube.com/watch?v=-LvXbl5d02U
• EuroPython 2013 Slides (PDF): https://ep2013.europython.eu/media/conference/slides/mocket-a-socket-mock-framework.pdf
• Blog Post: Mocket is alive and is fighting with us: https://medium.com/p/mocket-is-alive-and-is-fighting-with-us-b2810d52597a
• Blog Post: Make development great again: https://hackernoon.com/make-development-great-again-faab769d264e
• Blog Post: HTTPretty now supports asyncio: https://hackernoon.com/httpretty-now-supports-asyncio-e310814704c6
• Blog Post: How to make your tests fail when they try to access the network: https://medium.com/@mindflayer/how-to-make-your-tests-fail-when-they-try-to-access-the-network-python-eb80090a6d24
• Blog Post: Testing in an asyncio world: https://medium.com/@mindflayer/testing-in-an-asyncio-world-a9a0ad41b0c5