(test) add resolver mutation and route-gap regression coverage

Expand resolver openapi middleware coverage, add mutation route tests, and include route-gap regression workflows with supporting test config updates.

Author: stefankumarasinghe

.gitignore

@@ -12,3 +12,4 @@ coverage.xml
 .vscode/
 .hypothesis/
 .venv/
+mutants/

middleware/openapi.py

@@ -19,7 +19,7 @@
 from fastapi import FastAPI
 from fastapi.openapi.utils import get_openapi
 
-JSON_SCHEMA_DIALECT = "https://json-schema.org/draft/2020-12/schema"
+JSON_SCHEMA_DIALECT = "https://spec.openapis.org/oas/3.1/dialect/base"
 _METHOD_ACTIONS: dict[str, str] = {
     "GET": "Retrieve",
     "POST": "Create",

openapi.json

@@ -2877,4 +2877,4 @@ components:
       scheme: bearer
       bearerFormat: JWT
       description: Context JWT carrying tenant and user scope for internal calls.
-jsonSchemaDialect: https://json-schema.org/draft/2020-12/schema
+jsonSchemaDialect: https://spec.openapis.org/oas/3.1/dialect/base

openapi.yaml

@@ -159,7 +159,8 @@ exclude = [
 [tool.pylint.main]
 jobs = 1
 extension-pkg-allow-list = ["sqlalchemy"]
-ignore = [".git", "__pycache__", ".mypy_cache", ".pytest_cache", ".ruff_cache", ".venv", "venv", "build", "dist", "tmp", "vendor", "tests"]
+ignore = [".git", "__pycache__", ".mypy_cache", ".pytest_cache", ".ruff_cache", ".venv", "venv", "build", "dist", "tmp", "vendor", "tests", "mutants"]
+ignore-paths = ["^tests/", "^mutants/"]
 init-hook = "import os, sys; cwd = os.getcwd(); sys.path.insert(0, cwd) if cwd not in sys.path else None"
 
 [tool.pylint.messages_control]

pyproject.toml

@@ -13,6 +13,8 @@
 
 import pytest
 
+os.environ.setdefault("MUTANT_UNDER_TEST", "")
+
 ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
 if ROOT not in sys.path:
     sys.path.insert(0, ROOT)

tests/conftest.py

@@ -0,0 +1,224 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+import pytest
+from fastapi import HTTPException
+
+from api.requests import AnalyzeJobCreateRequest, AnalyzeRequest, TopologyRequest
+from api.responses import JobStatus
+from api.responses.jobs import AnalyzeJobSummary as JobView
+from api.routes import analyze as analyze_route
+from api.routes import jobs as jobs_route
+from api.routes import topology as topology_route
+from services.security_service import InternalContext
+
+
+def _ctx() -> InternalContext:
+    return InternalContext(
+        tenant_id="tenant-a",
+        org_id="tenant-a",
+        user_id="u1",
+        username="alice",
+        permissions=["create:rca", "read:rca", "delete:rca"],
+        group_ids=[],
+        role="user",
+        is_superuser=False,
+    )
+
+
+def _job(status: JobStatus = JobStatus.COMPLETED) -> JobView:
+    now = datetime.now(UTC)
+    return JobView(
+        job_id="job-1",
+        report_id="rep-1",
+        status=status,
+        created_at=now,
+        started_at=now,
+        finished_at=now,
+        duration_ms=5,
+        error="transient",
+        summary_preview="ok",
+        tenant_id="tenant-a",
+        requested_by="u1",
+    )
+
+
+@pytest.mark.asyncio
+async def test_analyze_and_template_routes(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def fake_run(req: AnalyzeRequest):
+        return {
+            "tenant_id": req.tenant_id,
+            "start": req.start,
+            "end": req.end,
+            "duration_seconds": req.end - req.start,
+            "metric_anomalies": [],
+            "log_bursts": [],
+            "log_patterns": [],
+            "service_latency": [],
+            "error_propagation": [],
+            "slo_alerts": [],
+            "root_causes": [],
+            "ranked_causes": [],
+            "change_points": [],
+            "log_metric_links": [],
+            "forecasts": [],
+            "degradation_signals": [],
+            "anomaly_clusters": [],
+            "granger_results": [],
+            "bayesian_scores": [],
+            "analysis_warnings": [],
+            "overall_severity": "low",
+            "summary": "ok",
+            "quality": {
+                "anomaly_density": {},
+                "suppression_counts": {},
+                "gating_profile": None,
+                "confidence_calibration_version": None,
+            },
+        }
+
+    monkeypatch.setattr(analyze_route, "run_analysis", fake_run)
+    monkeypatch.setattr(
+        analyze_route.analysis_config_service,
+        "template_response",
+        lambda: {
+            "version": 1,
+            "defaults": {"request": {"step": "15s"}},
+            "template_yaml": "version: 1",
+            "file_name": "resolver-rca-defaults.yaml",
+        },
+    )
+
+    out = await analyze_route.analyze(AnalyzeRequest(tenant_id="tenant-a", start=1, end=2))
+    template = await analyze_route.analyze_config_template()
+
+    assert out["tenant_id"] == "tenant-a"
+    assert template.file_name == "resolver-rca-defaults.yaml"
+
+
+@pytest.mark.asyncio
+async def test_jobs_routes(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(jobs_route, "ensure_permission", lambda _name: None)
+    monkeypatch.setattr(jobs_route, "get_internal_context", _ctx)
+
+    async def fake_create(payload, ctx):
+        assert payload.tenant_id == "tenant-a"
+        assert ctx.user_id == "u1"
+        return _job(JobStatus.QUEUED)
+
+    async def fake_list(ctx, status_filter, limit, cursor):
+        assert status_filter == JobStatus.COMPLETED
+        assert limit == 5
+        assert cursor == "c1"
+        return [_job(JobStatus.COMPLETED)], "c2"
+
+    async def fake_get(job_id, ctx):
+        assert job_id == "job-1"
+        assert ctx.tenant_id == "tenant-a"
+        return _job()
+
+    async def fake_result(job_id, ctx):
+        assert job_id == "job-1"
+        assert ctx.tenant_id == "tenant-a"
+        return _job(), {"report": True}
+
+    async def fake_report(report_id, ctx):
+        assert report_id == "rep-1"
+        assert ctx.user_id == "u1"
+        return _job(), {"report_id": report_id}
+
+    async def fake_delete(report_id, ctx):
+        assert report_id == "rep-1"
+        assert ctx.user_id == "u1"
+        return None
+
+    monkeypatch.setattr(jobs_route.rca_job_service, "create_job", fake_create)
+    monkeypatch.setattr(jobs_route.rca_job_service, "list_jobs", fake_list)
+    monkeypatch.setattr(jobs_route.rca_job_service, "get_job", fake_get)
+    monkeypatch.setattr(jobs_route.rca_job_service, "get_job_result", fake_result)
+    monkeypatch.setattr(jobs_route.rca_job_service, "get_report", fake_report)
+    monkeypatch.setattr(jobs_route.rca_job_service, "delete_report", fake_delete)
+
+    created = await jobs_route.create_job(AnalyzeJobCreateRequest(tenant_id="tenant-a", start=1, end=2))
+    listed = await jobs_route.list_jobs(status_filter=JobStatus.COMPLETED, limit=5, cursor="c1")
+    summary = await jobs_route.get_job("job-1")
+    result = await jobs_route.get_job_result("job-1")
+    report = await jobs_route.get_report("rep-1")
+    deleted = await jobs_route.delete_report("rep-1")
+
+    assert created.status == JobStatus.QUEUED
+    assert listed.next_cursor == "c2"
+    assert listed.items[0].summary_preview == "ok"
+    assert summary.job_id == "job-1"
+    assert summary.summary_preview == "ok"
+    assert result.result == {"report": True}
+    assert report.result == {"report_id": "rep-1"}
+    assert deleted.deleted is True
+    assert "started_at" in listed.items[0].model_fields_set
+    assert "finished_at" in listed.items[0].model_fields_set
+    assert "duration_ms" in listed.items[0].model_fields_set
+    assert listed.items[0].error == "transient"
+    assert "summary_preview" in listed.items[0].model_fields_set
+    assert summary.started_at is not None
+    assert summary.finished_at is not None
+    assert summary.duration_ms == 5
+    assert summary.error == "transient"
+    assert "started_at" in summary.model_fields_set
+    assert "finished_at" in summary.model_fields_set
+    assert "duration_ms" in summary.model_fields_set
+    assert "error" in summary.model_fields_set
+    assert "summary_preview" in summary.model_fields_set
+
+
+def test_jobs_required_context_raises_when_missing(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(jobs_route, "get_internal_context", lambda: None)
+    with pytest.raises(HTTPException) as exc:
+        jobs_route._required_context()
+    assert exc.value.status_code == 401
+    assert exc.value.detail == "Missing internal context"
+
+def test_jobs_require_permission_forwards_name(monkeypatch: pytest.MonkeyPatch) -> None:
+    seen = {}
+
+    def fake_ensure_permission(name: str) -> None:
+        seen["name"] = name
+
+    monkeypatch.setattr(jobs_route, "ensure_permission", fake_ensure_permission)
+    jobs_route._require_permission("read:rca")
+    assert seen["name"] == "read:rca"
+
+@pytest.mark.asyncio
+async def test_topology_route(monkeypatch: pytest.MonkeyPatch) -> None:
+    class DummyProvider:
+        async def query_traces(self, filters, start, end):
+            assert filters == {}
+            assert start == 1
+            assert end == 100
+            return {
+                "traces": [
+                    {
+                        "rootServiceName": "checkout",
+                        "spanSet": {
+                            "spans": [
+                                {
+                                    "attributes": [
+                                        {"key": "service.name", "value": {"stringValue": "checkout"}},
+                                        {"key": "peer.service", "value": {"stringValue": "payments"}},
+                                    ]
+                                }
+                            ]
+                        },
+                    }
+                ]
+            }
+
+    monkeypatch.setattr(topology_route, "enforce_request_tenant", lambda req: req)
+    monkeypatch.setattr(topology_route, "safe_call", lambda coro: coro)
+    monkeypatch.setattr(topology_route, "get_provider", lambda _tid: DummyProvider())
+
+    out = await topology_route.blast_radius(
+        TopologyRequest(tenant_id="tenant-a", start=1, end=100, root_service="checkout", max_depth=2)
+    )
+    assert out["root_service"] == "checkout"
+    assert "payments" in out["affected_downstream"]

tests/test_mutation_routes.py

@@ -10,6 +10,8 @@
 
 from __future__ import annotations
 
+import tomllib
+
 from fastapi import FastAPI
 
 from middleware import openapi as openapi_middleware
@@ -40,6 +42,60 @@ def test_apply_inferred_responses_for_secured_and_ready_paths() -> None:
     assert ready_responses["503"]["description"] == "Service Unavailable"
 
 
+def test_project_version_helpers_cover_success_and_fallback(monkeypatch) -> None:
+    original_read_text = openapi_middleware.Path.read_text
+
+    monkeypatch.setattr(
+        openapi_middleware.Path,
+        "read_text",
+        lambda self, encoding="utf-8": "[project]\nversion = '9.9.9'\n",
+    )
+    assert openapi_middleware._project_version() == "9.9.9"
+
+    monkeypatch.setattr(
+        openapi_middleware.Path,
+        "read_text",
+        lambda self, encoding="utf-8": "[project]\nversion = ''\n",
+    )
+    assert openapi_middleware._project_version() == openapi_middleware._DEFAULT_APP_VERSION
+
+    monkeypatch.setattr(
+        openapi_middleware.Path,
+        "read_text",
+        lambda self, encoding="utf-8": (_ for _ in ()).throw(OSError("boom")),
+    )
+    assert openapi_middleware._project_version() == openapi_middleware._DEFAULT_APP_VERSION
+
+    monkeypatch.setattr(
+        openapi_middleware.Path,
+        "read_text",
+        lambda self, encoding="utf-8": "bad-toml",
+    )
+    monkeypatch.setattr(
+        openapi_middleware.tomllib,
+        "loads",
+        lambda _text: (_ for _ in ()).throw(tomllib.TOMLDecodeError("bad", "", 0)),
+    )
+    assert openapi_middleware._project_version() == openapi_middleware._DEFAULT_APP_VERSION
+
+    monkeypatch.setattr(openapi_middleware.Path, "read_text", original_read_text)
+
+
+def test_install_custom_openapi_sets_info_version(monkeypatch) -> None:
+    app = FastAPI(title="Resolver", version="ignored", description="desc")
+    openapi_middleware.install_custom_openapi(app)
+
+    monkeypatch.setattr(
+        openapi_middleware,
+        "get_openapi",
+        lambda **_kwargs: {"info": {"title": "Resolver"}, "paths": {}},
+    )
+    monkeypatch.setattr(openapi_middleware, "_project_version", lambda: "1.2.3")
+
+    generated = app.openapi()
+    assert generated["info"]["version"] == "1.2.3"
+
+
 def test_install_custom_openapi_cache_and_schema_walk(monkeypatch) -> None:
     app = FastAPI()
     app.openapi_schema = {"cached": True}
@@ -105,7 +161,7 @@ def test_install_custom_openapi_cache_and_schema_walk(monkeypatch) -> None:
     ]
     ready_responses = generated["paths"]["/api/v1/ready"]["get"]["responses"]
     assert ready_responses["503"]["description"] == "Service Unavailable"
-    assert generated["jsonSchemaDialect"] == "https://json-schema.org/draft/2020-12/schema"
+    assert generated["jsonSchemaDialect"] == "https://spec.openapis.org/oas/3.1/dialect/base"
     schemes = generated["components"]["securitySchemes"]
     assert schemes["ServiceToken"]["name"] == "x-service-token"
     assert schemes["ContextBearer"]["scheme"] == "bearer"