363 lines
12 KiB
Python
363 lines
12 KiB
Python
|
"""V1-0 write-time invariant tests.
|
||
|
|
|
||
|
|
Covers the Engineering V1 completion plan Phase V1-0 acceptance:
|
||
|
|
- F-1 shared-header fields: extractor_version + canonical_home + hand_authored
|
||
|
|
land in the entities table with working defaults
|
||
|
|
- F-8 provenance enforcement: create_entity raises without source_refs
|
||
|
|
unless hand_authored=True
|
||
|
|
- F-5 synchronous conflict-detection hook on any active-entity write
|
||
|
|
(create_entity with status="active" + the pre-existing promote_entity
|
||
|
|
path); fail-open per conflict-model.md:256
|
||
|
|
- Q-3 "flag, never block": a conflict never 4xx-blocks the write
|
||
|
|
- Q-4 partial trust: get_entities scope_only filters candidates out
|
||
|
|
|
||
|
|
Plan: docs/plans/engineering-v1-completion-plan.md
|
||
|
|
Spec: docs/architecture/engineering-v1-acceptance.md
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from atocore.engineering.service import (
|
||
|
|
EXTRACTOR_VERSION,
|
||
|
|
create_entity,
|
||
|
|
create_relationship,
|
||
|
|
get_entities,
|
||
|
|
get_entity,
|
||
|
|
init_engineering_schema,
|
||
|
|
promote_entity,
|
||
|
|
supersede_entity,
|
||
|
|
)
|
||
|
|
from atocore.models.database import get_connection, init_db
|
||
|
|
|
||
|
|
|
||
|
|
# ---------- F-1: shared-header fields ----------
|
||
|
|
|
||
|
|
|
||
|
|
def test_entity_row_has_shared_header_fields(tmp_data_dir):
|
||
|
|
init_db()
|
||
|
|
init_engineering_schema()
|
||
|
|
with get_connection() as conn:
|
||
|
|
cols = {row["name"] for row in conn.execute("PRAGMA table_info(entities)").fetchall()}
|
||
|
|
assert "extractor_version" in cols
|
||
|
|
assert "canonical_home" in cols
|
||
|
|
assert "hand_authored" in cols
|
||
|
|
|
||
|
|
|
||
|
|
def test_created_entity_has_default_extractor_version_and_canonical_home(tmp_data_dir):
|
||
|
|
init_db()
|
||
|
|
init_engineering_schema()
|
||
|
|
e = create_entity(
|
||
|
|
entity_type="component",
|
||
|
|
name="Pivot Pin",
|
||
|
|
project="p04-gigabit",
|
||
|
|
source_refs=["test:fixture"],
|
||
|
|
)
|
||
|
|
assert e.extractor_version == EXTRACTOR_VERSION
|
||
|
|
assert e.canonical_home == "entity"
|
||
|
|
assert e.hand_authored is False
|
||
|
|
|
||
|
|
# round-trip through get_entity to confirm the row mapper returns
|
||
|
|
# the same values (not just the return-by-construct path)
|
||
|
|
got = get_entity(e.id)
|
||
|
|
assert got is not None
|
||
|
|
assert got.extractor_version == EXTRACTOR_VERSION
|
||
|
|
assert got.canonical_home == "entity"
|
||
|
|
assert got.hand_authored is False
|
||
|
|
|
||
|
|
|
||
|
|
def test_explicit_extractor_version_is_persisted(tmp_data_dir):
|
||
|
|
init_db()
|
||
|
|
init_engineering_schema()
|
||
|
|
e = create_entity(
|
||
|
|
entity_type="decision",
|
||
|
|
name="Pick GF-PTFE pads",
|
||
|
|
project="p04-gigabit",
|
||
|
|
source_refs=["interaction:abc"],
|
||
|
|
extractor_version="custom-v2.3",
|
||
|
|
)
|
||
|
|
got = get_entity(e.id)
|
||
|
|
assert got.extractor_version == "custom-v2.3"
|
||
|
|
|
||
|
|
|
||
|
|
# ---------- F-8: provenance enforcement ----------
|
||
|
|
|
||
|
|
|
||
|
|
def test_create_entity_without_provenance_raises(tmp_data_dir):
|
||
|
|
init_db()
|
||
|
|
init_engineering_schema()
|
||
|
|
with pytest.raises(ValueError, match="source_refs required"):
|
||
|
|
create_entity(
|
||
|
|
entity_type="component",
|
||
|
|
name="No Provenance",
|
||
|
|
project="p04-gigabit",
|
||
|
|
hand_authored=False, # explicit — bypasses the test-conftest auto-flag
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def test_create_entity_with_hand_authored_needs_no_source_refs(tmp_data_dir):
|
||
|
|
init_db()
|
||
|
|
init_engineering_schema()
|
||
|
|
e = create_entity(
|
||
|
|
entity_type="component",
|
||
|
|
name="Human Entry",
|
||
|
|
project="p04-gigabit",
|
||
|
|
hand_authored=True,
|
||
|
|
)
|
||
|
|
assert e.hand_authored is True
|
||
|
|
got = get_entity(e.id)
|
||
|
|
assert got.hand_authored is True
|
||
|
|
# source_refs stays empty — the hand_authored flag IS the provenance
|
||
|
|
assert got.source_refs == []
|
||
|
|
|
||
|
|
|
||
|
|
def test_create_entity_with_empty_source_refs_list_is_treated_as_missing(tmp_data_dir):
|
||
|
|
init_db()
|
||
|
|
init_engineering_schema()
|
||
|
|
with pytest.raises(ValueError, match="source_refs required"):
|
||
|
|
create_entity(
|
||
|
|
entity_type="component",
|
||
|
|
name="Empty Refs",
|
||
|
|
project="p04-gigabit",
|
||
|
|
source_refs=[],
|
||
|
|
hand_authored=False,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def test_promote_rejects_legacy_candidate_without_provenance(tmp_data_dir):
|
||
|
|
"""Regression (Codex V1-0 probe): candidate rows can exist in the DB
|
||
|
|
from before V1-0 enforcement (or from paths that bypass create_entity).
|
||
|
|
promote_entity must re-check the invariant and refuse to flip a
|
||
|
|
no-provenance candidate to active. Without this check, the active
|
||
|
|
store can leak F-8 violations in from legacy data."""
|
||
|
|
init_db()
|
||
|
|
init_engineering_schema()
|
||
|
|
|
||
|
|
# Simulate a pre-V1-0 candidate by inserting directly into the table,
|
||
|
|
# bypassing the service-layer invariant. Real legacy rows look exactly
|
||
|
|
# like this: empty source_refs, hand_authored=0.
|
||
|
|
import uuid as _uuid
|
||
|
|
entity_id = str(_uuid.uuid4())
|
||
|
|
with get_connection() as conn:
|
||
|
|
conn.execute(
|
||
|
|
"INSERT INTO entities (id, entity_type, name, project, "
|
||
|
|
"description, properties, status, confidence, source_refs, "
|
||
|
|
"extractor_version, canonical_home, hand_authored, "
|
||
|
|
"created_at, updated_at) "
|
||
|
|
"VALUES (?, 'component', 'Legacy Orphan', 'p04-gigabit', "
|
||
|
|
"'', '{}', 'candidate', 1.0, '[]', '', 'entity', 0, "
|
||
|
|
"CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)",
|
||
|
|
(entity_id,),
|
||
|
|
)
|
||
|
|
|
||
|
|
with pytest.raises(ValueError, match="source_refs required"):
|
||
|
|
promote_entity(entity_id)
|
||
|
|
|
||
|
|
# And the row stays a candidate — no half-transition.
|
||
|
|
got = get_entity(entity_id)
|
||
|
|
assert got is not None
|
||
|
|
assert got.status == "candidate"
|
||
|
|
|
||
|
|
|
||
|
|
def test_promote_accepts_candidate_flagged_hand_authored(tmp_data_dir):
|
||
|
|
"""The other side of the promote re-check: hand_authored=1 with
|
||
|
|
empty source_refs still lets promote succeed, matching
|
||
|
|
create_entity's symmetry."""
|
||
|
|
init_db()
|
||
|
|
init_engineering_schema()
|
||
|
|
|
||
|
|
import uuid as _uuid
|
||
|
|
entity_id = str(_uuid.uuid4())
|
||
|
|
with get_connection() as conn:
|
||
|
|
conn.execute(
|
||
|
|
"INSERT INTO entities (id, entity_type, name, project, "
|
||
|
|
"description, properties, status, confidence, source_refs, "
|
||
|
|
"extractor_version, canonical_home, hand_authored, "
|
||
|
|
"created_at, updated_at) "
|
||
|
|
"VALUES (?, 'component', 'Hand Authored Candidate', "
|
||
|
|
"'p04-gigabit', '', '{}', 'candidate', 1.0, '[]', '', "
|
||
|
|
"'entity', 1, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)",
|
||
|
|
(entity_id,),
|
||
|
|
)
|
||
|
|
|
||
|
|
assert promote_entity(entity_id) is True
|
||
|
|
assert get_entity(entity_id).status == "active"
|
||
|
|
|
||
|
|
|
||
|
|
# ---------- F-5: synchronous conflict-detection hook ----------
|
||
|
|
|
||
|
|
|
||
|
|
def test_active_create_runs_conflict_detection_hook(tmp_data_dir, monkeypatch):
|
||
|
|
"""status=active writes trigger detect_conflicts_for_entity."""
|
||
|
|
init_db()
|
||
|
|
init_engineering_schema()
|
||
|
|
|
||
|
|
called_with: list[str] = []
|
||
|
|
|
||
|
|
def _fake_detect(entity_id: str):
|
||
|
|
called_with.append(entity_id)
|
||
|
|
return []
|
||
|
|
|
||
|
|
import atocore.engineering.conflicts as conflicts_mod
|
||
|
|
monkeypatch.setattr(conflicts_mod, "detect_conflicts_for_entity", _fake_detect)
|
||
|
|
|
||
|
|
e = create_entity(
|
||
|
|
entity_type="component",
|
||
|
|
name="Active With Hook",
|
||
|
|
project="p04-gigabit",
|
||
|
|
source_refs=["test:hook"],
|
||
|
|
status="active",
|
||
|
|
)
|
||
|
|
|
||
|
|
assert called_with == [e.id]
|
||
|
|
|
||
|
|
|
||
|
|
def test_supersede_runs_conflict_detection_on_new_active(tmp_data_dir, monkeypatch):
|
||
|
|
"""Regression (Codex V1-0 probe): per plan's 'every active-entity
|
||
|
|
write path', supersede_entity must trigger synchronous conflict
|
||
|
|
detection. The subject is the `superseded_by` entity — the one
|
||
|
|
whose graph state just changed because a new `supersedes` edge was
|
||
|
|
rooted at it."""
|
||
|
|
init_db()
|
||
|
|
init_engineering_schema()
|
||
|
|
|
||
|
|
old = create_entity(
|
||
|
|
entity_type="component",
|
||
|
|
name="Old Pad",
|
||
|
|
project="p04-gigabit",
|
||
|
|
source_refs=["test:old"],
|
||
|
|
status="active",
|
||
|
|
)
|
||
|
|
new = create_entity(
|
||
|
|
entity_type="component",
|
||
|
|
name="New Pad",
|
||
|
|
project="p04-gigabit",
|
||
|
|
source_refs=["test:new"],
|
||
|
|
status="active",
|
||
|
|
)
|
||
|
|
|
||
|
|
called_with: list[str] = []
|
||
|
|
|
||
|
|
def _fake_detect(entity_id: str):
|
||
|
|
called_with.append(entity_id)
|
||
|
|
return []
|
||
|
|
|
||
|
|
import atocore.engineering.conflicts as conflicts_mod
|
||
|
|
monkeypatch.setattr(conflicts_mod, "detect_conflicts_for_entity", _fake_detect)
|
||
|
|
|
||
|
|
assert supersede_entity(old.id, superseded_by=new.id) is True
|
||
|
|
|
||
|
|
# The detector fires on the `superseded_by` entity — the one whose
|
||
|
|
# edges just grew a new `supersedes` relationship.
|
||
|
|
assert new.id in called_with
|
||
|
|
|
||
|
|
|
||
|
|
def test_supersede_hook_fails_open(tmp_data_dir, monkeypatch):
|
||
|
|
"""Supersede must survive a broken detector per Q-3 flag-never-block."""
|
||
|
|
init_db()
|
||
|
|
init_engineering_schema()
|
||
|
|
|
||
|
|
old = create_entity(
|
||
|
|
entity_type="component", name="Old2", project="p04-gigabit",
|
||
|
|
source_refs=["test:old"], status="active",
|
||
|
|
)
|
||
|
|
new = create_entity(
|
||
|
|
entity_type="component", name="New2", project="p04-gigabit",
|
||
|
|
source_refs=["test:new"], status="active",
|
||
|
|
)
|
||
|
|
|
||
|
|
def _boom(entity_id: str):
|
||
|
|
raise RuntimeError("synthetic detector failure")
|
||
|
|
|
||
|
|
import atocore.engineering.conflicts as conflicts_mod
|
||
|
|
monkeypatch.setattr(conflicts_mod, "detect_conflicts_for_entity", _boom)
|
||
|
|
|
||
|
|
# The supersede still succeeds despite the detector blowing up.
|
||
|
|
assert supersede_entity(old.id, superseded_by=new.id) is True
|
||
|
|
assert get_entity(old.id).status == "superseded"
|
||
|
|
|
||
|
|
|
||
|
|
def test_candidate_create_does_not_run_conflict_hook(tmp_data_dir, monkeypatch):
|
||
|
|
"""status=candidate writes do NOT trigger detection — the hook is
|
||
|
|
for active rows only, per V1-0 scope. Candidates are checked at
|
||
|
|
promote time."""
|
||
|
|
init_db()
|
||
|
|
init_engineering_schema()
|
||
|
|
|
||
|
|
called: list[str] = []
|
||
|
|
|
||
|
|
def _fake_detect(entity_id: str):
|
||
|
|
called.append(entity_id)
|
||
|
|
return []
|
||
|
|
|
||
|
|
import atocore.engineering.conflicts as conflicts_mod
|
||
|
|
monkeypatch.setattr(conflicts_mod, "detect_conflicts_for_entity", _fake_detect)
|
||
|
|
|
||
|
|
create_entity(
|
||
|
|
entity_type="component",
|
||
|
|
name="Candidate No Hook",
|
||
|
|
project="p04-gigabit",
|
||
|
|
source_refs=["test:cand"],
|
||
|
|
status="candidate",
|
||
|
|
)
|
||
|
|
|
||
|
|
assert called == []
|
||
|
|
|
||
|
|
|
||
|
|
# ---------- Q-3: flag, never block ----------
|
||
|
|
|
||
|
|
|
||
|
|
def test_conflict_detector_failure_does_not_block_write(tmp_data_dir, monkeypatch):
|
||
|
|
"""Per conflict-model.md:256: detection errors must not fail the
|
||
|
|
write. The entity is still created; only a warning is logged."""
|
||
|
|
init_db()
|
||
|
|
init_engineering_schema()
|
||
|
|
|
||
|
|
def _boom(entity_id: str):
|
||
|
|
raise RuntimeError("synthetic detector failure")
|
||
|
|
|
||
|
|
import atocore.engineering.conflicts as conflicts_mod
|
||
|
|
monkeypatch.setattr(conflicts_mod, "detect_conflicts_for_entity", _boom)
|
||
|
|
|
||
|
|
# The write still succeeds — no exception propagates.
|
||
|
|
e = create_entity(
|
||
|
|
entity_type="component",
|
||
|
|
name="Hook Fails Open",
|
||
|
|
project="p04-gigabit",
|
||
|
|
source_refs=["test:failopen"],
|
||
|
|
status="active",
|
||
|
|
)
|
||
|
|
assert get_entity(e.id) is not None
|
||
|
|
|
||
|
|
|
||
|
|
# ---------- Q-4 (partial): trust-hierarchy — scope_only filters candidates ----------
|
||
|
|
|
||
|
|
|
||
|
|
def test_scope_only_active_does_not_return_candidates(tmp_data_dir):
|
||
|
|
"""V1-0 partial Q-4: active-scoped listing never returns candidates.
|
||
|
|
Full trust-hierarchy coverage (no-auto-project-state, etc.) ships in
|
||
|
|
V1-E per plan."""
|
||
|
|
init_db()
|
||
|
|
init_engineering_schema()
|
||
|
|
|
||
|
|
active = create_entity(
|
||
|
|
entity_type="component",
|
||
|
|
name="Active Alpha",
|
||
|
|
project="p04-gigabit",
|
||
|
|
source_refs=["test:alpha"],
|
||
|
|
status="active",
|
||
|
|
)
|
||
|
|
candidate = create_entity(
|
||
|
|
entity_type="component",
|
||
|
|
name="Candidate Beta",
|
||
|
|
project="p04-gigabit",
|
||
|
|
source_refs=["test:beta"],
|
||
|
|
status="candidate",
|
||
|
|
)
|
||
|
|
|
||
|
|
listed = get_entities(project="p04-gigabit", status="active", scope_only=True)
|
||
|
|
ids = {e.id for e in listed}
|
||
|
|
assert active.id in ids
|
||
|
|
assert candidate.id not in ids
|