Testing

Running the test suite

# All tests (unit + integration — no external services needed)
poetry run pytest src -v

# One file
poetry run pytest src/tests/test_storage.py -v

# With coverage
poetry run pytest src --cov=opentremor_core --cov-report=term-missing

---

Test structure

src/tests/
├── opentremor-core-tests.yaml   In-memory config, auth disabled — used by every test
├── basic-plan.txt                      Sample multi-stack terraform plan fixture
├── test_api_integration.py             End-to-end API tests (httpx ASGI) — ingest/analyze/findings/quota/webhook flows
├── test_auth.py                        Auth disabled/bootstrap/key CRUD/org isolation/role enforcement
├── test_storage.py                     StorageBackend unit tests (InMemoryStorage + mocked MongoBackend), every collection
├── test_config.py                      AppConfig / load_config tests
├── test_formatter.py                   Jinja2 report template tests, incl. finding-status badges
├── test_hash.py                        Hash normalisation unit tests
├── test_plan_parser.py                 TerraformPlanParser unit tests
├── test_code_change_parser.py          TerraformCodeChangeParser unit tests
├── test_llm.py                         libs/llm/ providers — mocked SDK clients, no network
├── test_analysis_runner.py             run_analysis()/count_units() against InMemoryStorage + a fake LLMClient
├── test_github_client.py               GitHubClient — synthetic RSA keypair for JWT signing, mocked httpx transport
├── test_metrics.py                     Span recording, summary, usage
└── test_mcp_server.py                  MCP mount/disable/API key forwarding

---

Integration tests — no server required

test_api_integration.py uses httpx with ASGI transport. The FastAPI app is instantiated once per test module (scope="module"); no port is opened and no subprocess is started.

@pytest.fixture(scope="module")
def app():
    return create_app(config_file="tests/opentremor-core-tests.yaml")

@pytest.fixture
async def client(app):
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=app), base_url="http://test"
    ) as c:
        yield c

Benefits:

• Runs in CI with no external services
• Full coverage of routing, middleware, exception handlers, background jobs
• Each test class gets a unique namespace UUID → mostly isolated, no shared state

Module-scoped app means shared storage

Because app (and its underlying storage) is scoped to the whole test module, resource dedup (content-hash + org-scoped) means the same literal plan text used by two different test classes is only ever “analysed” once — the second usage sees it as already done. Tests that specifically exercise “did the LLM get called” (quota, findings triage, webhook flows) tag their input with the test’s own namespace UUID (a _tagged_plan()/similar helper regex-inserts a nonce comment into each resource block) so every test gets guaranteed-fresh content regardless of execution order. Follow this pattern for any new test that needs to assert on LLM call counts.

httpx.ASGITransport does not run the lifespan

httpx.ASGITransport never fires the ASGI lifespan.startup event, so any app-startup-only logic (e.g. _lifespan() hooks in api.py) does not run under these tests. Anything a test needs to exist (like the synthetic default org) must be seeded lazily on first request instead — see dependencies.py’s _ensure_default_org() for the pattern this codebase uses.

---

Mocking external services

LLM providers (test_llm.py, test_analysis_runner.py, test_api_integration.py) — never call a real provider. A FakeLLMClient(LLMClient) subclass returns a canned UnitAnalysis + token usage; get_llm_client is patched at the import site used by the router under test (opentremor_core.controllers.routers.analyzers.get_llm_client or ...routers.integrations.get_llm_client — these are separate references, patch the one actually imported by the code path you’re exercising).

GitHub API (test_github_client.py) — httpx.MockTransport(handler) injected into GitHubClient(transport=...); the handler inspects the request and returns a canned httpx.Response. test_api_integration.py’s webhook tests instead patch the GitHubClient class itself (an AsyncMock) since they’re exercising the router/job logic around it, not the client’s HTTP calls.

Motor / MongoDB (test_storage.py) — find() returns a sync MagicMock cursor; to_list() is an AsyncMock. No real MongoDB in unit tests — docker compose up mongodb is only needed for manual/exploratory testing against a MongoDB-backed config (the workspace root’s shared configs/opentremor-local.yaml).

cursor = MagicMock()
cursor.to_list = AsyncMock(return_value=docs)
col.find = MagicMock(return_value=cursor)

---

Example unit tests

test_storage.py

async def test_create_and_read(self):
    s = InMemoryStorage()
    await s.create_resource(ORG, "h1", _resource("h1"))
    doc = await s.read_resource(ORG, "h1")
    assert doc["hash"] == "h1"
    assert doc["analysis"] is None

test_plan_parser.py

def test_tricky_non_header_line_not_matched():
    # Lines ending with "will be destroyed" but without the "  # " prefix
    # must NOT be treated as resource block headers.
    units = TerraformPlanParser(TRICKY_LINE).parse()
    assert len(units) == 1

---

pytest configuration

pyproject.toml

[tool.pytest.ini_options]
asyncio_mode = "auto"   # all async def test_* run automatically
testpaths = ["src"]

---

Conventions

• Place tests in src/tests/test_<module>.py
• Use async def for any test touching a storage backend or making HTTP calls
• Use InMemoryStorage directly in unit tests — do not mock it
• Mock Motor collections for MongoBackend tests (no real MongoDB in unit tests)
• Generate a fresh uuid4() namespace per test (or reuse the shared namespace fixture) to prevent cross-test state
• When a test needs a guaranteed-fresh, never-before-analysed resource inside a module-scoped app, tag the input content with the namespace UUID rather than assuming isolation
• test_mcp_server.py tests are deselected when fastmcp is not installed