adapter_python_pg/tests/test_ontology_spec.py

"""本体 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")