From 4e2823ca662454717610ecf95a020a6280d59878 Mon Sep 17 00:00:00 2001 From: Candice0313 Date: Tue, 31 Mar 2026 16:25:13 -0500 Subject: [PATCH] feat(opm): add validation, extraction, schemas, and tests --- web/app/db.py | 59 ++++++++ web/app/main.py | 3 +- web/app/routers/opm.py | 54 +++++++ web/app/schemas/__init__.py | 0 web/app/schemas/opm.py | 64 +++++++++ web/app/services/opm_extract.py | 17 +++ web/app/services/opm_validate.py | 45 ++++++ web/data/app.db | Bin 16384 -> 20480 bytes web/tests/test_opm.py | 214 ++++++++++++++++++++++++++++ web/tests/test_opm_validate.py | 235 +++++++++++++++++++++++++++++++ 10 files changed, 690 insertions(+), 1 deletion(-) create mode 100644 web/app/routers/opm.py create mode 100644 web/app/schemas/__init__.py create mode 100644 web/app/schemas/opm.py create mode 100644 web/app/services/opm_extract.py create mode 100644 web/app/services/opm_validate.py create mode 100644 web/tests/test_opm.py create mode 100644 web/tests/test_opm_validate.py diff --git a/web/app/db.py b/web/app/db.py index 40f2385..8a6e16f 100644 --- a/web/app/db.py +++ b/web/app/db.py @@ -46,6 +46,16 @@ def init_db() -> None: ); """ ) + cursor.execute( + """ + CREATE TABLE IF NOT EXISTS opm_diagrams ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + note_id INTEGER, + payload TEXT NOT NULL, + created_at TEXT DEFAULT (datetime('now')) + ); + """ + ) connection.commit() @@ -114,3 +124,52 @@ def mark_action_item_done(action_item_id: int, done: bool) -> None: connection.commit() +import json as _json + + +def insert_opm_diagram(payload: dict, note_id: Optional[int] = None) -> int: + with get_connection() as connection: + cursor = connection.cursor() + cursor.execute( + "INSERT INTO opm_diagrams (note_id, payload) VALUES (?, ?)", + (note_id, _json.dumps(payload)), + ) + connection.commit() + return int(cursor.lastrowid) + + +def list_opm_diagrams() -> list[dict]: + with get_connection() as connection: + cursor = connection.cursor() + cursor.execute( + "SELECT id, note_id, payload, created_at FROM opm_diagrams ORDER BY id DESC" + ) + rows = cursor.fetchall() + return [ + { + "id": r["id"], + "note_id": r["note_id"], + "created_at": r["created_at"], + "diagram": _json.loads(r["payload"]), + } + for r in rows + ] + + +def get_opm_diagram(diagram_id: int) -> Optional[dict]: + with get_connection() as connection: + cursor = connection.cursor() + cursor.execute( + "SELECT id, note_id, payload, created_at FROM opm_diagrams WHERE id = ?", + (diagram_id,), + ) + row = cursor.fetchone() + if row is None: + return None + return { + "id": row["id"], + "note_id": row["note_id"], + "created_at": row["created_at"], + "diagram": _json.loads(row["payload"]), + } + diff --git a/web/app/main.py b/web/app/main.py index 6a3315e..92872e9 100644 --- a/web/app/main.py +++ b/web/app/main.py @@ -8,7 +8,7 @@ from fastapi.staticfiles import StaticFiles from .db import init_db -from .routers import action_items, notes +from .routers import action_items, notes, opm from . import db init_db() @@ -24,6 +24,7 @@ def index() -> str: app.include_router(notes.router) app.include_router(action_items.router) +app.include_router(opm.router) static_dir = Path(__file__).resolve().parents[1] / "frontend" diff --git a/web/app/routers/opm.py b/web/app/routers/opm.py new file mode 100644 index 0000000..e38fbc5 --- /dev/null +++ b/web/app/routers/opm.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +from typing import Any, Dict, Optional + +from fastapi import APIRouter, HTTPException +from pydantic import ValidationError + +from .. import db +from ..services.opm_extract import extract_opm_diagram +from ..services.opm_validate import validate_diagram + + +router = APIRouter(prefix="/opm", tags=["opm"]) + + +@router.post("/extract") +def extract(payload: Dict[str, Any]) -> Dict[str, Any]: + text = str(payload.get("text", "")).strip() + if not text: + raise HTTPException(status_code=400, detail="text is required") + + note_id: Optional[int] = None + if payload.get("save_note"): + note_id = db.insert_note(text) + + raw_dict = extract_opm_diagram(text) + + try: + validated = validate_diagram(raw_dict) + except (ValidationError, ValueError) as exc: + raise HTTPException( + status_code=422, + detail={ + "error": "opm_extraction_failed", + "stage": "validation", + "detail": str(exc), + }, + ) + + diagram_id = db.insert_opm_diagram(validated.model_dump(), note_id=note_id) + return {"note_id": note_id, "diagram_id": diagram_id, "diagram": validated.model_dump()} + + +@router.get("") +def list_all() -> Dict[str, Any]: + return {"diagrams": db.list_opm_diagrams()} + + +@router.get("/{diagram_id}") +def get_one(diagram_id: int) -> Dict[str, Any]: + row = db.get_opm_diagram(diagram_id) + if row is None: + raise HTTPException(status_code=404, detail="diagram not found") + return row diff --git a/web/app/schemas/__init__.py b/web/app/schemas/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/web/app/schemas/opm.py b/web/app/schemas/opm.py new file mode 100644 index 0000000..92d2574 --- /dev/null +++ b/web/app/schemas/opm.py @@ -0,0 +1,64 @@ +from __future__ import annotations + +from enum import Enum +from typing import List + +from pydantic import BaseModel, Field, model_validator + + +class NodeKind(str, Enum): + object = "object" + process = "process" + state = "state" + + +class Relation(str, Enum): + agent = "agent" + instrument = "instrument" + consumption = "consumption" + result = "result" + effect = "effect" + aggregation = "aggregation" + specialization = "specialization" + characterization = "characterization" + + +class OpmNode(BaseModel): + id: str = Field(..., pattern=r'^[a-z0-9]+(-[a-z0-9]+)*$') + kind: NodeKind + label: str = Field(..., min_length=1, max_length=80) + + +class OpmLink(BaseModel): + id: str = Field(..., min_length=1) + source: str + target: str + relation: Relation + + +class OpmDiagram(BaseModel): + version: str = Field(default="1.0") + nodes: List[OpmNode] + links: List[OpmLink] + + @model_validator(mode="after") + def check_graph_integrity(self) -> "OpmDiagram": + node_ids = {n.id for n in self.nodes} + + # Rule: node IDs must be unique + if len(node_ids) != len(self.nodes): + raise ValueError("Duplicate node IDs detected") + + # Rule: link IDs must be unique + link_ids = [lk.id for lk in self.links] + if len(set(link_ids)) != len(link_ids): + raise ValueError("Duplicate link IDs detected") + + # Rule: link endpoints must reference existing nodes + for lk in self.links: + if lk.source not in node_ids: + raise ValueError(f"Link '{lk.id}' source '{lk.source}' not in nodes") + if lk.target not in node_ids: + raise ValueError(f"Link '{lk.id}' target '{lk.target}' not in nodes") + + return self diff --git a/web/app/services/opm_extract.py b/web/app/services/opm_extract.py new file mode 100644 index 0000000..cb0331b --- /dev/null +++ b/web/app/services/opm_extract.py @@ -0,0 +1,17 @@ +from __future__ import annotations + + +def extract_opm_diagram(text: str) -> dict: + """ + Phase 1 stub implementation. + + Ignores input text and returns a deterministic hardcoded diagram. + This function will be replaced by an LLM-backed implementation in Phase 2. + """ + return { + "version": "1.0", + "nodes": [ + {"id": "example-object", "kind": "object", "label": "example"} + ], + "links": [], + } diff --git a/web/app/services/opm_validate.py b/web/app/services/opm_validate.py new file mode 100644 index 0000000..1a204b6 --- /dev/null +++ b/web/app/services/opm_validate.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +import logging + +from pydantic import ValidationError # noqa: F401 — re-exported for callers + +from ..schemas.opm import OpmDiagram + +logger = logging.getLogger(__name__) + + +def validate_diagram(data: dict) -> OpmDiagram: + """ + Validate a candidate diagram dict. + + Returns: + OpmDiagram on success + + Raises: + pydantic.ValidationError on schema/field failures + ValueError on graph-integrity failures + """ + diagram = OpmDiagram.model_validate(data) + _warn_semantic(diagram) + return diagram + + +def _warn_semantic(diagram: OpmDiagram) -> None: + # Self-loops + for lk in diagram.links: + if lk.source == lk.target: + logger.warning("Self-loop on node '%s' via link '%s'", lk.source, lk.id) + + # Duplicate (source, target, relation) triples + seen: set[tuple] = set() + for lk in diagram.links: + key = (lk.source, lk.target, lk.relation) + if key in seen: + logger.warning( + "Duplicate relation %s on (%s → %s)", + lk.relation, + lk.source, + lk.target, + ) + seen.add(key) diff --git a/web/data/app.db b/web/data/app.db index caa5a12a3f35896620d9bece24c3c1c5f2a112b9..d014ccfbb0c92310318692239ec6a49859b21575 100644 GIT binary patch literal 20480 zcmeI3-)~lE=|b^45UT8ab_Wj%iYbiifFs}tI0BA9|ns1tlfOC&Gz>3d#lxYvYD3%ZiIh`w@tR? zTg_`bw_0ts?&GkLp;&(@Qb#XsY@ATA@t{|+Ab$UX1;ITdRdk5=V49E$*fw^2Dj9l@qj2om`K2D5|(}e;Cq#qc2$g!6#T#v(DK5Ap`m6Ko(7%{SSv)set%6@v*- z$uzNUVl&r3aaVaMu8NqBg=eXZdsUqe1q$jQ>op(w#Bfyz45cB37(gbh8;E}LE|Rew9_LWR2gS+zzuyyXM=gl zaJhmkhTziT2?Vxy)Jr*bZrG@@^;YYK-oPk*0UL&CU}PK!1fQW1+3iTeYrfQ&wr7~f zV{WNXh&&Ypykn6J$TYbHXr{o&4A*zE4#0>re~6L^w~A;W6BUu0H(493Y=}+9np)Dp zkdHpG)YCo>0(F#GR@}1&fd$ag2gwAxSr~u>8iUlQ06h*#B1|I1fCtrYFkc=4oa-YJ zo*sE#OQ6US2%zC*cpGf=IAodfd`c4}K)NH$k$`0~f>kEM816-EDG`XbRWguBJ7CFY zFDPvj#gSy>A@1El&OA<-QI3+9T=Y1>AW;KYh;%nFCUmNr}W!j*#p}=KW|=VxCzvEyz?IgXG9B?8-!&T8#SS zk;Bk0Wx!{j{5dOu$i{)}PR}|39@*T>GY4Iq%BfK{7|JBNFce;&N2s)u zl`DffSwG0B<8^_GLv9RsgDTWm&|1#jgw zn-aN}Wzs~xQGKKfuzrYSG;SINf@@S&)}z<1EXI0;b%h70T3u^bt0OD#6ir(sYU83h zuHJg=$1MCYN*zmMYa2cgQHmjG)tQ)5UZUX!sXYa+qKY6I8LrK(`qpc;dZSjq!Y*Ce zuD`xruRrq2SZIIl7FxH^x`oy)v~HpOz82axmJPSc((Bjfe3iYj{c3~W|5sN2Q(F0I z<#Rl^%@J?}905nb5pVz zSuJNbF3|h`^74EMKW=jb905nb5pVT|NP8UY2~BkFPEdGFP1uIKRx^InSY=8&Emfnf4T6_g{}F2%>QEU z?{hz!{mbm~%%9NIZH|B=;0Qb?1dc1`XG^7H-uQTp?x4qA>1*3;?b2p_t;*ISe0M_Q z`THNQNuQ37Zm%ER6irR=WfyxUyG@#o;HwfMm}tCq{{uRT@5R`zKDeLX?cAudt;Y7{ zH@3GL$CaNJLd8!Cq2k47uVU-C@QV11Og-}s0fQqjH5!`SF delta 81 zcmZozz}V2hI6+#Fm4ShQ1&CpQX`+s?Fe`&z?@wN!5HoKf1HUfcZQjDof(rk6HW%_v c7UE#yf5O23mH!D)=pg^Z3F4cd=*tKI0C`grZU6uP diff --git a/web/tests/test_opm.py b/web/tests/test_opm.py new file mode 100644 index 0000000..49e4e7f --- /dev/null +++ b/web/tests/test_opm.py @@ -0,0 +1,214 @@ +from __future__ import annotations + +import json +import sys +from pathlib import Path +from typing import Generator +from unittest.mock import MagicMock, patch + +import pytest +from fastapi.testclient import TestClient + +from ..app.services.opm_extract import extract_opm_diagram + + +# --------------------------------------------------------------------------- +# Unit tests: stub extractor +# --------------------------------------------------------------------------- + + +def test_stub_returns_dict(): + result = extract_opm_diagram("any text") + assert isinstance(result, dict) + + +def test_stub_version_field(): + result = extract_opm_diagram("any text") + assert result["version"] == "1.0" + + +def test_stub_has_nodes_and_links(): + result = extract_opm_diagram("any text") + assert "nodes" in result + assert "links" in result + + +def test_stub_is_deterministic(): + assert extract_opm_diagram("foo") == extract_opm_diagram("bar") + + +# --------------------------------------------------------------------------- +# Unit tests: JSON round-trip +# --------------------------------------------------------------------------- + + +def test_payload_round_trip(): + original = { + "version": "1.0", + "nodes": [{"id": "example-object", "kind": "object", "label": "example"}], + "links": [], + } + encoded = json.dumps(original) + decoded = json.loads(encoded) + assert decoded == original + + +def test_version_preserved_in_round_trip(): + payload = {"version": "1.0", "nodes": [], "links": []} + assert json.loads(json.dumps(payload))["version"] == "1.0" + + +# --------------------------------------------------------------------------- +# Fixtures: in-memory DB for isolation +# --------------------------------------------------------------------------- + + +@pytest.fixture() +def tmp_db(tmp_path: Path) -> Generator[Path, None, None]: + db_file = tmp_path / "test.db" + with patch("web.app.db.DB_PATH", db_file): + from web.app import db as db_module + db_module.init_db() + yield db_file + + +# --------------------------------------------------------------------------- +# Unit tests: DB helpers +# --------------------------------------------------------------------------- + + +def test_insert_opm_diagram_returns_int(tmp_db: Path): + from web.app import db as db_module + payload = {"version": "1.0", "nodes": [], "links": []} + diagram_id = db_module.insert_opm_diagram(payload) + assert isinstance(diagram_id, int) + + +def test_insert_opm_diagram_multiple_distinct_ids(tmp_db: Path): + from web.app import db as db_module + payload = {"version": "1.0", "nodes": [], "links": []} + id1 = db_module.insert_opm_diagram(payload) + id2 = db_module.insert_opm_diagram(payload) + assert id1 != id2 + + +def test_get_opm_diagram_returns_parsed_dict(tmp_db: Path): + from web.app import db as db_module + payload = {"version": "1.0", "nodes": [{"id": "x"}], "links": []} + diagram_id = db_module.insert_opm_diagram(payload) + row = db_module.get_opm_diagram(diagram_id) + assert row is not None + assert isinstance(row["diagram"], dict) + assert row["diagram"]["version"] == "1.0" + + +def test_get_opm_diagram_not_found_returns_none(tmp_db: Path): + from web.app import db as db_module + assert db_module.get_opm_diagram(99999) is None + + +def test_note_id_null_when_not_provided(tmp_db: Path): + from web.app import db as db_module + payload = {"version": "1.0", "nodes": [], "links": []} + diagram_id = db_module.insert_opm_diagram(payload) + row = db_module.get_opm_diagram(diagram_id) + assert row is not None + assert row["note_id"] is None + + +def test_list_opm_diagrams_returns_all(tmp_db: Path): + from web.app import db as db_module + payload = {"version": "1.0", "nodes": [], "links": []} + db_module.insert_opm_diagram(payload) + db_module.insert_opm_diagram(payload) + diagrams = db_module.list_opm_diagrams() + assert len(diagrams) == 2 + + +def test_stored_payload_matches_inserted(tmp_db: Path): + from web.app import db as db_module + payload = {"version": "1.0", "nodes": [{"id": "n1"}], "links": []} + diagram_id = db_module.insert_opm_diagram(payload) + row = db_module.get_opm_diagram(diagram_id) + assert row is not None + assert row["diagram"] == payload + + +# --------------------------------------------------------------------------- +# Integration tests: API +# --------------------------------------------------------------------------- + + +@pytest.fixture() +def client(tmp_db: Path) -> Generator[TestClient, None, None]: + # ollama is not installed in this environment; stub it so the app can be imported + sys.modules.setdefault("ollama", MagicMock()) + # Remove cached app import so the patched DB_PATH takes effect + for mod in list(sys.modules): + if mod.startswith("web.app"): + del sys.modules[mod] + from web.app.main import app + with TestClient(app) as c: + yield c + + +def test_post_extract_inserts_row(client: TestClient, tmp_db: Path): + from web.app import db as db_module + response = client.post("/opm/extract", json={"text": "some text", "save_note": False}) + assert response.status_code == 200 + data = response.json() + assert "diagram_id" in data + row = db_module.get_opm_diagram(data["diagram_id"]) + assert row is not None + + +def test_post_extract_response_shape(client: TestClient): + response = client.post("/opm/extract", json={"text": "some text", "save_note": False}) + assert response.status_code == 200 + data = response.json() + assert "note_id" in data + assert "diagram_id" in data + assert "diagram" in data + diagram = data["diagram"] + assert "version" in diagram + assert "nodes" in diagram + assert "links" in diagram + + +def test_post_extract_note_id_null_when_save_note_false(client: TestClient): + response = client.post("/opm/extract", json={"text": "some text", "save_note": False}) + assert response.status_code == 200 + assert response.json()["note_id"] is None + + +def test_get_by_id_returns_same_payload(client: TestClient): + post_resp = client.post("/opm/extract", json={"text": "some text", "save_note": False}) + diagram_id = post_resp.json()["diagram_id"] + inserted_diagram = post_resp.json()["diagram"] + + get_resp = client.get(f"/opm/{diagram_id}") + assert get_resp.status_code == 200 + assert get_resp.json()["diagram"] == inserted_diagram + + +def test_get_by_id_404_when_missing(client: TestClient): + response = client.get("/opm/99999") + assert response.status_code == 404 + + +def test_get_list_returns_diagrams_key(client: TestClient): + client.post("/opm/extract", json={"text": "a", "save_note": False}) + response = client.get("/opm") + assert response.status_code == 200 + assert "diagrams" in response.json() + + +def test_multiple_inserts_distinct_ids(client: TestClient): + r1 = client.post("/opm/extract", json={"text": "a", "save_note": False}) + r2 = client.post("/opm/extract", json={"text": "b", "save_note": False}) + assert r1.json()["diagram_id"] != r2.json()["diagram_id"] + + +def test_post_extract_missing_text_returns_400(client: TestClient): + response = client.post("/opm/extract", json={"save_note": False}) + assert response.status_code == 400 diff --git a/web/tests/test_opm_validate.py b/web/tests/test_opm_validate.py new file mode 100644 index 0000000..f617a3c --- /dev/null +++ b/web/tests/test_opm_validate.py @@ -0,0 +1,235 @@ +from __future__ import annotations + +import logging +import sys +from pathlib import Path +from typing import Generator +from unittest.mock import MagicMock, patch + +import pytest +from pydantic import ValidationError + +from web.app.services.opm_validate import validate_diagram +from web.app.schemas.opm import OpmDiagram + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +VALID_DIAGRAM: dict = { + "version": "1.0", + "nodes": [ + {"id": "farmer", "kind": "object", "label": "Farmer"}, + {"id": "grow", "kind": "process", "label": "Grow"}, + {"id": "crop", "kind": "object", "label": "Crop"}, + ], + "links": [ + {"id": "l1", "source": "farmer", "target": "grow", "relation": "agent"}, + {"id": "l2", "source": "grow", "target": "crop", "relation": "result"}, + ], +} + + +def _diagram(**overrides) -> dict: + """Return a copy of VALID_DIAGRAM with top-level fields overridden.""" + return {**VALID_DIAGRAM, **overrides} + + +# --------------------------------------------------------------------------- +# Unit tests: valid diagram +# --------------------------------------------------------------------------- + + +def test_valid_diagram_passes(): + result = validate_diagram(VALID_DIAGRAM) + assert isinstance(result, OpmDiagram) + + +def test_valid_diagram_version_preserved(): + result = validate_diagram(VALID_DIAGRAM) + assert result.version == "1.0" + + +def test_empty_nodes_list_accepted(): + result = validate_diagram({"version": "1.0", "nodes": [], "links": []}) + assert isinstance(result, OpmDiagram) + assert result.nodes == [] + assert result.links == [] + + +# --------------------------------------------------------------------------- +# Unit tests: hard rejections +# --------------------------------------------------------------------------- + + +def test_invalid_node_kind_rejected(): + bad = _diagram(nodes=[{"id": "x", "kind": "thing", "label": "X"}]) + with pytest.raises(ValidationError): + validate_diagram(bad) + + +def test_invalid_relation_rejected(): + bad = _diagram( + links=[{"id": "l1", "source": "farmer", "target": "grow", "relation": "causes"}] + ) + with pytest.raises(ValidationError): + validate_diagram(bad) + + +def test_label_too_long_rejected(): + long_label = "a" * 81 + bad = _diagram(nodes=[{"id": "farmer", "kind": "object", "label": long_label}]) + with pytest.raises(ValidationError): + validate_diagram(bad) + + +def test_empty_label_rejected(): + bad = _diagram(nodes=[{"id": "farmer", "kind": "object", "label": ""}]) + with pytest.raises(ValidationError): + validate_diagram(bad) + + +def test_non_kebab_node_id_rejected(): + bad = _diagram(nodes=[{"id": "MyNode", "kind": "object", "label": "My Node"}]) + with pytest.raises(ValidationError): + validate_diagram(bad) + + +def test_uppercase_node_id_rejected(): + bad = _diagram(nodes=[{"id": "UPPERCASE", "kind": "object", "label": "X"}]) + with pytest.raises(ValidationError): + validate_diagram(bad) + + +def test_duplicate_node_ids_rejected(): + bad = _diagram( + nodes=[ + {"id": "farmer", "kind": "object", "label": "Farmer"}, + {"id": "farmer", "kind": "process", "label": "Farmer 2"}, + ] + ) + with pytest.raises((ValidationError, ValueError)): + validate_diagram(bad) + + +def test_duplicate_link_ids_rejected(): + bad = _diagram( + links=[ + {"id": "l1", "source": "farmer", "target": "grow", "relation": "agent"}, + {"id": "l1", "source": "grow", "target": "crop", "relation": "result"}, + ] + ) + with pytest.raises((ValidationError, ValueError)): + validate_diagram(bad) + + +def test_dangling_link_source_rejected(): + bad = _diagram( + links=[{"id": "l1", "source": "ghost", "target": "grow", "relation": "agent"}] + ) + with pytest.raises((ValidationError, ValueError)): + validate_diagram(bad) + + +def test_dangling_link_target_rejected(): + bad = _diagram( + links=[{"id": "l1", "source": "farmer", "target": "ghost", "relation": "agent"}] + ) + with pytest.raises((ValidationError, ValueError)): + validate_diagram(bad) + + +# --------------------------------------------------------------------------- +# Unit tests: warning-only semantic checks +# --------------------------------------------------------------------------- + + +def test_self_loop_warns_not_rejects(caplog): + diagram = { + "version": "1.0", + "nodes": [{"id": "node-a", "kind": "object", "label": "A"}], + "links": [ + {"id": "l1", "source": "node-a", "target": "node-a", "relation": "effect"} + ], + } + with caplog.at_level(logging.WARNING, logger="web.app.services.opm_validate"): + result = validate_diagram(diagram) + assert isinstance(result, OpmDiagram) + assert any("Self-loop" in r.message for r in caplog.records) + + +def test_duplicate_relation_warns(caplog): + diagram = { + "version": "1.0", + "nodes": [ + {"id": "node-a", "kind": "object", "label": "A"}, + {"id": "node-b", "kind": "process", "label": "B"}, + ], + "links": [ + {"id": "l1", "source": "node-a", "target": "node-b", "relation": "agent"}, + {"id": "l2", "source": "node-a", "target": "node-b", "relation": "agent"}, + ], + } + with caplog.at_level(logging.WARNING, logger="web.app.services.opm_validate"): + result = validate_diagram(diagram) + assert isinstance(result, OpmDiagram) + assert any("Duplicate relation" in r.message for r in caplog.records) + + +# --------------------------------------------------------------------------- +# Integration tests: router behaviour +# --------------------------------------------------------------------------- + + +@pytest.fixture() +def client() -> Generator: + sys.modules.setdefault("ollama", MagicMock()) + for mod in list(sys.modules): + if mod.startswith("web.app"): + del sys.modules[mod] + from fastapi.testclient import TestClient + from web.app.main import app + with TestClient(app) as c: + yield c + + +def test_valid_stub_passes_validation_and_stores(client, tmp_path): + with patch("web.app.db.DB_PATH", tmp_path / "test.db"): + from web.app import db as db_module + db_module.init_db() + response = client.post("/opm/extract", json={"text": "some text", "save_note": False}) + assert response.status_code == 200 + data = response.json() + assert data["diagram"]["version"] == "1.0" + + +def test_invalid_diagram_blocked_before_db_insert(client, tmp_path, monkeypatch): + """An invalid dict from extraction must be rejected with 422, not stored.""" + bad_diagram = {"version": "1.0", "nodes": [{"id": "BadID", "kind": "object", "label": "X"}], "links": []} + # Patch the name as imported in the router module + monkeypatch.setattr("web.app.routers.opm.extract_opm_diagram", lambda text: bad_diagram) + + with patch("web.app.db.DB_PATH", tmp_path / "test.db"): + from web.app import db as db_module + db_module.init_db() + response = client.post("/opm/extract", json={"text": "some text", "save_note": False}) + + assert response.status_code == 422 + body = response.json() + assert body["detail"]["stage"] == "validation" + assert body["detail"]["error"] == "opm_extraction_failed" + + +def test_invalid_diagram_not_persisted(client, tmp_path, monkeypatch): + """After a validation failure, no row should appear in opm_diagrams.""" + bad_diagram = {"version": "1.0", "nodes": [{"id": "Bad", "kind": "object", "label": "X"}], "links": []} + monkeypatch.setattr("web.app.routers.opm.extract_opm_diagram", lambda text: bad_diagram) + + with patch("web.app.db.DB_PATH", tmp_path / "test.db"): + from web.app import db as db_module + db_module.init_db() + client.post("/opm/extract", json={"text": "some text", "save_note": False}) + diagrams = db_module.list_opm_diagrams() + + assert diagrams == []