Up and running in under 2 minutes

Install, initialize, and start using your SDK unchanged — no credentials, no Docker. See Docs for configuration details.

01

Prerequisites

MockMesh requires Python 3.9+ and pip.

bash
python --version   # 3.9 or higher required
pip --version      # 20.0 or higher
02

Install MockMesh

Install via pip with optional provider extras. Auto-detects installed packages — only activates interceptors for SDKs you actually use.

bash
# All providers (installs all optional dependencies)
pip install mockmesh

# Install only what you need
pip install mockmesh[aws]              # boto3 + botocore
pip install mockmesh[azure]            # azure-core + azure-* SDKs
pip install mockmesh[gcp]              # google-cloud-* SDKs
pip install mockmesh[aws,azure,gcp]    # multi-cloud
pip install mockmesh[kafka]            # confluent-kafka + kafka-python
pip install mockmesh[rabbitmq]         # pika
pip install mockmesh[sql]              # psycopg2, pymysql, asyncpg, etc.
pip install mockmesh[mongodb]          # pymongo
pip install mockmesh[redis]            # redis-py

# Recommended: virtual environment
python -m venv .venv && source .venv/bin/activate
pip install mockmesh

# Verify version
python -c "import mockmesh; print(mockmesh.__version__)"
# 0.0.1
03

Initialize in One Line

Call initialize() once at the top of your entry point or conftest.py. MockMesh auto-detects installed providers and activates only relevant interceptors — no SDK changes required.

python
import mockmesh

# Minimal — auto-detects providers, intercepts everything
engine = mockmesh.initialize()

# With custom config + dedicated workspace
engine = mockmesh.initialize(
    config_path    = "config/custom_overrides.json",
    storage_path   = "/tmp/mockmesh-local",
    fallback_mode  = "error",       # strict mode for tests
)

# With folder of per-service overrides (aws.json, azure.json, gcp.json, …)
engine = mockmesh.initialize(
    responses_path = "config/overrides/",
)

# Context manager — auto-shutdown on exit
with mockmesh.engine(fallback_mode="error") as mm:
    run_tests()

mockmesh.shutdown()   # manual cleanup if not using context manager
04

Use Your SDK Unchanged

Zero application code changes. boto3, azure-sdk, google-cloud, pymongo, redis, requests, confluent-kafka, pika — all behave identically to real services, but locally.

python
import boto3, mockmesh
engine = mockmesh.initialize()

# DynamoDB — stateful (Tier 3: storage-backed)
ddb = boto3.client("dynamodb", region_name="us-east-1")
ddb.put_item(TableName="Users", Item={"id":{"S":"u1"},"name":{"S":"Alice"}})
item = ddb.get_item(TableName="Users", Key={"id":{"S":"u1"}})
print(item["Item"]["name"]["S"])   # → Alice

# S3 — bytes saved to .mockmesh/blob/
s3 = boto3.client("s3", region_name="us-east-1")
s3.put_object(Bucket="assets", Key="logo.png", Body=b"\x89PNG...")
obj = s3.get_object(Bucket="assets", Key="logo.png")
print(obj["Body"].read()[:4])      # → b'\x89PNG'

# MongoDB — pymongo intercepted, in-memory store
import pymongo
db = pymongo.MongoClient().mydb
db.users.insert_one({"name": "Alice", "age": 30})
print(db.users.find_one({"name": "Alice"}))  # → {"name": "Alice", ...}

# GCP — google-cloud SDK intercepted via HTTP dispatch
from google.cloud import storage
client = storage.Client()
bucket = client.bucket("my-gcs-bucket")
blob = bucket.blob("data.json")
blob.upload_from_string('{"key": "value"}')
05

pytest Integration

Session-scoped for shared state, or use the context manager for full per-test isolation.

python
# conftest.py
import pytest, mockmesh

@pytest.fixture(scope="session", autouse=True)
def mock_cloud():
    engine = mockmesh.initialize(
        config_path    = "tests/fixtures/overrides.json",
        storage_path   = "/tmp/mm-test",
        fallback_mode  = "error",
    )
    yield engine
    mockmesh.shutdown()

# Full isolation per test (context manager):
def test_order_flow(tmp_path):
    with mockmesh.engine(
        config_path  = "tests/fixtures/overrides.json",
        storage_path = tmp_path,
    ) as mm:
        import boto3
        ddb = boto3.client("dynamodb", region_name="us-east-1")
        ddb.put_item(TableName="Orders",
                     Item={"id":{"S":"o1"},"total":{"N":"49.99"}})
        order = ddb.get_item(TableName="Orders",
                             Key={"id":{"S":"o1"}})["Item"]
        assert order["total"]["N"] == "49.99"

        # Pre-seed storage directly
        mm.storage.s3_put("assets", "config.json", b'{"env":"test"}')

All Parameters

config_pathstr | PathoptionalPath to a unified custom_overrides.json covering HTTP URL-match rules, AWS/Azure/GCP per-operation overrides, Kafka topic rules, and RabbitMQ queue rules.
responses_pathstr | PathoptionalFolder containing per-service override files: aws.json, azure.json, gcp.json, http.json, kafka.json, rabbitmq.json, sql.json, nosql.json, streaming.json. Only files present are applied; absent files fall through to built-in defaults.
storage_pathstr | PathoptionalRoot directory for the .mockmesh/ workspace. Defaults to cwd/.mockmesh. Use /tmp/... for ephemeral tests. Any *.json files at the workspace root are auto-detected as Tier 2 response overrides.
fallback_modestroptionalHow to handle interceptor errors: "mock" (default) returns generic response, "passthrough" calls real service, "error" raises InterceptError.
on_intercept_errorcallableoptionalCallback invoked when an interceptor fails. Receives (provider, operation, exception). Return value used as the response.
console_logbooloptionalMirror log records to stdout (default False — logs go to .mockmesh/logs/mockmesh.log only).
json_consolebooloptionalUse JSON format on stdout when console_log=True (default True). Set False for human-readable output.
log_levelintoptionalPython logging level (default logging.DEBUG).
pluginslist | NoneoptionalExplicit list of plugin instances. None (default) auto-detects installed providers and loads relevant interceptors.

Local Storage Layout

.mockmesh/logs/mockmesh.logStructured JSON log — rotating, 5 MB × 5 files. Nothing on stdout by default.
.mockmesh/sql/mockmesh.dbSQLite database for SQL interceptor and key-value config data.
.mockmesh/nosql/<Table>.jsonDynamoDB tables, SQS queues (sqs_<name>.json), Cosmos DB, and other NoSQL stores.
.mockmesh/blob/<bucket>/<key>Raw S3, Azure Blob, and GCS bytes. Companion .meta.json stores content-type and ETag.
.mockmesh/cache/ElastiCache metadata and Redis cache data.
.mockmesh/audit.logOne-line-per-operation audit trail for debugging.

Clean Install

pip uninstall mockmesh -yRemove the old installation first.
find . -name '*.pyc' -deleteDelete cached bytecode — stale .pyc files will run the old code even after reinstall.
find . -name '__pycache__' -type d | xargs rm -rfRemove all __pycache__ directories.
pip install -e .Reinstall in editable mode from the new source.
python -c "import mockmesh; print(mockmesh.__version__)"Confirm you see 0.0.1 (or later) before running tests.