"""Unit tests for graph_schema_loader and graph_schema.yaml changes. Covers: - Phase 2A / 2B split naming - Phase 2B → 2A dependency validation - Document status / admin_level enum comments - Phase 1 APPLIES_TO_REGION for Matter→Region - active_phases filtering Run: pytest tests/test_graph_schema_unit.py -v """ from __future__ import annotations import tempfile from pathlib import Path from typing import Any import pytest import yaml # --------------------------------------------------------------------------- # Helpers — we test the loader in isolation by patching _SCHEMA_PATH # --------------------------------------------------------------------------- def _make_schema_yaml(overrides: dict[str, Any] | None = None) -> dict[str, Any]: """Return a minimal valid schema dict, optionally overridden.""" base: dict[str, Any] = { "entity_types": [ {"name": "Organization", "phase": "phase_0", "key_property": "name", "description": "org", "icon": "bank", "color": "#52c41a", "extraction_mode": "llm", "properties_schema": {"name": {"type": "string"}}}, {"name": "Region", "phase": "phase_0", "key_property": "name", "description": "region", "icon": "env", "color": "#f5222d", "extraction_mode": "llm", "properties_schema": {"name": {"type": "string"}}}, {"name": "PolicyTheme", "phase": "phase_0", "key_property": "name", "description": "theme", "icon": "app", "color": "#2f54eb", "extraction_mode": "llm", "properties_schema": {"name": {"type": "string"}}}, {"name": "Article", "phase": "phase_2a", "key_property": "article_id", "description": "article", "icon": "list", "color": "#597ef7", "extraction_mode": "hybrid", "properties_schema": {"article_id": {"type": "string"}}}, ], "relationship_types": [ {"name": "ISSUED_BY", "phase": "phase_0", "source_labels": ["Document"], "target_labels": ["Organization"]}, {"name": "HAS_ARTICLE", "phase": "phase_2a", "source_labels": ["Document"], "target_labels": ["Article"]}, {"name": "SETS_CONDITION", "phase": "phase_2b", "source_labels": ["Article"], "target_labels": ["Condition"]}, ], "normalization_rules": {}, "document_properties": ["doc_id", "title", "status"], "knowledge_category_mapping": {"政策与法规": "policy_regulation"}, "active_phases": ["phase_0"], } if overrides: base.update(overrides) return base def _load_with_yaml(data: dict[str, Any]): """Write data to a temp file and load via graph_schema_loader.""" import app.core.graph_schema_loader as loader tmp = tempfile.NamedTemporaryFile( mode="w", suffix=".yaml", delete=False, encoding="utf-8" ) yaml.dump(data, tmp, allow_unicode=True) tmp.close() original_path = loader._SCHEMA_PATH try: loader._SCHEMA_PATH = Path(tmp.name) schema = loader.get_schema(force_reload=True) return schema finally: loader._SCHEMA_PATH = original_path loader.get_schema(force_reload=True) Path(tmp.name).unlink(missing_ok=True) def _load_with_yaml_expect_error(data: dict[str, Any], error_cls=ValueError): """Write data to a temp file, load, and assert it raises.""" import app.core.graph_schema_loader as loader tmp = tempfile.NamedTemporaryFile( mode="w", suffix=".yaml", delete=False, encoding="utf-8" ) yaml.dump(data, tmp, allow_unicode=True) tmp.close() original_path = loader._SCHEMA_PATH try: loader._SCHEMA_PATH = Path(tmp.name) with pytest.raises(error_cls): loader.get_schema(force_reload=True) finally: loader._SCHEMA_PATH = original_path loader.get_schema(force_reload=True) Path(tmp.name).unlink(missing_ok=True) # =========================================================================== # Test: Phase 2A / 2B split naming in real graph_schema.yaml # =========================================================================== class TestRealSchemaPhaseNaming: """Verify the actual graph_schema.yaml has correct phase_2a/2b naming.""" @pytest.fixture(autouse=True) def _load_real_schema(self): schema_path = ( Path(__file__).resolve().parent.parent / "app" / "config" / "graph_schema.yaml" ) with open(schema_path, encoding="utf-8") as f: self.raw = yaml.safe_load(f) def test_no_bare_phase_2(self): """No entity or relationship should use plain 'phase_2'.""" for et in self.raw["entity_types"]: assert et["phase"] != "phase_2", ( f"Entity '{et['name']}' still uses 'phase_2'; " f"should be 'phase_2a' or 'phase_2b'" ) for rt in self.raw["relationship_types"]: assert rt["phase"] != "phase_2", ( f"Relationship '{rt['name']}' still uses 'phase_2'; " f"should be 'phase_2a' or 'phase_2b'" ) def test_article_is_phase_2a(self): articles = [e for e in self.raw["entity_types"] if e["name"] == "Article"] assert len(articles) == 1 assert articles[0]["phase"] == "phase_2a" def test_has_article_is_phase_2a(self): rels = [r for r in self.raw["relationship_types"] if r["name"] == "HAS_ARTICLE"] assert len(rels) == 1 assert rels[0]["phase"] == "phase_2a" def test_sets_condition_is_phase_2b(self): rels = [r for r in self.raw["relationship_types"] if r["name"] == "SETS_CONDITION"] assert len(rels) == 1 assert rels[0]["phase"] == "phase_2b" def test_sets_time_limit_is_phase_2b(self): rels = [r for r in self.raw["relationship_types"] if r["name"] == "SETS_TIME_LIMIT"] assert len(rels) == 1 assert rels[0]["phase"] == "phase_2b" def test_assigns_to_is_phase_2b(self): rels = [r for r in self.raw["relationship_types"] if r["name"] == "ASSIGNS_TO"] assert len(rels) == 1 assert rels[0]["phase"] == "phase_2b" def test_active_phases_no_bare_phase_2(self): """active_phases (including commented ones) should not have 'phase_2'.""" active = self.raw.get("active_phases", []) assert "phase_2" not in active # =========================================================================== # Test: Phase 2B depends on Phase 2A # =========================================================================== class TestPhase2bDependency: """Phase 2B cannot be enabled without Phase 2A.""" def test_phase_2b_without_2a_raises(self): data = _make_schema_yaml({"active_phases": ["phase_0", "phase_2b"]}) _load_with_yaml_expect_error(data, ValueError) def test_phase_2b_with_2a_succeeds(self): data = _make_schema_yaml({"active_phases": ["phase_0", "phase_2a", "phase_2b"]}) schema = _load_with_yaml(data) assert "phase_2a" in schema.active_phases assert "phase_2b" in schema.active_phases def test_phase_2a_alone_succeeds(self): data = _make_schema_yaml({"active_phases": ["phase_0", "phase_2a"]}) schema = _load_with_yaml(data) assert "phase_2a" in schema.active_phases assert "phase_2b" not in schema.active_phases def test_phase_0_only_succeeds(self): data = _make_schema_yaml({"active_phases": ["phase_0"]}) schema = _load_with_yaml(data) assert schema.active_phases == ["phase_0"] # =========================================================================== # Test: active_phases filtering # =========================================================================== class TestActivePhaseFiltering: """Entity/relationship types are filtered by active_phases.""" def test_phase_0_only_loads_phase_0_types(self): data = _make_schema_yaml({"active_phases": ["phase_0"]}) schema = _load_with_yaml(data) assert "Organization" in schema.entity_type_names() assert "Article" not in schema.entity_type_names() assert "ISSUED_BY" in schema.rel_type_names() assert "HAS_ARTICLE" not in schema.rel_type_names() assert "SETS_CONDITION" not in schema.rel_type_names() def test_phase_2a_loads_article_and_has_article(self): data = _make_schema_yaml({"active_phases": ["phase_0", "phase_2a"]}) schema = _load_with_yaml(data) assert "Article" in schema.entity_type_names() assert "HAS_ARTICLE" in schema.rel_type_names() assert "SETS_CONDITION" not in schema.rel_type_names() def test_phase_2a_2b_loads_all(self): data = _make_schema_yaml({ "active_phases": ["phase_0", "phase_2a", "phase_2b"], }) schema = _load_with_yaml(data) assert "Article" in schema.entity_type_names() assert "HAS_ARTICLE" in schema.rel_type_names() assert "SETS_CONDITION" in schema.rel_type_names() # =========================================================================== # Test: Document status/admin_level enum comments in YAML # =========================================================================== class TestDocumentPropertyComments: """Verify YAML file comments match PRD enum values.""" @pytest.fixture(autouse=True) def _load_raw_yaml_text(self): schema_path = ( Path(__file__).resolve().parent.parent / "app" / "config" / "graph_schema.yaml" ) self.raw_text = schema_path.read_text(encoding="utf-8") def test_status_comment_complete(self): """status comment should contain all 5 enum values.""" required = ["有效", "部分失效", "已废止", "失效", "待确认"] # Find the status comment line for line in self.raw_text.splitlines(): if "- status" in line and "#" in line: comment = line.split("#", 1)[1] for val in required: assert val in comment, ( f"status comment missing '{val}': {line.strip()}" ) break else: pytest.fail("No '- status' line with comment found in YAML") def test_admin_level_comment_complete(self): """admin_level comment should contain all 7 enum values.""" required = ["国家级", "省级", "市级", "区县级", "乡镇街道级", "其他", "未知"] for line in self.raw_text.splitlines(): if "- admin_level" in line and "#" in line: comment = line.split("#", 1)[1] for val in required: assert val in comment, ( f"admin_level comment missing '{val}': {line.strip()}" ) break else: pytest.fail("No '- admin_level' line with comment found in YAML") # =========================================================================== # Test: Phase 1 APPLIES_TO_REGION for Matter→Region # =========================================================================== # =========================================================================== # Test: key_prop_for() public method # =========================================================================== class TestKeyPropFor: """GraphSchema.key_prop_for() returns the correct key property for labels.""" def test_person_key_property(self): """Person uses person_id, not name.""" data = _make_schema_yaml({ "entity_types": [ {"name": "Organization", "phase": "phase_0", "key_property": "name", "description": "org", "icon": "bank", "color": "#52c41a", "extraction_mode": "llm", "properties_schema": {"name": {"type": "string"}}}, {"name": "Person", "phase": "phase_3", "key_property": "person_id", "description": "person", "icon": "user", "color": "#faad14", "extraction_mode": "llm", "properties_schema": {"person_id": {"type": "string"}}}, ], "relationship_types": [], "active_phases": ["phase_0"], }) schema = _load_with_yaml(data) assert schema.key_prop_for("Person") == "person_id" def test_default_key_property_is_name(self): data = _make_schema_yaml() schema = _load_with_yaml(data) assert schema.key_prop_for("Organization") == "name" def test_unknown_label_falls_back_to_name(self): data = _make_schema_yaml() schema = _load_with_yaml(data) assert schema.key_prop_for("NonExistentType") == "name" # =========================================================================== # Test: entity_types_for_scene() scene-based type resolution # =========================================================================== class TestEntityTypesForScene: """Scene-based type set uses active_phases + scene's extra_phases.""" def test_empty_scene_returns_active_set(self): data = _make_schema_yaml({"active_phases": ["phase_0"]}) schema = _load_with_yaml(data) scene_types = schema.entity_types_for_scene("") names = {et["name"] for et in scene_types} assert "Organization" in names assert "Article" not in names def test_scene_with_extra_phases(self): data = _make_schema_yaml({ "active_phases": ["phase_0"], "extraction_scenes": { "test_scene": { "extra_phases": ["phase_2a"], }, }, }) schema = _load_with_yaml(data) scene_types = schema.entity_types_for_scene("test_scene") names = {et["name"] for et in scene_types} assert "Organization" in names assert "Article" in names def test_unknown_scene_returns_active_set(self): data = _make_schema_yaml({"active_phases": ["phase_0"]}) schema = _load_with_yaml(data) scene_types = schema.entity_types_for_scene("nonexistent_scene") names = {et["name"] for et in scene_types} assert "Organization" in names assert "Article" not in names # =========================================================================== # Test: Phase 1 APPLIES_TO_REGION for Matter→Region # =========================================================================== class TestPhase1AppliestoRegion: """Phase 1 should include APPLIES_TO_REGION with source_labels=[Matter].""" @pytest.fixture(autouse=True) def _load_real_schema(self): schema_path = ( Path(__file__).resolve().parent.parent / "app" / "config" / "graph_schema.yaml" ) with open(schema_path, encoding="utf-8") as f: self.raw = yaml.safe_load(f) def test_applies_to_region_has_matter_source(self): """There should be an APPLIES_TO_REGION with source_labels=[Matter] in phase_1.""" phase1_atr = [ r for r in self.raw["relationship_types"] if r["name"] == "APPLIES_TO_REGION" and r["phase"] == "phase_1" ] assert len(phase1_atr) == 1, ( "Expected one APPLIES_TO_REGION in phase_1, " f"found {len(phase1_atr)}" ) assert "Matter" in phase1_atr[0]["source_labels"] def test_applies_to_region_still_has_document_source(self): """Phase 0 APPLIES_TO_REGION with source_labels=[Document] should remain.""" phase0_atr = [ r for r in self.raw["relationship_types"] if r["name"] == "APPLIES_TO_REGION" and r["phase"] == "phase_0" ] assert len(phase0_atr) == 1 assert "Document" in phase0_atr[0]["source_labels"]