vee
124 lines
4.7 KiB
Python
124 lines
4.7 KiB
Python
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from cognitive_agent.application.native_can_relative import NativeCanRelativeResolver
|
|
from cognitive_agent.application.seat_service import REPOSITORY_ROOT, SeatDemoService
|
|
|
|
|
|
class CanTestBranchFusionTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.service = SeatDemoService(REPOSITORY_ROOT)
|
|
self.temporary_directory = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temporary_directory.cleanup)
|
|
self.service.artifact_root = Path(self.temporary_directory.name)
|
|
|
|
def test_negated_vehicle_control_never_compiles_or_executes(self) -> None:
|
|
result = self.service.process(
|
|
"不要打开左前车窗",
|
|
provider="local",
|
|
conversation_id="conv-negative-window",
|
|
)
|
|
|
|
self.assertEqual(result["status"], "no_action")
|
|
spec = result["intent_ir"]["spec"]
|
|
self.assertEqual(spec["actions"], [])
|
|
self.assertEqual(spec["constraints"][0]["capability"], "window.set_motion")
|
|
self.assertFalse(spec["execution_eligible"])
|
|
self.assertEqual(result["function_configs"], {})
|
|
|
|
def test_multiple_windows_and_exclusion_are_deterministic(self) -> None:
|
|
multiple = self.service.process(
|
|
"打开左前和右前车窗",
|
|
provider="local",
|
|
conversation_id="conv-two-windows",
|
|
)
|
|
self.assertEqual(multiple["status"], "succeeded")
|
|
self.assertEqual(
|
|
[item["target"]["zone"] for item in multiple["intent_ir"]["spec"]["actions"]],
|
|
["front_left", "front_right"],
|
|
)
|
|
|
|
excluded = self.service.process(
|
|
"打开全部车窗，左前除外",
|
|
provider="local",
|
|
conversation_id="conv-excluded-window",
|
|
)
|
|
self.assertEqual(excluded["status"], "succeeded")
|
|
self.assertEqual(
|
|
[item["target"]["zone"] for item in excluded["intent_ir"]["spec"]["actions"]],
|
|
["front_right", "rear_left", "rear_right"],
|
|
)
|
|
self.assertEqual(
|
|
excluded["intent_ir"]["spec"]["constraints"][0]["type"],
|
|
"exclude_target",
|
|
)
|
|
|
|
def test_unresolved_mixed_control_fails_closed_before_compile(self) -> None:
|
|
result = self.service.process(
|
|
"打开左前车窗，再把座椅调舒服点",
|
|
provider="local",
|
|
conversation_id="conv-unresolved-mixed",
|
|
)
|
|
|
|
self.assertEqual(result["status"], "needs_confirmation")
|
|
self.assertFalse(result["intent_ir"]["spec"]["execution_eligible"])
|
|
self.assertEqual(result["function_configs"], {})
|
|
self.assertIn(
|
|
"UNRESOLVED_UTTERANCE_SPAN",
|
|
{item["code"] for item in result["validation"]},
|
|
)
|
|
|
|
def test_relative_fan_uses_current_state_and_records_resolution(self) -> None:
|
|
result = self.service.process(
|
|
"把风量提高两档",
|
|
provider="local",
|
|
conversation_id="conv-relative-fan",
|
|
)
|
|
|
|
self.assertEqual(result["status"], "succeeded")
|
|
action = result["intent_ir"]["spec"]["actions"][0]
|
|
self.assertEqual(action["capability"], "climate.set_fan_level")
|
|
self.assertEqual(action["parameters"], {"level": 2})
|
|
self.assertEqual(action["resolution"]["current"], 0)
|
|
self.assertEqual(action["resolution"]["delta"], 2)
|
|
|
|
def test_relative_vehicle_control_requires_physical_feedback_in_hardware_context(self) -> None:
|
|
resolver = NativeCanRelativeResolver(REPOSITORY_ROOT)
|
|
context = self.service.conversations.snapshot("conv-relative-physical")
|
|
context["domain_devices"]["vehicle_status"]["feedback_source"] = "physical_can_ch1"
|
|
|
|
result = resolver.resolve("把风量提高一档", context)
|
|
|
|
self.assertEqual(result.actions, [])
|
|
self.assertEqual(
|
|
result.ambiguities[0]["reason"],
|
|
"native_can_relative_state_not_physical",
|
|
)
|
|
|
|
def test_reasoning_effort_reaches_openai_payload(self) -> None:
|
|
self.service.openai.api_key = "test-key"
|
|
plan = {
|
|
"detected_domains": ["window"],
|
|
"requests": [],
|
|
"ambiguities": [],
|
|
"decision_summary": ["no action"],
|
|
"confidence": 0.9,
|
|
}
|
|
with patch.object(self.service.openai, "interpret", return_value=plan) as interpret:
|
|
self.service.process(
|
|
"你好",
|
|
provider="openai",
|
|
conversation_id="conv-reasoning-effort",
|
|
reasoning_effort="high",
|
|
)
|
|
|
|
self.assertEqual(interpret.call_args.kwargs["reasoning_effort"], "high")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|