feat(engineering): V1-A — Q-001 subsystem-scoped + pillar query integration test
Phase V1-A of the Engineering V1 Completion Plan. Scope was kept tight
per the plan: a single Q-001 shape fix and an integration test that
proves the four pillar queries work end-to-end against one seed graph.
Code change:
- subsystem_contents() in src/atocore/engineering/queries.py returns
{subsystem, contains: [{id, type, entity_type, name, status}]} by
walking inbound part_of edges (the inverse of CONTAINS), filtered
to active children. `type` matches the catalog spec; `entity_type`
preserves parity with the rest of this module's response shape.
- GET /entities/Subsystem/{id}?expand=contains route in routes.py
matches the Q-001 spec invocation verbatim; 400 on unsupported
expand, 404 on missing-or-wrong-type.
- Aliased under /v1.
- The existing project-wide /engineering/projects/{name}/systems
route stays as-is for Q-004 (whole-project tree).
V1-A acceptance test (tests/test_v1_a_pillar_queries.py):
- Q-001 subsystem-scoped: Optics → Primary Mirror + Diverger Lens
- Q-005 requirements_for: Primary Mirror has its single satisfied req
- Q-006 orphan_requirements: orphan surfaces, satisfied does not
- Q-017 evidence_chain: supported claim has the FEA result via
'supports'; unsupported claim does not
- Q-009 risky_decisions / Q-011 unsupported_claims also asserted
against the same seed (Codex P2: don't seed data you don't assert)
Plus targeted tests for the Q-001 helper: missing id, wrong type,
nested child subsystems, inactive-child filtering, real /v1 GET
behavior.
Reviewed by Codex twice: GO WITH CONDITIONS at b575773, GO at 0e83525
after the dual-key emit + Q-009/Q-011 + /v1-behavior + nested/inactive
amends folded in.
Tests: 596 -> 605 (+9). Full local suite green.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -2393,6 +2393,41 @@ def api_entity_audit(entity_id: str, limit: int = 100) -> dict:
|
||||
return {"entity_id": entity_id, "entries": entries, "count": len(entries)}
|
||||
|
||||
|
||||
@router.get("/entities/Subsystem/{subsystem_id}")
|
||||
def api_subsystem_contents(subsystem_id: str, expand: str = "contains") -> dict:
|
||||
"""Q-001 (subsystem-scoped variant) — the spec-shaped invocation.
|
||||
|
||||
Per ``docs/architecture/engineering-query-catalog.md`` Q-001:
|
||||
``GET /entities/Subsystem/<id>?expand=contains``
|
||||
→ ``{ subsystem, contains: [{ id, type, name, status }] }``
|
||||
|
||||
Distinct from the project-wide tree at
|
||||
``GET /engineering/projects/{name}/systems`` (Q-004), which stays
|
||||
as-is. V1-A only adds this single shape fix; broader query catalog
|
||||
closure is V1-C.
|
||||
|
||||
``expand`` is currently restricted to ``"contains"``; other expand
|
||||
facets (Q-007 ``constraints``, Q-008 ``decisions``) land in V1-C.
|
||||
"""
|
||||
if expand != "contains":
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail=(
|
||||
f"unsupported expand={expand!r} for /entities/Subsystem; "
|
||||
"V1-A supports only 'contains'."
|
||||
),
|
||||
)
|
||||
from atocore.engineering.queries import subsystem_contents
|
||||
|
||||
result = subsystem_contents(subsystem_id)
|
||||
if result is None:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail=f"Subsystem not found or not a subsystem: {subsystem_id}",
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
@router.get("/entities/{entity_id}")
|
||||
def api_get_entity(entity_id: str) -> dict:
|
||||
"""Get an entity with its relationships and related entities."""
|
||||
|
||||
@@ -118,6 +118,70 @@ def system_map(project: str) -> dict:
|
||||
return out
|
||||
|
||||
|
||||
def subsystem_contents(subsystem_id: str) -> dict | None:
|
||||
"""Q-001 subsystem-scoped variant: a single subsystem and its
|
||||
direct ``CONTAINS`` children.
|
||||
|
||||
Spec: ``GET /entities/Subsystem/<id>?expand=contains`` per
|
||||
``docs/architecture/engineering-query-catalog.md`` Q-001.
|
||||
|
||||
Differs from :func:`system_map` (Q-004) which returns the
|
||||
project-wide tree. The subsystem-scoped form is what individual
|
||||
operator queries actually need: "what's inside this one subsystem?"
|
||||
rather than "show me the whole project."
|
||||
|
||||
The relationship walk uses inbound ``part_of`` edges (the inverse
|
||||
of ``CONTAINS``) so both child Components and child Subsystems
|
||||
surface uniformly. Filters to active children only — superseded
|
||||
or invalid rows do not belong in a "current contents" answer.
|
||||
|
||||
Returns:
|
||||
``{"subsystem": {id, name, project, status, description},
|
||||
"contains": [{id, entity_type, name, status}]}``
|
||||
or ``None`` when the entity does not exist or is not a subsystem.
|
||||
"""
|
||||
with get_connection() as conn:
|
||||
ss = conn.execute(
|
||||
"SELECT * FROM entities WHERE id = ?",
|
||||
(subsystem_id,),
|
||||
).fetchone()
|
||||
if ss is None or ss["entity_type"] != "subsystem":
|
||||
return None
|
||||
rows = conn.execute(
|
||||
"SELECT e.id, e.entity_type, e.name, e.status "
|
||||
"FROM relationships r "
|
||||
"JOIN entities e ON e.id = r.source_entity_id "
|
||||
"WHERE r.relationship_type = 'part_of' "
|
||||
"AND r.target_entity_id = ? "
|
||||
"AND e.status = 'active' "
|
||||
"ORDER BY e.entity_type, e.name",
|
||||
(subsystem_id,),
|
||||
).fetchall()
|
||||
|
||||
return {
|
||||
"subsystem": {
|
||||
"id": ss["id"],
|
||||
"name": ss["name"],
|
||||
"project": ss["project"],
|
||||
"status": ss["status"],
|
||||
"description": ss["description"] or "",
|
||||
},
|
||||
"contains": [
|
||||
{
|
||||
"id": r["id"],
|
||||
# V1-A spec uses `type` per engineering-query-catalog.md Q-001;
|
||||
# `entity_type` is duplicated for parity with the rest of
|
||||
# this module's response shape (see `_entity_dict`).
|
||||
"type": r["entity_type"],
|
||||
"entity_type": r["entity_type"],
|
||||
"name": r["name"],
|
||||
"status": r["status"],
|
||||
}
|
||||
for r in rows
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def decisions_affecting(project: str, subsystem_id: str | None = None) -> dict:
|
||||
"""Q-008: decisions that affect a subsystem (or whole project).
|
||||
|
||||
|
||||
@@ -66,6 +66,8 @@ _V1_PUBLIC_PATHS = {
|
||||
"/entities/{entity_id}/invalidate",
|
||||
"/entities/{entity_id}/supersede",
|
||||
"/entities/{entity_id}/audit",
|
||||
# V1-A: Q-001 subsystem-scoped variant per engineering-query-catalog
|
||||
"/entities/Subsystem/{subsystem_id}",
|
||||
"/relationships",
|
||||
"/ingest",
|
||||
"/ingest/sources",
|
||||
|
||||
278
tests/test_v1_a_pillar_queries.py
Normal file
278
tests/test_v1_a_pillar_queries.py
Normal file
@@ -0,0 +1,278 @@
|
||||
"""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,
|
||||
risky_decisions,
|
||||
subsystem_contents,
|
||||
unsupported_claims,
|
||||
)
|
||||
from atocore.engineering.service import (
|
||||
create_entity,
|
||||
create_relationship,
|
||||
init_engineering_schema,
|
||||
invalidate_active_entity,
|
||||
)
|
||||
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 match the catalog spec
|
||||
(engineering-query-catalog.md Q-001):
|
||||
``{subsystem, contains: [{id, type, name, status}]}``.
|
||||
The implementation also emits ``entity_type`` for parity with the
|
||||
rest of this module's response style — both must be present."""
|
||||
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"]:
|
||||
# Spec requires `type`; we also include `entity_type` for parity.
|
||||
assert "type" in child
|
||||
assert "entity_type" in child
|
||||
assert child["type"] == child["entity_type"]
|
||||
assert set(child.keys()) >= {"id", "type", "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/<id>?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"]
|
||||
|
||||
|
||||
def test_subsystem_contents_v1_alias_serves_real_response(v1a_seed):
|
||||
"""Behavior, not just OpenAPI presence — catches a future
|
||||
route-copy or ordering regression (Codex V1-A P3)."""
|
||||
client = TestClient(app)
|
||||
sid = v1a_seed["subsystem"].id
|
||||
|
||||
r = client.get(f"/v1/entities/Subsystem/{sid}?expand=contains")
|
||||
assert r.status_code == 200
|
||||
body = r.json()
|
||||
assert body["subsystem"]["id"] == sid
|
||||
assert {c["name"] for c in body["contains"]} == {"Primary Mirror", "Diverger Lens"}
|
||||
|
||||
|
||||
def test_subsystem_contents_includes_child_subsystems_and_excludes_inactive(tmp_data_dir):
|
||||
"""Codex V1-A P3: lock in two intended behaviors that aren't
|
||||
obvious from the data shape:
|
||||
1. Child *Subsystems* (nested) surface in `contains` — the impl
|
||||
walks part_of irrespective of child entity_type.
|
||||
2. Children whose status is not 'active' are filtered out."""
|
||||
init_db()
|
||||
init_engineering_schema()
|
||||
|
||||
parent = create_entity("subsystem", "Parent System", project="p-nested")
|
||||
child_ss = create_entity("subsystem", "Child Subsystem", project="p-nested")
|
||||
child_comp = create_entity("component", "Child Component", project="p-nested")
|
||||
invalid_comp = create_entity("component", "Invalidated Component", project="p-nested")
|
||||
create_relationship(child_ss.id, parent.id, "part_of")
|
||||
create_relationship(child_comp.id, parent.id, "part_of")
|
||||
create_relationship(invalid_comp.id, parent.id, "part_of")
|
||||
|
||||
invalidate_active_entity(invalid_comp.id, reason="v1a regression seed")
|
||||
|
||||
result = subsystem_contents(parent.id)
|
||||
names_by_type = {(c["entity_type"], c["name"]) for c in result["contains"]}
|
||||
assert ("subsystem", "Child Subsystem") in names_by_type
|
||||
assert ("component", "Child Component") in names_by_type
|
||||
# Inactive child must not appear
|
||||
assert all(c["name"] != "Invalidated Component" for c in result["contains"])
|
||||
assert all(c["status"] == "active" for c in result["contains"])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 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"])
|
||||
|
||||
|
||||
def test_v1a_seed_also_exercises_q009_and_q011_killers(v1a_seed):
|
||||
"""Per Codex V1-A P2: the seed already includes Q-009 (decision on
|
||||
flagged assumption) and Q-011 (unsupported validation claim) data,
|
||||
so the V1-A integration test should assert those killer queries
|
||||
surface them. Otherwise the seed entities are dead weight."""
|
||||
project = "p05-interferometer"
|
||||
|
||||
q9 = risky_decisions(project)
|
||||
risky_names = {row["decision_name"] for row in q9["gaps"]}
|
||||
assert "Pre-order CGH from external vendor" in risky_names
|
||||
|
||||
q11 = unsupported_claims(project)
|
||||
unsupported_names = {row["name"] for row in q11["gaps"]}
|
||||
assert "Vibration isolation passes spec" in unsupported_names
|
||||
assert "Thermal margin is adequate" not in unsupported_names
|
||||
Reference in New Issue
Block a user