Cognitive_Agent/tests/unit/test_mode_profile_augmentation_intent.py

from __future__ import annotations

from copy import deepcopy

import pytest

from cognitive_agent.application.codex_assistant import (
    CodexAssistantService,
    _authorize_vehicle_request,
    _validated_mode_profile_draft_context,
)
from cognitive_agent.application.mode_profile_intent import (
    is_mode_profile_update_shape,
    parse_mode_profile_update,
)

DRAFT_REFERENCE = {
    "name": "躺平模式", "source_request_id": "req-augmentation-source-123456", "vehicle_id": "denza_n7",
}
CONTROL_TERMS = ("空调", "氛围灯", "阅读灯", "车窗", "座椅")


@pytest.mark.parametrize("prefix", ["在", "给", "为", "往", "向", ""])
@pytest.mark.parametrize("verb", ["加入", "添加", "增加", "补充", "加上", "追加", "新增"])
def test_named_mode_device_augmentation_is_recognized(prefix, verb):
    message = f"{prefix}躺平模式中{verb}空调、氛围灯、阅读灯、车窗等设备动作"
    assert parse_mode_profile_update(message) == ("躺平模式", f"{verb}空调、氛围灯、阅读灯、车窗等设备动作")
    assert _validated_mode_profile_draft_context(DRAFT_REFERENCE, message=message) == DRAFT_REFERENCE
    authorization = _authorize_vehicle_request(message, CONTROL_TERMS, DRAFT_REFERENCE)
    assert authorization.instruction == message
    assert authorization.reason == "explicit_mode_profile_draft_update"


@pytest.mark.parametrize(("message", "change"), [
    ("请在躺平模式里补充空调", "补充空调"),
    ("帮我给躺平模式添加空调", "添加空调"),
    ("为躺平模式新增阅读灯", "新增阅读灯"),
    ("躺平模式再补充车窗", "补充车窗"),
    ("躺平模式中再加入阅读灯", "再加入阅读灯"),
    ("在躺平模式中加入空调24度、阅读灯打开", "加入空调24度、阅读灯打开"),
    ("把空调24度、阅读灯打开加入到躺平模式中", "加入空调24度、阅读灯打开"),
    ("将车窗、氛围灯追加进躺平模式", "追加车窗、氛围灯"),
    ("请把空调加入躺平模式吧", "加入空调"),
    ("把副驾空调温度25.5度补充到躺平模式", "补充副驾空调温度25.5度"),
    ("在“躺平模式”中添加空调", "添加空调"),
])
def test_append_word_order_preserves_device_parameters_and_scoped_name(message, change):
    parsed = parse_mode_profile_update(message)
    assert parsed is not None and parsed[0] == "躺平模式"
    # A leading 再 modifier is harmless; neither parameters nor target words
    # may be lost when the reverse-order request is normalized.
    assert parsed[1].removeprefix("再") == change.removeprefix("再")


@pytest.mark.parametrize("message", [
    "不要在躺平模式中加入空调", "在躺平模式中不要加入空调",
    "躺平模式中先不要补充阅读灯", "给躺平模式添加空调，但不要执行",
    "给躺平模式添加空调，算了", "给躺平模式添加空调，撤销",
    "躺平模式中取消加入空调", "如何在躺平模式中加入空调？",
    "在躺平模式中加入空调好吗？", "在躺平模式中能否加入空调",
    "把空调加入到躺平模式会怎样？", "解释在躺平模式中加入空调这句话",
    "在躺平模式中加入哪些设备？", "引用在躺平模式中加入空调",
    "在空调出风模式中加入阅读灯", "在副驾按摩模式中添加车窗",
    "在躺平模式中加入", "给躺平模式补充",
])
def test_negated_read_only_or_device_property_requests_do_not_authorize_augmentation(message):
    assert parse_mode_profile_update(message) is None
    assert _validated_mode_profile_draft_context(DRAFT_REFERENCE, message=message) is None
    authorization = _authorize_vehicle_request(message, CONTROL_TERMS, DRAFT_REFERENCE)
    assert authorization.reason not in {"explicit_mode_profile_draft_update", "explicit_mode_profile_update"}


def test_other_named_mode_never_borrows_the_selected_draft_reference():
    message = "在阅读模式中加入空调"
    assert parse_mode_profile_update(message) == ("阅读模式", "加入空调")
    assert _validated_mode_profile_draft_context(DRAFT_REFERENCE, message=message) is None
    assert _authorize_vehicle_request(message, CONTROL_TERMS, DRAFT_REFERENCE).reason == "explicit_mode_profile_update"


def test_augmentation_reference_cannot_include_browser_supplied_device_actions():
    reference = {**DRAFT_REFERENCE, "operations": [{"capability": "seat.anything"}]}
    with pytest.raises(ValueError, match="fields are invalid"):
        _validated_mode_profile_draft_context(reference, message="在躺平模式中加入空调")


def test_speech_update_shape_preserves_named_augmentation_even_when_negated():
    assert is_mode_profile_update_shape("在躺平模式中加入空调，但不要执行")
    assert parse_mode_profile_update("在躺平模式中加入空调，但不要执行") is None


def test_device_augmentation_bypasses_general_agent_and_partial_control_paths():
    client = type("UnavailableClient", (), {"available": False})()
    assistant = CodexAssistantService(client)
    message = "在躺平模式中加入空调、氛围灯、阅读灯、车窗等设备动作"
    calls = []
    events = []

    def execute_control(instruction, _progress):
        calls.append(instruction)
        return {
            "request_id": "req-revised-123456", "status": "succeeded", "execution_mode": "dry_run",
            "hardware_execution_allowed": False,
            "orchestration_route": "mode_profile_create_preview",
            "intent_ir": {"spec": {"actions": [{"domain": "climate"}]}},
            "mode_profile_draft": {
                "name": "躺平模式", "source_request_id": "req-revised-123456", "vehicle_id": "denza_n7",
                "instruction": "进入躺平模式", "operation_count": 1,
                "generation_source": "draft_revision", "draft_revision": 2,
            },
        }

    reference = deepcopy(DRAFT_REFERENCE)
    result = assistant.stream(
        message, conversation_id="conv-augmentation-intent-123456", principal="test",
        history=[], emit=events.append, execute_control=execute_control,
        preview_control=lambda *_args: pytest.fail("augmentation must not run partial hardware controls"),
        supplement_control=lambda *_args: pytest.fail("augmentation must not become a vehicle supplement"),
        mode_profile_draft_context=reference, control_terms=CONTROL_TERMS,
    )
    assert calls == [message]
    assert result["success"] is True
    assert result["result"]["mode_profile_draft"]["name"] == "躺平模式"
    assert reference == DRAFT_REFERENCE
    assert events[0]["route"] == "vehicle_control"