"""V1-A — minimum query slice that proves the entity model end-to-end. Per ``docs/plans/engineering-v1-completion-plan.md`` Phase V1-A: the four pillar queries (Q-001 subsystem-scoped, Q-005, Q-006, Q-017) must run against a single seed graph and report the expected shapes. This file is intentionally separate from ``test_engineering_queries.py`` so the V1-A acceptance is auditable in isolation. The seed graph mirrors what the V1-A plan called for: one satisfying Component, one orphan Requirement, one supported ValidationClaim, one unsupported one, one Decision on a flagged Assumption. """ from __future__ import annotations import pytest from fastapi.testclient import TestClient from atocore.engineering.queries import ( evidence_chain, orphan_requirements, requirements_for, subsystem_contents, ) from atocore.engineering.service import ( create_entity, create_relationship, init_engineering_schema, ) from atocore.main import app from atocore.models.database import init_db @pytest.fixture def v1a_seed(tmp_data_dir): """Seed the Q-6 integration data exactly as V1-A's plan describes: one satisfying Component + one orphan Requirement + one Decision on a flagged Assumption + one supported ValidationClaim + one unsupported ValidationClaim, plus the Subsystem they live under.""" init_db() init_engineering_schema() # Subsystem (Q-001 target) optics = create_entity("subsystem", "Optics", project="p05-interferometer") # Components — one with PART_OF the subsystem and SATISFIES a requirement, # one without parents (orphan_components in Q-004; not the focus here). primary = create_entity("component", "Primary Mirror", project="p05-interferometer") diverger = create_entity("component", "Diverger Lens", project="p05-interferometer") create_relationship(primary.id, optics.id, "part_of") create_relationship(diverger.id, optics.id, "part_of") # Requirements — one satisfied by primary, one orphan (Q-006) req_satisfied = create_entity( "requirement", "Surface figure < 25 nm RMS", project="p05-interferometer" ) req_orphan = create_entity( "requirement", "Calibration repeatable to lambda/20", project="p05-interferometer" ) create_relationship(primary.id, req_satisfied.id, "satisfies") # Decision on a flagged Assumption (Q-009 surface — not asserted in V1-A # acceptance but useful as background data) flagged_assumption = create_entity( "parameter", "Vendor lead time = 6 weeks", project="p05-interferometer", properties={"flagged": True}, ) risky = create_entity( "decision", "Pre-order CGH from external vendor", project="p05-interferometer" ) create_relationship(risky.id, flagged_assumption.id, "based_on_assumption") # Validation claims — one supported by a Result, one unsupported (Q-017 # touches the supported one; Q-011 surfaces the unsupported one) fea_result = create_entity( "result", "FEA thermal sweep 2026-04", project="p05-interferometer" ) claim_supported = create_entity( "validation_claim", "Thermal margin is adequate", project="p05-interferometer" ) claim_unsupported = create_entity( "validation_claim", "Vibration isolation passes spec", project="p05-interferometer" ) create_relationship(fea_result.id, claim_supported.id, "supports") return { "subsystem": optics, "primary": primary, "diverger": diverger, "req_satisfied": req_satisfied, "req_orphan": req_orphan, "claim_supported": claim_supported, "claim_unsupported": claim_unsupported, "fea_result": fea_result, "risky_decision": risky, "flagged_assumption": flagged_assumption, } # --------------------------------------------------------------------------- # Q-001 subsystem-scoped — the V1-A code change # --------------------------------------------------------------------------- def test_subsystem_contents_returns_spec_shape(v1a_seed): """The shape must be exactly what the catalog promises: {subsystem, contains: [{id, entity_type, name, status}]}.""" result = subsystem_contents(v1a_seed["subsystem"].id) assert result is not None assert set(result.keys()) == {"subsystem", "contains"} ss = result["subsystem"] assert ss["id"] == v1a_seed["subsystem"].id assert ss["name"] == "Optics" assert ss["project"] == "p05-interferometer" assert ss["status"] == "active" contained = {c["name"] for c in result["contains"]} assert contained == {"Primary Mirror", "Diverger Lens"} for child in result["contains"]: assert set(child.keys()) == {"id", "entity_type", "name", "status"} assert child["entity_type"] == "component" assert child["status"] == "active" def test_subsystem_contents_returns_none_for_missing_id(tmp_data_dir): init_db() init_engineering_schema() assert subsystem_contents("no-such-id") is None def test_subsystem_contents_returns_none_when_entity_is_not_a_subsystem(v1a_seed): """Calling the subsystem-scoped query against a Component must return None, not the component dressed up as a subsystem.""" assert subsystem_contents(v1a_seed["primary"].id) is None def test_subsystem_contents_route_matches_spec(v1a_seed): """Spec invocation: GET /entities/Subsystem/?expand=contains. Verify the route is registered, the expand=contains path works, and unsupported expand values 400.""" client = TestClient(app) sid = v1a_seed["subsystem"].id r = client.get(f"/entities/Subsystem/{sid}?expand=contains") assert r.status_code == 200 body = r.json() assert body["subsystem"]["id"] == sid names = {c["name"] for c in body["contains"]} assert names == {"Primary Mirror", "Diverger Lens"} # Default expand is "contains" so omitting the param should also work r = client.get(f"/entities/Subsystem/{sid}") assert r.status_code == 200 # Unsupported expand value 400s with an informative message r = client.get(f"/entities/Subsystem/{sid}?expand=parents") assert r.status_code == 400 assert "expand" in r.json()["detail"].lower() # Missing/wrong-type subsystem 404s r = client.get(f"/entities/Subsystem/{v1a_seed['primary'].id}") assert r.status_code == 404 def test_subsystem_contents_v1_alias_is_present(): """The subsystem-scoped variant must also be reachable under /v1.""" client = TestClient(app) spec = client.get("/openapi.json").json() assert "/v1/entities/Subsystem/{subsystem_id}" in spec["paths"] # --------------------------------------------------------------------------- # V1-A acceptance — the four pillar queries against the single seed graph # --------------------------------------------------------------------------- def test_v1a_pillar_queries_run_end_to_end_against_single_seed(v1a_seed): """The V1-A acceptance test. All four pillar queries must report the expected shape against the same seed graph. If this test fails, V1-A is not done — full stop. (Per the plan: "If the four pillar queries don't work, stopping here is cheap.")""" project = "p05-interferometer" # Q-001 subsystem-scoped q1 = subsystem_contents(v1a_seed["subsystem"].id) assert q1 is not None assert {c["name"] for c in q1["contains"]} == {"Primary Mirror", "Diverger Lens"} # Q-005 — requirements satisfied by Primary Mirror q5 = requirements_for(v1a_seed["primary"].id) assert q5["count"] == 1 assert q5["requirements"][0]["name"] == "Surface figure < 25 nm RMS" # Q-006 (killer correctness) — orphan requirements in the project q6 = orphan_requirements(project) orphan_names = {r["name"] for r in q6["gaps"]} assert "Calibration repeatable to lambda/20" in orphan_names assert "Surface figure < 25 nm RMS" not in orphan_names # Q-017 — evidence chain for the supported ValidationClaim q17 = evidence_chain(v1a_seed["claim_supported"].id) via_supports = [edge for edge in q17["evidence_chain"] if edge["via"] == "supports"] assert any(edge["source_name"] == "FEA thermal sweep 2026-04" for edge in via_supports) # And the unsupported claim must have an empty supports chain q17_unsup = evidence_chain(v1a_seed["claim_unsupported"].id) assert not any(edge["via"] == "supports" for edge in q17_unsup["evidence_chain"])