// documentation

MockMesh Documentation — Configuration, Providers & API

Provider guides, configuration reference, fallback modes, and plugin development.

Getting Started

Installation

bash
# All providers
pip install mockmesh

# Specific providers
pip install mockmesh[aws]              # AWS only
pip install mockmesh[aws,kafka]        # AWS + Kafka
pip install mockmesh[aws,azure,gcp]    # Multi-cloud
pip install mockmesh[sql]              # SQL databases
pip install mockmesh[mongodb]          # MongoDB
pip install mockmesh[redis]            # Redis

MockMesh auto-detects installed packages at startup. Only interceptors whose dependencies are present will activate.

Quick Start

python
import mockmesh

mockmesh.initialize()

# Now use any supported SDK normally.
import boto3
s3 = boto3.client("s3", region_name="us-east-1")
response = s3.list_buckets()  # Returns mock data, no network call

Context Manager

python
import mockmesh

with mockmesh.engine() as mm:
    import requests
    resp = requests.get("https://api.example.com/users")
    print(resp.json())  # Mock response

    # Pre-seed storage
    mm.storage.s3_put("my-bucket", "key.txt", b"hello world")

# MockMesh is automatically shut down here.

Storage Pre-seeding

python
with mockmesh.engine() as mm:
    # S3
    mm.storage.s3_put("my-bucket", "data.csv", b"id,name\n1,Alice\n")

    # DynamoDB
    mm.storage.dynamo_put("users", {"id": {"S": "user-1"}, "name": {"S": "Alice"}})

    # SSM Parameter Store
    mm.storage.ssm_put("/app/db-host", "localhost")

    # Secrets Manager
    mm.storage.secret_put("prod/api-key", "sk-test-12345")

    # Azure Blob
    mm.storage.az_blob_put("my-container", "file.txt", b"contents")

    # CosmosDB
    mm.storage.cosmos_upsert("users", {"id": "u-1", "name": "Alice"})

Supported Providers

ProviderSentinel PackageInstall Extra
HTTPrequestsmockmesh[http]
AWSbotocoremockmesh[aws]
Azureazure.coremockmesh[azure]
GCPrequestsmockmesh[gcp]
Kafkaconfluent_kafka or kafkamockmesh[kafka]
RabbitMQpikamockmesh[rabbitmq]
SQLpsycopg2, pymysql, asyncpg, etc.mockmesh[sql]
MongoDBpymongomockmesh[mongodb]
Redisredismockmesh[redis]

Provider Guides

AWS Provider

MockMesh patches botocore.endpoint.Endpoint.make_request, the single method through which every botocore service call passes. Once active, every boto3.client(...) call is intercepted. No AWS credentials required.

Covers 17 AWS services with 290+ operations: S3, DynamoDB, SQS, SNS, Lambda, ECS, EKS, ECR, RDS/Aurora (47 operations), ElastiCache, SecretsManager, SSM, STS, IAM, CloudWatch, CloudWatch Logs, EventBridge.

Storage Integration

ServiceStorage LocationDescription
S3.mockmesh/blob/Object bodies stored as files on disk
DynamoDB.mockmesh/nosql/Items stored as JSON documents keyed by table + primary key
SQS.mockmesh/nosql/Queue messages stored and dequeued in FIFO order

Code Examples

python
import mockmesh
mockmesh.initialize()
import boto3

# S3 — stateful write + read
s3 = boto3.client("s3", region_name="us-east-1")
s3.put_object(Bucket="my-bucket", Key="hello.txt", Body=b"Hello World")
obj = s3.get_object(Bucket="my-bucket", Key="hello.txt")
print(obj["Body"].read())  # b"Hello World"

# DynamoDB — write and query
ddb = boto3.client("dynamodb", region_name="us-east-1")
ddb.put_item(TableName="Users", Item={"id": {"S": "u-1"}, "name": {"S": "Alice"}})
resp = ddb.get_item(TableName="Users", Key={"id": {"S": "u-1"}})
print(resp["Item"]["name"]["S"])  # "Alice"

# SQS — send and receive
sqs = boto3.client("sqs", region_name="us-east-1")
sqs.send_message(QueueUrl="https://sqs.us-east-1.amazonaws.com/123/MyQueue",
                 MessageBody='{"order": 1}')
resp = sqs.receive_message(QueueUrl="https://sqs.us-east-1.amazonaws.com/123/MyQueue")
print(resp["Messages"][0]["Body"])

# RDS Aurora cluster
rds = boto3.client("rds", region_name="us-east-1")
rds.create_db_cluster(DBClusterIdentifier="aurora-cluster", Engine="aurora-mysql",
                      MasterUsername="admin", MasterUserPassword="secret")
clusters = rds.describe_db_clusters()
print(clusters["DBClusters"][0]["DBClusterIdentifier"])

Custom Configuration

Configuration Tiers

text
Tier 1 (highest)   responses_path folder     explicit folder passed to initialize()
Tier 2             .mockmesh/ auto-detect     *.json files found at the workspace root
Tier 3             Storage layer              live data from PutObject / PutItem / Enqueue
Tier 4 (lowest)    Built-in defaults          bundled aws.json / azure.json / gcp.json ...

Folder-Based Overrides (recommended)

python
# Option A — explicit folder
mockmesh.initialize(responses_path="my_overrides/")

# Option B — auto-detect via .mockmesh/
# Just drop files into .mockmesh/ and they are picked up automatically
mockmesh.initialize()

# You can combine both:
mockmesh.initialize(
    config_path="config/routing_rules.json",      # routing rules
    responses_path="config/overrides/",            # response bodies
)

_config Placeholders

Every built-in default JSON file includes a _config section. Provider-specific values (account IDs, regions) are defined once and referenced via {_config.xxx} placeholders, resolved at load time. Override _config values in your own files to change defaults across all responses.

json
{
  "_config": {
    "account_id": "999888777666",
    "region": "eu-west-1"
  },
  "s3": {
    "GetObject": {
      "Body": "eyJlbnYiOiJwcm9kIn0=",
      "ContentType": "application/json"
    }
  }
}

GCP Rules

json
{
  "gcp": {
    "rules": [
      {
        "match": { "service": "storage", "operation": "get" },
        "response": {
          "status": 200,
          "body": { "kind": "storage#object", "name": "custom-object", "bucket": "custom-bucket" }
        }
      },
      {
        "match": { "service": "secretmanager", "operation": "access" },
        "response": {
          "status": 200,
          "body": { "name": "projects/my-project/secrets/my-secret/versions/latest",
                    "payload": { "data": "Y3VzdG9tLXNlY3JldC12YWx1ZQ==" } }
        }
      }
    ]
  }
}

Hot-Swap at Runtime

python
from mockmesh.config.loader import ConfigLoader

engine = mockmesh.initialize()
original_resolver = engine.response_engine._resolver

# Load a different config mid-run
engine.response_engine.load(
    ConfigLoader(user_config_path="tests/fixtures/error_scenario.json").load()
)
try:
    run_error_scenario_tests()
finally:
    engine.response_engine.load(original_resolver)

Fallback Modes

The fallback system controls what happens when an interceptor fails to handle a request.

Three Modes

"mock" (default)

On interceptor error, return a generic mock response {"mocked": true}. The application continues running with synthetic data. Safest mode for development and testing.

"passthrough"

On interceptor error, call the original unpatched function so the request reaches the real service. Useful during gradual migration. Some interceptors that replace entire classes (Kafka, RabbitMQ) degrade to mock mode.

"error"

On interceptor error, raise InterceptError so the caller sees a clear failure. Use in CI pipelines or strict testing environments where silent fallbacks would mask problems.

Per-Provider Overrides

python
engine = mockmesh.initialize(fallback_mode="mock")

engine.fallback_config.provider_overrides["aws"] = "passthrough"
engine.fallback_config.provider_overrides["http"] = "error"
# All other providers use the global "mock" mode

engine.fallback_config.effective_mode("aws")    # "passthrough"
engine.fallback_config.effective_mode("kafka")  # "mock"

register_handler()

Register custom handlers for specific (provider, operation) pairs. Checked before built-in interceptor logic.

python
mockmesh.initialize()

def handle_transcribe(request):
    return {"TranscriptionJob": {"TranscriptionJobName": "my-job", "TranscriptionJobStatus": "COMPLETED"}}

mockmesh.register_handler("aws", "StartTranscriptionJob", handle_transcribe)

register_fallback()

Per-provider fallback functions invoked when the interceptor fails. Different from on_intercept_error — applies to a single provider.

python
mockmesh.initialize()

def aws_fallback(provider, operation, exception):
    print(f"AWS mock failed for {operation}, returning empty response")
    return {}

mockmesh.register_fallback("aws", aws_fallback)

Error Hierarchy

text
MockMeshError
  +-- InterceptError          (provider, operation, cause)
        +-- UnknownOperationError

Execution Order

  1. Custom handler — if registered for exact (provider, operation) pair
  2. Normal handler — interceptor's built-in handler
  3. on_intercept_error callback — if handler fails and callback registered
  4. Fallback mode — mock / passthrough / error

Plugin Development

Each provider is a plugin that registers with the central PluginRegistry. You can create custom plugins to intercept additional services.

BasePlugin ABC

python
from mockmesh.plugins.base import BasePlugin

class MyPlugin(BasePlugin):
    name = "my-service"

    def activate(self, config, *, storage=None, provider_defaults=None,
                 fallback_config=None, custom_handlers=None, **kwargs):
        self.storage = storage
        self._fallback_config = fallback_config
        self._custom_handlers = custom_handlers or {}
        self._active = True

    def deactivate(self):
        self._active = False

Registration

python
import mockmesh
from mockmesh.plugins.base import BasePlugin
from mockmesh.response.engine import ResponseEngine

engine = ResponseEngine()
mockmesh.register_plugin(MyPlugin(engine))
mockmesh.initialize()  # Your plugin activates alongside built-ins

# Or pass explicit plugin list:
mockmesh.initialize(plugins=[
    HttpInterceptor(engine),
    AwsInterceptor(engine),
    MyPlugin(engine),
])

URL Dispatch System

The HTTP interceptor supports URL dispatch — other interceptors can claim specific URL patterns. This is how GCP works.

python
from mockmesh.interceptors.http import HttpInterceptor

def is_my_service_url(url):
    return "my-service.example.com" in url

def handle_my_service(adapter_self, request, **kwargs):
    import requests.models
    resp = requests.models.Response()
    resp.status_code = 200
    resp._content = b'{"service": "my-service", "mocked": true}'
    resp.headers["Content-Type"] = "application/json"
    return resp

HttpInterceptor.register_url_dispatcher(
    provider="my-service",
    url_matcher=is_my_service_url,
    handler=handle_my_service,
)