Cognitive_Agent/tests/unit/test_mode_scene_catalog.py

from __future__ import annotations

import json
from pathlib import Path

from cognitive_agent.application.mode_profile import ModeProfileRegistry
from cognitive_agent.application.mode_scene_catalog import (
    ModeSceneCatalogPublisher,
    build_mode_scene_catalog,
)
from cognitive_agent.application.seat_service import REPOSITORY_ROOT


def _write_office_preset(path: Path) -> None:
    path.parent.mkdir(parents=True)
    path.write_text(
        """
id: office
name: 办公模式
description: N7 办公场景。
icon: briefcase
scope:
  vehicle_ids: [denza_n7]
operations:
  - domain: climate
    capability: climate.set_temperature
    target_scope: passenger
    operation: set
    parameters:
      temperature_c: 24
""".strip()
        + "\n",
        encoding="utf-8",
    )


def test_publisher_tracks_create_override_and_delete(tmp_path: Path) -> None:
    preset_root = tmp_path / "presets"
    user_root = tmp_path / "user"
    _write_office_preset(
        preset_root / "vehicles" / "denza_n7" / "preset" / "office.yaml"
    )
    registry = ModeProfileRegistry(preset_root, user_root=user_root)
    catalog_path = tmp_path / "scenses" / "demo-scenes.json"
    publisher = ModeSceneCatalogPublisher(
        catalog_path,
        "denza_n7",
        server_id="test-n7",
    )
    registry.add_change_listener(publisher.publish)

    initial = json.loads(catalog_path.read_text(encoding="utf-8"))
    assert initial["service"] == "n7-scenes"
    assert initial["schema_version"] == 1
    assert initial["server_id"] == "test-n7"
    assert initial["scene_count"] == 1
    assert initial["scenes"][0]["origin"] == "preset"
    assert initial["scenes"][0]["actions"] == [
        {
            "type": "climate_temperature",
            "at_ms": 0,
            "domain": "climate",
            "capability": "climate.set_temperature",
            "target_scope": "passenger",
            "operation": "set",
            "parameters": {"temperature_c": 24},
            "zone": "passenger",
            "celsius": 24,
        }
    ]

    created = registry.save_mode(
        "阅读模式",
        [
            {
                "domain": "ambient_light",
                "capability": "ambient_light.set_illumination",
                "target_scope": "vehicle",
                "parameters": {"color_hex": "#E8F3FF", "brightness_pct": 40},
            }
        ],
        vehicle_id="denza_n7",
    )
    after_create = json.loads(catalog_path.read_text(encoding="utf-8"))
    created_scene = next(
        scene for scene in after_create["scenes"] if scene["id"] == created.id
    )
    assert after_create["scene_count"] == 2
    assert created_scene["origin"] == "user"
    assert created_scene["deletable"] is True
    assert created_scene["actions"][0]["parameters"]["brightness_pct"] == 40

    fan_mode = registry.save_mode(
        "微风模式",
        [
            {
                "domain": "climate",
                "capability": "climate.set_fan_level",
                "target_scope": "vehicle",
                "parameters": {"level": 2},
            }
        ],
        vehicle_id="denza_n7",
    )
    after_fan_create = json.loads(catalog_path.read_text(encoding="utf-8"))
    fan_scene = next(
        scene for scene in after_fan_create["scenes"] if scene["id"] == fan_mode.id
    )
    assert fan_scene["actions"][0]["type"] == "climate_fan"
    assert fan_scene["actions"][0]["level"] == 2

    registry.save_mode(
        "专注办公模式",
        [
            {
                "domain": "climate",
                "capability": "climate.set_temperature",
                "target_scope": "passenger",
                "parameters": {"temperature_c": 26},
            }
        ],
        mode_id="office",
        vehicle_id="denza_n7",
    )
    after_override = json.loads(catalog_path.read_text(encoding="utf-8"))
    office = next(scene for scene in after_override["scenes"] if scene["id"] == "office")
    assert office["label"] == "专注办公模式"
    assert office["origin"] == "customized"
    assert office["preset_override"] is True
    assert office["actions"][0]["celsius"] == 26

    registry.delete_user_mode(created.id, "denza_n7")
    registry.delete_user_mode(fan_mode.id, "denza_n7")
    registry.delete_user_mode("office", "denza_n7")
    after_delete = json.loads(catalog_path.read_text(encoding="utf-8"))
    assert [scene["id"] for scene in after_delete["scenes"]] == ["office"]
    assert after_delete["scenes"][0]["label"] == "办公模式"
    assert after_delete["scenes"][0]["origin"] == "preset"


def test_repository_presets_are_exported_without_internal_command_flows() -> None:
    registry = ModeProfileRegistry(REPOSITORY_ROOT / "config" / "modes")

    catalog = build_mode_scene_catalog(
        registry,
        "denza_n7",
        generated_at="2026-09-09T00:00:00Z",
    )

    assert catalog["scene_count"] >= 13
    assert len({scene["id"] for scene in catalog["scenes"]}) == catalog["scene_count"]
    by_id = {scene["id"]: scene for scene in catalog["scenes"]}
    assert by_id["office"]["operation_count"] == 10
    assert by_id["rocking_horse"]["behavior"] == "cyclic"
    assert by_id["rocking_horse"]["timeline"]["repeat"] == "until_stopped_or_timeout"
    encoded = json.dumps(catalog, ensure_ascii=False)
    assert "command_flow" not in encoded
    assert "source_request_id" not in encoded