Python SDK

Official Python client library for Mockarty — a multi-protocol mock server for HTTP, gRPC, MCP, GraphQL, SOAP, SSE, WebSocket, Kafka, RabbitMQ, and SMTP.

Installation

pip install mockarty

For async HTTP/2 support:

pip install mockarty[async]

Quick Start

Synchronous Client

from mockarty import MockartyClient, MockBuilder, AssertAction

# Create a client
client = MockartyClient(
    base_url="http://localhost:5770",
    api_key="your-api-key",
    namespace="sandbox",
)

# Create a mock using the builder
mock = (
    MockBuilder.http("/api/users/:id", "GET")
    .id("user-get")
    .respond(200, body={"id": "$.pathParam.id", "name": "$.fake.FirstName"})
    .ttl(3600)
    .build()
)

result = client.mocks.create(mock)
print(f"Created mock: {result.mock.id}")

# List mocks
page = client.mocks.list(namespace="sandbox", limit=10)
for m in page.items:
    print(f"  {m.id}")

# Health check
health = client.health.check()
print(f"Status: {health.status}")

client.close()

Async Client

import asyncio
from mockarty import AsyncMockartyClient, MockBuilder

async def main():
    async with AsyncMockartyClient(base_url="http://localhost:5770") as client:
        mock = MockBuilder.http("/api/hello", "GET").respond(200, body={"msg": "hello"}).build()
        result = await client.mocks.create(mock)
        print(f"Created: {result.mock.id}")

asyncio.run(main())

Context Manager

with MockartyClient() as client:
    client.mocks.create(mock)
    # client.close() is called automatically

Mock Builder

from mockarty import MockBuilder, AssertAction

# HTTP mock with conditions
mock = (
    MockBuilder.http("/api/orders", "POST")
    .id("create-order")
    .namespace("production")
    .tags("orders", "v2")
    .priority(10)
    .condition("$.body.amount", AssertAction.NOT_EMPTY)
    .header_condition("Authorization", AssertAction.NOT_EMPTY)
    .respond(201, body={
        "orderId": "$.fake.UUID",
        "amount": "$.req.amount",
        "status": "created",
    })
    .callback("https://webhook.site/test", method="POST", body={"event": "order.created"})
    .ttl(7200)
    .build()
)

# gRPC mock
grpc_mock = (
    MockBuilder.grpc("user.UserService", "GetUser")
    .id("grpc-get-user")
    .condition("$.id", AssertAction.NOT_EMPTY)
    .respond(200, body={"id": "$.req.id", "name": "$.fake.FirstName"})
    .build()
)

# MCP mock
mcp_mock = (
    MockBuilder.mcp("get_weather")
    .id("mcp-weather")
    .respond(200, body={"temperature": 22, "city": "$.req.city"})
    .build()
)

Working with Stores

with MockartyClient() as client:
    # Global store
    client.stores.global_set("counter", 0)
    data = client.stores.global_get()
    print(data)  # {"counter": 0}

    # Chain store
    client.stores.chain_set("order-flow", "orderId", "abc-123")
    chain_data = client.stores.chain_get("order-flow")
    print(chain_data)  # {"orderId": "abc-123"}

Error Handling

from mockarty import MockartyClient
from mockarty.errors import MockartyNotFoundError, MockartyAPIError

client = MockartyClient()
try:
    mock = client.mocks.get("non-existent")
except MockartyNotFoundError:
    print("Mock not found")
except MockartyAPIError as e:
    print(f"API error {e.status_code}: {e.message}")

Configuration

The client reads these environment variables as defaults:

Variable	Description	Default
MOCKARTY_BASE_URL	Mockarty server URL	http://localhost:5770
MOCKARTY_API_KEY	API authentication key	None

Fluent Tester DSL

For end-to-end tests that exercise multiple protocols, the mockarty.tester sub-package provides a fluent chain that mirrors the Go SDK 1:1 — so the same test reads identically across Go / Python / Java codebases:

from mockarty.tester import Tester

def test_user_signup():
    with Tester(base_url="http://localhost:8080") as t:
        (t.http().post("/signup")
            .json({"email": "a@b.c"})
            .expect_status(201)
            .extract("$.token", "token"))
        (t.http().get("/me")
            .header("Authorization", "Bearer {{token}}")
            .expect_status(200)
            .expect_json_path("$.email", "a@b.c"))
        t.graphql("/gql").query(
            "{ me { id } }", None,
        ).expect_status(200).expect_no_errors()

        assert t.ok(), t.errors()

Vocabulary: expect_status, expect_header, expect_body_contains, expect_json_path, expect_json_array_len, extract. Each chain emits one step record; group with wrap(t, "name", fn), retry with eventually(t, within, interval, fn), fan out with parallel(t, branch_a, branch_b).

Upload to TCM external runs

Ship a Tester chain as a TCM external run in one call:

from mockarty import MockartyClient
from mockarty.tester import Tester, to_report_kwargs

with MockartyClient(base_url="http://...", api_key="...", namespace="qa") as client:
    with Tester(base_url="http://localhost:8080") as t:
        t.http().get("/me").expect_status(200)

        client.external_runs.report(
            **to_report_kwargs(t, case_name="me-endpoint", auto_create=True),
        )

to_report_kwargs(t, **opts) maps the Tester report to the ExternalRunsAPI.report() kwargs shape: per-step protocol/method/url/statusOrCode go into metadata, multi-failure errors join with "; ", ISO-8601 UTC timestamps. Same vocabulary as the Go (tester.ToExternalRun) and Java (ExternalRunBridge) SDKs.

Sync the test inventory to TCM (discovery)

Where external_runs ships per-test results, discovery ships the full inventory of tests your suite knows about so the TCM catalogue mirrors the code base — including tests that didn't run. New tests are created, existing ones keep their human-authored metadata, and (with prune_missing=True) tests removed from the code are marked orphaned.

from mockarty import DiscoveryCase, MockartyClient

with MockartyClient(base_url="http://...", api_key="...", namespace="qa") as client:
    result = client.discovery.sync(
        source="pytest:auth-suite",   # scope key; pruning is scoped to it
        cases=[
            DiscoveryCase(
                full_name="tests/auth_test.py::test_login",  # deterministic identity
                name="test_login",
                suite="auth",
                source_ref="tests/auth_test.py:12",
                labels=["smoke"],
            ),
        ],
        framework="pytest",
        prune_missing=True,
    )
    print(result.created, result.updated, result.orphaned, result.total)

Or let the bundled pytest plugin build the manifest from every collected item for you — opt in with the --mockarty-discover flag:

MOCKARTY_BASE_URL=http://... MOCKARTY_API_KEY=... \
    pytest --mockarty-discover --mockarty-discover-source pytest:auth-suite

Add --mockarty-discover-prune to orphan tests removed from the code. The plugin is dormant until the flag (or MOCKARTY_DISCOVER=1) is set.

Protocol Clients

Beyond configuring mocks, the SDK ships test clients to drive the system under test for SOAP / GraphQL / SSE / WebSocket — each captures every call as a TCM step so the external run shows a per-call timeline:

• mockarty.protocols.soap — SOAP 1.1 / 1.2 with stdlib XML parsing
• mockarty.protocols.graphql — GraphQL with auto operation-name extraction
• mockarty.protocols.sse — Server-Sent Events (WHATWG parser, collect + stream)
• mockarty.protocols.websocket — WebSocket via the optional protocols extra
• mockarty.protocols.telemetry — shared Step / AccumulatingRecorder

from mockarty import Client
from mockarty.protocols import AccumulatingRecorder
from mockarty.protocols.graphql import GraphQLClient

client = Client("http://localhost:5770", api_token="...")
rec = AccumulatingRecorder()

gql = GraphQLClient("http://app/graphql", recorder=rec)
gql.execute("query GetUser { user { id } }")

# At test finish, push the captured timeline:
client.external_runs().report(case_name="my case", status="passed", steps=rec.steps())

For WebSocket support:

pip install "mockarty[protocols]"

Full cross-language reference (Python / Go / Java side-by-side, every protocol, options, classification rules, troubleshooting): SDK Protocol Clients.

pytest Integration

Install the test extras:

pip install mockarty[test]

Use the provided fixtures in your tests:

# conftest.py
pytest_plugins = ["mockarty.testing.fixtures"]

# test_example.py
def test_create_mock(mock_cleanup):
    from mockarty import MockBuilder
    mock = MockBuilder.http("/test", "GET").respond(200, body="ok").build()
    created = mock_cleanup(mock)
    assert created.id is not None

Examples

The examples/ directory has 25+ runnable scripts covering every facet of the SDK. The two most useful starting points:

Example	What it shows
kitchen_sink/	Full adopter showcase — Tester DSL chain (token-chain → HTTP → GraphQL → JSONPath assertions), wrap() grouping, TCM external-run upload via to_report_kwargs. Runs against the testbackend; no external infrastructure.
ci_cd_pipeline.py	Pytest test that emits a TCM run from a CI step — JUnit-style structure, Allure-compatible attachments.
agent_tasks.py	Tester DSL emitting external-run reports straight from a pytest function.

For protocol-specific snippets see http_mocks.py, graphql_mocks.py, grpc_mocks.py, soap_mocks.py, messaging_mocks.py, sse_mocks.py.

Allure Compatibility (default-ON)

Mockarty ships seamless Allure interop — existing Allure-based test suites run through Mockarty with zero refactor. Three usage styles work in the same project, in the same test file, mixed freely:

pip install mockarty[allure]   # pulls in allure-pytest 2.13+

1. Pure Allure — your existing code:

   import allure
   
   @allure.feature("auth")
   @allure.severity(allure.severity_level.CRITICAL)
   def test_login():
       with allure.step("submit login form"):
           ...

   Steps, attachments and labels flow into Mockarty's test-case accumulator automatically. The native allure-results/ output continues to work too — both sinks receive every event.

2. Mockarty native:

   from mockarty.testing import test_case, step, attach
   
   @test_case("CASE-LOGIN-1")
   def test_login():
       with step("submit login form"):
           attach("payload.json", b"{...}", content_type="application/json")

3. Drop-in alias — for migrating codebases that prefer a clean import swap:

   import mockarty.allure as allure   # surface mirrors `allure` 1:1
   
   @allure.feature("auth")
   def test_login():
       with allure.step("submit"):
           ...

Configuration

Env var	Default	Effect
MOCKARTY_ALLURE_MIRROR	on	Set to off to disable the Allure → Mockarty mirror entirely. The pytest plugin's listener is not registered when off.
MOCKARTY_ALLURE_SHIM	off	Set to on (advisory) to register mockarty.allure as sys.modules["allure"] when allure-pytest is not installed. Lets pure-Allure code run as no-op in environments without the real package.

How it works

The pytest plugin's pytest_configure hook registers a listener with allure_commons.plugin_manager that observes Allure's native lifecycle hooks (start_step, stop_step, attach_data, add_label, etc.). Each observed event is mirrored onto the active Mockarty CaseFrame — if the user's test has no explicit @mockarty.testing.test_case binding, the plugin opens an implicit frame keyed on the test node id so Allure decorators still produce a Mockarty case run.

The reverse direction (mockarty.testing.step → allure.step) has existed since 0.3.0 and continues to work. When both decorator families participate in a single step, the suppression context-var prevents double-counting.

License

This SDK is proprietary software, not open source. It is licensed under the Mockarty SDK License Agreement — see LICENSE for the full terms.

• Free for evaluation, learning, and non-commercial / community use.
• Commercial use requires a valid, paid Mockarty subscription. Using this SDK in production or for commercial advantage without a subscription is not permitted.

For commercial subscriptions and licensing inquiries, see mockarty.ru or contact orlovich.artem@gmail.com.