vee
352 lines
10 KiB
Python
352 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
from copy import deepcopy
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
from cognitive_agent.application.scene_templates import SceneTemplateRegistry
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
def _operation(
|
|
capability: str = "climate.set_temperature",
|
|
*,
|
|
domain: str = "climate",
|
|
target_scope: str = "vehicle",
|
|
parameters: dict[str, object] | None = None,
|
|
) -> dict[str, object]:
|
|
return {
|
|
"domain": domain,
|
|
"capability": capability,
|
|
"target_scope": target_scope,
|
|
"operation": "set",
|
|
"parameters": parameters or {"temperature_c": 24},
|
|
"reason": "模板定义的安全、稳定目标。",
|
|
}
|
|
|
|
|
|
def _template_document(
|
|
template_id: str,
|
|
name: str,
|
|
*,
|
|
aliases: list[str] | None = None,
|
|
operations: list[dict[str, object]] | None = None,
|
|
) -> dict[str, object]:
|
|
return {
|
|
"id": template_id,
|
|
"name": name,
|
|
"aliases": aliases or [],
|
|
"description": f"{name}的本地快速组合模板。",
|
|
"icon": "sparkles",
|
|
"operations": operations or [_operation()],
|
|
}
|
|
|
|
|
|
def _write_template(root: Path, filename: str, document: object) -> Path:
|
|
root.mkdir(parents=True, exist_ok=True)
|
|
path = root / filename
|
|
path.write_text(
|
|
yaml.safe_dump(document, allow_unicode=True, sort_keys=False),
|
|
encoding="utf-8",
|
|
)
|
|
return path
|
|
|
|
|
|
@pytest.fixture
|
|
def template_root(tmp_path: Path) -> Path:
|
|
root = tmp_path / "scene_templates"
|
|
_write_template(
|
|
root,
|
|
"viewing.yaml",
|
|
_template_document(
|
|
"viewing",
|
|
"观影模式",
|
|
aliases=["电影模式", "MOVIE MODE"],
|
|
operations=[
|
|
_operation(),
|
|
_operation(
|
|
"roof.set_shade_position",
|
|
domain="roof",
|
|
parameters={"position_pct": 0},
|
|
),
|
|
],
|
|
),
|
|
)
|
|
_write_template(
|
|
root,
|
|
"quiet.yaml",
|
|
_template_document("quiet", "静谧模式", aliases=["安静模式"]),
|
|
)
|
|
return root
|
|
|
|
|
|
def test_registry_loads_independent_directory_and_exports_safe_public_plans(
|
|
template_root: Path,
|
|
) -> None:
|
|
registry = SceneTemplateRegistry(template_root)
|
|
|
|
assert [template.id for template in registry.templates()] == ["quiet", "viewing"]
|
|
public = registry.public_templates(include_plan=True)
|
|
assert [item["id"] for item in public] == ["quiet", "viewing"]
|
|
viewing = next(item for item in public if item["id"] == "viewing")
|
|
assert viewing["name"] == "观影模式"
|
|
assert {"观影模式", "观影", "电影模式", "MOVIE MODE"}.issubset(
|
|
set(viewing["aliases"])
|
|
)
|
|
assert viewing["origin"] == "builtin_scene_template"
|
|
assert viewing["read_only"] is True
|
|
assert viewing["plan"] == registry.get_template("viewing").plan()
|
|
|
|
|
|
def test_template_plan_has_complete_guard_shape_and_auditable_metadata(
|
|
template_root: Path,
|
|
) -> None:
|
|
registry = SceneTemplateRegistry(template_root)
|
|
plan = registry.get_template("viewing").plan()
|
|
|
|
assert set(plan) == {
|
|
"detected_domains",
|
|
"requests",
|
|
"ambiguities",
|
|
"decision_summary",
|
|
"confidence",
|
|
"metadata",
|
|
}
|
|
assert plan["detected_domains"] == ["climate", "roof"]
|
|
assert plan["ambiguities"] == []
|
|
assert plan["decision_summary"]
|
|
assert plan["confidence"] == 1.0
|
|
assert plan["metadata"]["source"] == "scene_template"
|
|
assert plan["metadata"]["template_id"] == "viewing"
|
|
assert isinstance(plan["metadata"]["template_revision"], str)
|
|
assert plan["metadata"]["template_revision"]
|
|
|
|
assert [request["capability"] for request in plan["requests"]] == [
|
|
"climate.set_temperature",
|
|
"roof.set_shade_position",
|
|
]
|
|
assert all(
|
|
set(request)
|
|
== {
|
|
"domain",
|
|
"capability",
|
|
"target_scope",
|
|
"operation",
|
|
"duration_ms",
|
|
"parameters",
|
|
"reason",
|
|
}
|
|
for request in plan["requests"]
|
|
)
|
|
assert plan["requests"][0]["duration_ms"] is None
|
|
assert plan["requests"][0]["parameters"] == [{"name": "temperature_c", "value": 24}]
|
|
|
|
|
|
def test_template_plan_and_public_projection_are_fresh_copies(template_root: Path) -> None:
|
|
registry = SceneTemplateRegistry(template_root)
|
|
original = registry.get_template("viewing").plan()
|
|
|
|
mutated_plan = registry.get_template("viewing").plan()
|
|
mutated_plan["requests"][0]["parameters"][0]["value"] = 31
|
|
mutated_public = registry.public_templates(include_plan=True)
|
|
mutated_public[1]["plan"]["requests"].clear()
|
|
|
|
assert registry.get_template("viewing").plan() == original
|
|
assert registry.public_templates(include_plan=True)[1]["plan"]["requests"]
|
|
|
|
|
|
def test_template_revision_tracks_semantics_but_not_operation_order(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
operations = [
|
|
_operation(),
|
|
_operation(
|
|
"roof.set_shade_position",
|
|
domain="roof",
|
|
parameters={"position_pct": 0},
|
|
),
|
|
]
|
|
revisions = []
|
|
for directory, current_operations in (
|
|
("first", operations),
|
|
("reordered", list(reversed(operations))),
|
|
(
|
|
"changed",
|
|
[
|
|
operations[0],
|
|
{
|
|
**operations[1],
|
|
"parameters": {"position_pct": 100},
|
|
},
|
|
],
|
|
),
|
|
):
|
|
root = tmp_path / directory
|
|
_write_template(
|
|
root,
|
|
"viewing.yaml",
|
|
_template_document(
|
|
"viewing",
|
|
"观影模式",
|
|
operations=current_operations,
|
|
),
|
|
)
|
|
revisions.append(SceneTemplateRegistry(root).get_template("viewing").revision)
|
|
|
|
assert revisions[0] == revisions[1]
|
|
assert revisions[0] != revisions[2]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"query",
|
|
(
|
|
"viewing",
|
|
"VIEWING",
|
|
"观影模式",
|
|
" 观 影 模 式 ",
|
|
"电影模式",
|
|
" movie mode ",
|
|
"“观影模式”",
|
|
"《观影模式》",
|
|
),
|
|
)
|
|
def test_exact_name_resolves_id_name_and_alias_after_normalization(
|
|
template_root: Path,
|
|
query: str,
|
|
) -> None:
|
|
registry = SceneTemplateRegistry(template_root)
|
|
|
|
assert registry.resolve_exact_name(query).id == "viewing"
|
|
|
|
|
|
@pytest.mark.parametrize("query", ("", "观", "我的观影模式", "电影模式增强版", "未知模式"))
|
|
def test_exact_name_never_uses_substring_or_fuzzy_matching(
|
|
template_root: Path,
|
|
query: str,
|
|
) -> None:
|
|
registry = SceneTemplateRegistry(template_root)
|
|
|
|
assert registry.resolve_exact_name(query) is None
|
|
|
|
|
|
def test_ambiguous_alias_is_rejected_without_first_file_wins(tmp_path: Path) -> None:
|
|
root = tmp_path / "scene_templates"
|
|
_write_template(
|
|
root,
|
|
"quiet.yaml",
|
|
_template_document("quiet", "静谧模式", aliases=["休息模式"]),
|
|
)
|
|
_write_template(
|
|
root,
|
|
"rest.yaml",
|
|
_template_document("rest", "休憩模式", aliases=["休息模式"]),
|
|
)
|
|
with pytest.raises(ValueError, match="ambiguous scene template name"):
|
|
SceneTemplateRegistry(root)
|
|
|
|
|
|
def test_unknown_template_id_raises_key_error(template_root: Path) -> None:
|
|
registry = SceneTemplateRegistry(template_root)
|
|
|
|
with pytest.raises(KeyError, match="missing"):
|
|
registry.get_template("missing")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"documents",
|
|
(
|
|
[_template_document("Bad ID", "坏标识模式")],
|
|
[{"id": "missing_operations", "name": "缺动作模式", "operations": []}],
|
|
[
|
|
_template_document("duplicate", "重复一"),
|
|
_template_document("duplicate", "重复二"),
|
|
],
|
|
[
|
|
_template_document(
|
|
"relative",
|
|
"相对控制模式",
|
|
operations=[
|
|
{
|
|
**_operation(),
|
|
"operation": "increase",
|
|
}
|
|
],
|
|
)
|
|
],
|
|
[
|
|
_template_document(
|
|
"unsupported",
|
|
"越权模式",
|
|
operations=[_operation("vehicle.fly", domain="vehicle")],
|
|
)
|
|
],
|
|
[
|
|
_template_document(
|
|
"too-many",
|
|
"动作过多模式",
|
|
operations=[deepcopy(_operation()) for _index in range(13)],
|
|
)
|
|
],
|
|
[
|
|
_template_document(
|
|
"duplicate-operation",
|
|
"重复动作模式",
|
|
operations=[_operation(), deepcopy(_operation())],
|
|
)
|
|
],
|
|
[
|
|
_template_document(
|
|
"nested-parameter",
|
|
"嵌套参数模式",
|
|
operations=[_operation(parameters={"temperature_c": {"value": 24}})],
|
|
)
|
|
],
|
|
[
|
|
_template_document(
|
|
"wrong-domain",
|
|
"错误领域模式",
|
|
operations=[_operation(domain="roof")],
|
|
)
|
|
],
|
|
),
|
|
ids=(
|
|
"invalid-id",
|
|
"empty-operations",
|
|
"duplicate-id",
|
|
"relative-operation",
|
|
"unsupported-capability",
|
|
"too-many-operations",
|
|
"duplicate-operation",
|
|
"nested-parameter",
|
|
"capability-domain-mismatch",
|
|
),
|
|
)
|
|
def test_invalid_template_items_are_rejected_at_load_time(
|
|
tmp_path: Path,
|
|
documents: list[dict[str, object]],
|
|
) -> None:
|
|
root = tmp_path / "scene_templates"
|
|
for index, document in enumerate(deepcopy(documents)):
|
|
_write_template(root, f"template-{index}.yaml", document)
|
|
|
|
with pytest.raises(ValueError):
|
|
SceneTemplateRegistry(root)
|
|
|
|
|
|
def test_production_template_directory_is_nonempty_and_self_consistent() -> None:
|
|
root = ROOT / "config" / "scene_templates"
|
|
assert root.is_dir()
|
|
|
|
registry = SceneTemplateRegistry(root)
|
|
templates = registry.templates()
|
|
public = registry.public_templates(include_plan=True)
|
|
|
|
assert templates
|
|
assert len(public) == len(templates)
|
|
assert all(item["read_only"] is True for item in public)
|
|
assert all(item["plan"]["metadata"]["template_id"] == item["id"] for item in public)
|
|
assert registry.resolve_exact_name("观影模式") is not None
|