"""Tests for Mantara Type-S adapter.""" import json from pathlib import Path from pipeline.mantara_adapter import ( AdapterResult, Step4Bundle, build_mantara_input, ) def _seed_run_dir(tmp_path: Path) -> Path: """Create a synthetic Step-4 output directory.""" run_dir = tmp_path / "runs" / "test123" cir_dir = run_dir / "cir" prd_dir = run_dir / "prd" diagrams = prd_dir / "diagrams" cir_dir.mkdir(parents=True) prd_dir.mkdir(parents=True) diagrams.mkdir(parents=True) enriched = { "manifest": {"name": "Test Order System", "domain": "warehouse", "type": "web", "description": "A warehouse order management app."}, "entities": [ {"name": "Order", "table_name": "order", "fields": [ {"name": "id", "type": "uuid", "primary_key": True, "required": True}, {"name": "orderId", "type": "string", "required": True, "unique": True}, {"name": "status", "type": "enum", "required": True, "enum_values": ["draft", "submitted", "fulfilled"]}, {"name": "total", "type": "numeric"}, ], "relationships": [ {"type": "many-to-one", "target": "Customer", "foreign_key": "customerId"} ], "confidence": 0.9}, {"name": "Customer", "fields": [ {"name": "id", "type": "uuid", "primary_key": True, "required": True}, {"name": "email", "type": "email", "required": True}, ]}, ], "workflows": [ {"name": "order_lifecycle", "states": ["draft", "submitted", "fulfilled"], "initial_state": "draft", "bound_entity": "Order", "transitions": [ {"from_state": "draft", "to_state": "submitted", "trigger": "submit"} ]}, ], "rules": [ {"name": "min_one_item", "applies_to": "Order", "description": "Order must have at least one item"}, ], "roles": [{"name": "warehouse_user"}], "actions": [ {"name": "create_order", "screen": "Create Order", "trigger": "Create button", "expected_outcome": "order created"}, {"name": "submit_order", "screen": "Order Detail", "trigger": "Submit", "expected_outcome": "order moves to submitted"}, ], "screens": [], "states": [], "unknowns": [], "dropped": [], "confidence": {"overall": 0.9}, "fidelity": {}, "run_id": "test123", "generated_at": "2026-04-26T00:00:00Z", "source_step02_extract": "n/a", } (cir_dir / "enriched_cir.json").write_text(json.dumps(enriched)) # Minimal Step-4 outputs (prd_dir / "PRD.md").write_text("# Test Order System\n\n## Goal\nManage orders.\n") (prd_dir / "entities.json").write_text(json.dumps({ "$schema": "https://json-schema.org/draft/2020-12/schema", "$defs": {} })) (prd_dir / "workflows.json").write_text(json.dumps({ "machines": [{"id": "order_lifecycle", "x-id": "WF-001"}] })) (prd_dir / "nfrs.json").write_text(json.dumps({ "x-app-type": "web", "attributes": { "security": {"requirements": ["TLS 1.2+", "OWASP Top 10"]}, "reliability": {"requirements": ["99.5% availability"]}, } })) (prd_dir / "glossary.json").write_text(json.dumps({"terms": []})) (prd_dir / "open_questions.json").write_text(json.dumps({"questions": []})) (prd_dir / "stories.json").write_text(json.dumps({ "stories": [ {"id": "US-001", "priority": "P1", "action": "create_order"}, {"id": "US-002", "priority": "P2", "action": "submit_order"}, ] })) (prd_dir / "codegen_brief.json").write_text(json.dumps({"schemaTargets": []})) (diagrams / "er.mmd").write_text("erDiagram\n Order { uuid id PK }\n") return run_dir # --------------------------------------------------------------------------- # # Loader # --------------------------------------------------------------------------- # def test_step4_bundle_loads(tmp_path: Path) -> None: run_dir = _seed_run_dir(tmp_path) bundle = Step4Bundle.load(run_dir) assert bundle.enriched_cir["manifest"]["name"] == "Test Order System" assert len(bundle.enriched_cir["entities"]) == 2 assert bundle.nfrs_json["x-app-type"] == "web" assert "erDiagram" in bundle.er_mmd # --------------------------------------------------------------------------- # # Build the input # --------------------------------------------------------------------------- # def test_build_mantara_input_basic(tmp_path: Path) -> None: run_dir = _seed_run_dir(tmp_path) bundle = Step4Bundle.load(run_dir) result = build_mantara_input(bundle) assert isinstance(result, AdapterResult) assert result.system_name == "Test Order System" assert result.schema_name and len(result.schema_name) <= 5 # NL prelude assert "Test Order System" in result.text assert "warehouse" in result.text # Type S body assert "Menu:" in result.text assert "Submenu:" in result.text assert "Fields:" in result.text # FSD ANALYSIS marker assert "=== FSD ANALYSIS" in result.text assert "DETECTED ENTITIES" in result.text # NFR hints assert "ISO/IEC 25010" in result.text def test_build_input_groups_entities_into_at_least_2_menus(tmp_path: Path) -> None: """v8 hard rule: every menu MUST have ≥ 2 submenus AND we need ≥ 2 menus.""" run_dir = _seed_run_dir(tmp_path) bundle = Step4Bundle.load(run_dir) result = build_mantara_input(bundle) assert len(result.menu_groups) >= 2 def test_enum_fields_become_cfg_lookup_seeds(tmp_path: Path) -> None: """v8: status enum → cfg__status seed (NO PostgreSQL ENUMs).""" run_dir = _seed_run_dir(tmp_path) bundle = Step4Bundle.load(run_dir) result = build_mantara_input(bundle) # workflow.states should populate cfg_order_status assert any("cfg_order_status" in k for k in result.enum_seeds.keys()) # values should be snake_case for vals in result.enum_seeds.values(): for v in vals: assert v.islower() or "_" in v or v.isalnum() # enum field on Order.status should also generate cfg assert any("order_status" in k or "_status" in k for k in result.enum_seeds.keys()) def test_field_types_use_v8_compliant_sql(tmp_path: Path) -> None: """v8: enum → VARCHAR(30), numeric → NUMERIC(12,2), no FLOAT/DOUBLE.""" run_dir = _seed_run_dir(tmp_path) bundle = Step4Bundle.load(run_dir) result = build_mantara_input(bundle) # enum field rendered as VARCHAR(30) FK assert "VARCHAR(30) FK" in result.text or "VARCHAR(30)" in result.text # numeric → NUMERIC(12,2) assert "NUMERIC(12,2)" in result.text # no FLOAT mentioned for business fields # (FLOAT may appear in glossary text, but not in our type_hints) def test_relationships_become_fk_hints(tmp_path: Path) -> None: run_dir = _seed_run_dir(tmp_path) bundle = Step4Bundle.load(run_dir) result = build_mantara_input(bundle) assert "Customer" in result.text # FK is mentioned in Type S body or FSD block assert "FK" in result.text def test_business_rules_appear_in_fsd_block(tmp_path: Path) -> None: run_dir = _seed_run_dir(tmp_path) bundle = Step4Bundle.load(run_dir) result = build_mantara_input(bundle) assert "min_one_item" in result.text or "at least one item" in result.text def test_priority_indicators_from_stories(tmp_path: Path) -> None: run_dir = _seed_run_dir(tmp_path) bundle = Step4Bundle.load(run_dir) result = build_mantara_input(bundle) assert "PRIORITY INDICATORS" in result.text assert "create_order" in result.text # P1 from stories def test_include_prd_prose_flag(tmp_path: Path) -> None: run_dir = _seed_run_dir(tmp_path) bundle = Step4Bundle.load(run_dir) with_prd = build_mantara_input(bundle, include_prd_prose=True) without_prd = build_mantara_input(bundle, include_prd_prose=False) assert with_prd.char_count > without_prd.char_count assert "## Goal" in with_prd.text assert "## Goal" not in without_prd.text or "## Goal" in without_prd.text # noqa def test_schema_name_is_short_snake_case(tmp_path: Path) -> None: run_dir = _seed_run_dir(tmp_path) bundle = Step4Bundle.load(run_dir) result = build_mantara_input(bundle) assert result.schema_name.islower() assert len(result.schema_name) <= 5 assert " " not in result.schema_name def test_each_entity_gets_its_own_submenu(tmp_path: Path) -> None: """v8: one table = one unique submenu.""" run_dir = _seed_run_dir(tmp_path) bundle = Step4Bundle.load(run_dir) result = build_mantara_input(bundle) # 2 entities → at least 2 submenus total_submenus = sum(len(s) for _, s in result.menu_groups) assert total_submenus >= 2 def test_density_constraint_in_prelude(tmp_path: Path) -> None: """The prelude warns Mantara about the 1-submenu menu trap (caught 4 of 13 errors in P2 live run).""" run_dir = _seed_run_dir(tmp_path) bundle = Step4Bundle.load(run_dir) result = build_mantara_input(bundle) assert "MENU DENSITY CONSTRAINT" in result.text assert "≥2 submenus" in result.text or ">=2 submenus" in result.text def test_anticipated_operational_menus_pre_declared(tmp_path: Path) -> None: """FSD block pre-declares User Management + Notifications + Configuration with ≥2 submenus each so Mantara doesn't auto-add 1-submenu menus.""" run_dir = _seed_run_dir(tmp_path) bundle = Step4Bundle.load(run_dir) result = build_mantara_input(bundle) assert "User Management" in result.text assert "User Directory" in result.text assert "Roles & Permissions" in result.text assert "Notifications & Audit" in result.text assert "Audit Log" in result.text