"""本体 spec 测试。 其中 `test_readme_example_loads` 直接解析 README 里那段 YAML —— 文档示例一旦写坏， 这里先红，读者不会抄到坏例子。 """ from __future__ import annotations import re from pathlib import Path import pytest import yaml from adapter_python_pg.errors import OntologyError from adapter_python_pg.ontology.spec import ( Cardinality, EntityKind, PropertyType, RuleKind, load_ontology, ) README = Path(__file__).resolve().parents[1] / "README.md" def parse(mapping: dict[str, object]) -> object: """走公开入口（YAML 文本 → Ontology），与真实使用路径一致。""" return load_ontology(yaml.safe_dump(mapping, allow_unicode=True)) def readme_yaml() -> str: text = README.read_text(encoding="utf-8") block = re.search(r"```yaml\n(.*?)```", text, re.S) assert block, "README 里没有 yaml 代码块" return block.group(1) def minimal(**overrides: object) -> dict[str, object]: """最简本体，供各用例覆盖局部字段。""" base: dict[str, object] = { "ontology": "t", "entities": { "A": {}, "B": {}, }, "links": {"A_B": {"type": "rel", "from": "A", "to": "B"}}, } base.update(overrides) return base # -- README 契约 ----------------------------------------------------------- def test_readme_example_loads() -> None: onto = load_ontology(readme_yaml()) assert onto.ontology == "factory-twin" assert set(onto.entities) == {"Factory", "Line", "Device", "SensorReading"} assert set(onto.links) == {"Line_Factory", "Device_Line"} assert onto.entities["SensorReading"].kind is EntityKind.TIMESERIES assert onto.entities["SensorReading"].time == "ts" assert onto.entities["Factory"].properties["location"].type is PropertyType.GEO assert onto.entities["Line"].properties["status"].values == ("running", "idle", "fault") assert onto.links["Device_Line"].cardinality is Cardinality.N_TO_ONE assert onto.links["Device_Line"].from_ == "Device" assert onto.rules["Device_inFactory"].kind is RuleKind.DERIVE_LINK assert onto.rules["Device_inFactory"].via == ("mounted_on", "belongs_to") def test_readme_rule_chain_resolves_device_to_factory() -> None: onto = load_ontology(readme_yaml()) assert onto.link_types()["mounted_on"] == [("Device", "Line")] assert onto.link_types()["belongs_to"] == [("Line", "Factory")] def test_timeseries_entities_helper() -> None: onto = load_ontology(readme_yaml()) assert list(onto.timeseries_entities()) == ["SensorReading"] # -- 跨引用校验 ------------------------------------------------------------ def test_link_target_must_exist() -> None: bad = minimal(links={"A_B": {"type": "rel", "from": "A", "to": "Nope"}}) with pytest.raises(OntologyError, match=r"link A_B\.to 引用了不存在的实体"): parse(bad) def test_link_source_must_exist() -> None: bad = minimal(links={"A_B": {"type": "rel", "from": "Nope", "to": "B"}}) with pytest.raises(OntologyError, match=r"link A_B\.from 引用了不存在的实体"): parse(bad) def test_rule_via_must_be_connected() -> None: bad = minimal( rules={ "backwards": { "kind": "derive_link", "from": "B", "to": "A", "via": ["rel"], # rel 只从 A 出发，B 走不动 } } ) with pytest.raises(OntologyError, match="via 链在第 1 步断了"): parse(bad) def test_rule_chain_end_must_match_to() -> None: bad = minimal( entities={"A": {}, "B": {}, "C": {}}, links={ "A_B": {"type": "rel", "from": "A", "to": "B"}, "B_C": {"type": "rel2", "from": "B", "to": "C"}, }, rules={ "short": {"kind": "derive_link", "from": "A", "to": "C", "via": ["rel"]}, }, ) with pytest.raises(OntologyError, match="via 链终点是 'B'"): parse(bad) def test_rule_via_unknown_type() -> None: bad = minimal( rules={"r": {"kind": "derive_link", "from": "A", "to": "B", "via": ["nope"]}}, ) with pytest.raises(OntologyError, match="不存在的 link type"): parse(bad) # -- 字段级校验 ------------------------------------------------------------ def test_enum_requires_values() -> None: bad = minimal(entities={"A": {"properties": {"s": {"type": "enum"}}}}) with pytest.raises(OntologyError, match="enum 属性必须给 values"): parse(bad) def test_values_only_on_enum() -> None: bad = minimal(entities={"A": {"properties": {"s": {"type": "string", "values": ["x"]}}}}) with pytest.raises(OntologyError, match="只有 enum 属性可以给 values"): parse(bad) def test_list_requires_of() -> None: bad = minimal(entities={"A": {"properties": {"tags": {"type": "list"}}}}) with pytest.raises(OntologyError, match="list 属性必须给 of"): parse(bad) def test_timeseries_requires_declared_time() -> None: bad = minimal( entities={ "A": {"kind": "timeseries", "time": "ts"}, # properties 里没有 ts } ) with pytest.raises(OntologyError, match="未在 properties 中定义"): parse(bad) def test_timeseries_time_must_be_datetime() -> None: bad = minimal( entities={ "A": { "kind": "timeseries", "time": "ts", "properties": {"ts": {"type": "string"}}, } } ) with pytest.raises(OntologyError, match="time 属性类型必须是 datetime"): parse(bad) def test_unknown_field_is_rejected() -> None: """拼错字段名必须报错，而不是被静默忽略。""" bad = minimal(entities={"A": {"properties": {"x": {"type": "string", "indx": True}}}}) with pytest.raises(OntologyError, match="indx"): parse(bad) def test_bad_identifier() -> None: bad = minimal(entities={"A; drop table x": {}}) with pytest.raises(OntologyError, match="名字非法"): parse(bad) # -- 加载入口 -------------------------------------------------------------- def test_load_from_text_and_path(tmp_path: Path) -> None: text = readme_yaml() assert load_ontology(text).ontology == "factory-twin" path = tmp_path / "onto.yaml" path.write_text(text, encoding="utf-8") assert load_ontology(path).ontology == "factory-twin" def test_missing_file(tmp_path: Path) -> None: with pytest.raises(OntologyError, match="本体文件不存在"): load_ontology(tmp_path / "nope.yaml") def test_yaml_syntax_error() -> None: with pytest.raises(OntologyError, match="YAML 语法错误"): load_ontology("ontology: x\nentities: [unclosed") def test_root_must_be_mapping() -> None: with pytest.raises(OntologyError, match="根节点必须是映射"): load_ontology("- a\n- b\n")