From 47f2ca8d5feacfb0266e6128419d14ece3a56dd5 Mon Sep 17 00:00:00 2001 From: anti Date: Mon, 20 Apr 2026 01:27:39 -0400 Subject: [PATCH] added(tests): schemathesis contract fuzzing at the agent and swarmctl level --- tests/api/test_schemathesis_agent.py | 100 +++++++++++++++++++++++++++ tests/api/test_schemathesis_swarm.py | 67 ++++++++++++++++++ 2 files changed, 167 insertions(+) create mode 100644 tests/api/test_schemathesis_agent.py create mode 100644 tests/api/test_schemathesis_swarm.py diff --git a/tests/api/test_schemathesis_agent.py b/tests/api/test_schemathesis_agent.py new file mode 100644 index 0000000..ae3f0da --- /dev/null +++ b/tests/api/test_schemathesis_agent.py @@ -0,0 +1,100 @@ +"""Schemathesis contract tests for the worker-side agent API +(``decnet.agent.app``). + +Uses schemathesis's ASGI transport. The agent's real security is +transport-layer mTLS — out of scope here; we're validating schema +conformance only. + +The executor and heartbeat modules are stubbed so fuzzed requests don't +actually deploy containers, tear down services, or self-destruct the host. +""" +from __future__ import annotations + +import pytest +import schemathesis as st +from schemathesis.specs.openapi.checks import ( + status_code_conformance, + content_type_conformance, + response_headers_conformance, + response_schema_conformance, +) +from hypothesis import settings, HealthCheck + +from decnet.agent import app as _agent_app_mod +from decnet.agent import executor as _exec +from decnet.agent import heartbeat as _heartbeat + + +# --------------------------------------------------------------------------- +# Safety stubs — fuzzer must never touch real docker / systemd / disk. +# Applied via autouse fixture (NOT module-level assignment) so the stubs +# don't leak into tests/swarm/test_agent_app.py which imports the same +# executor module. +# --------------------------------------------------------------------------- + +async def _noop_deploy(*a, **kw): + return {"status": "stub"} + +async def _noop_teardown(*a, **kw): + return {"status": "stub"} + +async def _noop_self_destruct(*a, **kw): + return {"status": "stub"} + +async def _noop_status(*a, **kw): + return {"deckies": [], "running": False, "deployed": False} + + +@pytest.fixture(autouse=True) +def _stub_agent_executor(monkeypatch): + monkeypatch.setattr(_exec, "deploy", _noop_deploy) + monkeypatch.setattr(_exec, "teardown", _noop_teardown) + monkeypatch.setattr(_exec, "self_destruct", _noop_self_destruct) + monkeypatch.setattr(_exec, "status", _noop_status) + async def _noop_async(*a, **kw): + return None + monkeypatch.setattr(_heartbeat, "start", lambda *a, **kw: None) + # stop() is awaited by the lifespan — must be a coroutine function. + monkeypatch.setattr(_heartbeat, "stop", _noop_async) + yield + +# OpenAPI is disabled on the worker by default (narrow attack surface). +# FastAPI only wires up /openapi.json during __init__; changing the attribute +# after the fact is a no-op, so register the route explicitly for the fuzzer. +_agent_app_mod.app.openapi_url = "/openapi.json" + +@_agent_app_mod.app.get("/openapi.json", include_in_schema=False) +async def _openapi_contract_test(): + return _agent_app_mod.app.openapi() + + +SCHEMA = st.openapi.from_asgi("/openapi.json", _agent_app_mod.app) + +pytestmark = pytest.mark.fuzz + +CHECKS = ( + # Intentionally omit `not_a_server_error`: /mutate returns a documented + # 501 Not Implemented, which that check flags as a failure regardless of + # whether the status is in the schema. `status_code_conformance` already + # catches *undocumented* 5xx responses. + status_code_conformance, + content_type_conformance, + response_headers_conformance, + response_schema_conformance, +) + + +@pytest.mark.fuzz +@SCHEMA.parametrize() +@settings( + max_examples=300, + deadline=None, + suppress_health_check=[ + HealthCheck.filter_too_much, + HealthCheck.too_slow, + HealthCheck.data_too_large, + ], +) +def test_agent_schema_compliance(case): + """Fuzz the agent routes against the worker OpenAPI schema.""" + case.call_and_validate(checks=CHECKS) diff --git a/tests/api/test_schemathesis_swarm.py b/tests/api/test_schemathesis_swarm.py new file mode 100644 index 0000000..0c5dfb3 --- /dev/null +++ b/tests/api/test_schemathesis_swarm.py @@ -0,0 +1,67 @@ +"""Schemathesis contract tests for the swarm-controller API +(``decnet.web.swarm_api``). + +Uses schemathesis's ASGI transport so we don't have to stand up uvicorn +with mTLS. The controller's transport-layer mTLS is out of scope here — +we're validating schema/behavioral conformance of its routes. +""" +from __future__ import annotations + +import os + +# Must be set BEFORE importing the swarm_api module — the repo factory +# reads DECNET_DB_TYPE at import time via dependencies.py. +os.environ["DECNET_DB_TYPE"] = "sqlite" +os.environ["DECNET_MODE"] = "master" +os.environ.setdefault("DECNET_JWT_SECRET", "schemathesis-swarm-secret-32chars-min-pad") + +import pytest +import schemathesis as st +from schemathesis.checks import not_a_server_error +from schemathesis.specs.openapi.checks import ( + status_code_conformance, + content_type_conformance, + response_headers_conformance, + response_schema_conformance, +) +from hypothesis import settings, HealthCheck + +from decnet.web import swarm_api as _swarm_api + +# OpenAPI is disabled by default on the controller (internal surface). +# FastAPI only wires /openapi.json during __init__; toggling the attribute +# post-hoc is a no-op, so register the route explicitly here. +_swarm_api.app.openapi_url = "/openapi.json" + +@_swarm_api.app.get("/openapi.json", include_in_schema=False) +async def _openapi_contract_test(): + return _swarm_api.app.openapi() + + +SCHEMA = st.openapi.from_asgi("/openapi.json", _swarm_api.app) + +pytestmark = pytest.mark.fuzz + +CHECKS = ( + not_a_server_error, + status_code_conformance, + content_type_conformance, + response_headers_conformance, + response_schema_conformance, +) + + +@pytest.mark.fuzz +@SCHEMA.parametrize() +@settings( + max_examples=200, + deadline=None, + suppress_health_check=[ + HealthCheck.filter_too_much, + HealthCheck.too_slow, + HealthCheck.data_too_large, + ], +) +def test_swarm_schema_compliance(case): + """Fuzz the swarm-controller routes against its OpenAPI schema.""" + case.call_and_validate(checks=CHECKS)