Miayui
364 lines
12 KiB
Python
364 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import tempfile
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from cognitive_agent.application.extension_control import (
|
|
ExtensionControlAuthorizationError,
|
|
ExtensionControlConsumer,
|
|
ExtensionControlError,
|
|
authorize_extension_control_actions,
|
|
parse_extension_control_payload,
|
|
)
|
|
from cognitive_agent.application.self_extension import (
|
|
ExtensionIntegrationError,
|
|
SelfExtensionService,
|
|
SimulationGateway,
|
|
)
|
|
|
|
|
|
def wait_until(predicate, timeout: float = 1.0) -> bool:
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
if predicate():
|
|
return True
|
|
time.sleep(0.005)
|
|
return bool(predicate())
|
|
|
|
|
|
def action(
|
|
slot: int,
|
|
action_name: str,
|
|
*,
|
|
target_scope: str = "toy-light-001",
|
|
at_ms: int = 0,
|
|
**extra: Any,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"type": "infrared_emit",
|
|
"at_ms": at_ms,
|
|
"domain": "self_extension",
|
|
"capability": "infrared.emit",
|
|
"target_scope": target_scope,
|
|
"operation": "emit",
|
|
"parameters": {"action": action_name, "slot": slot},
|
|
**extra,
|
|
}
|
|
|
|
|
|
def request(*actions: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"domain": "self_extension",
|
|
"operation_count": len(actions),
|
|
"actions": list(actions),
|
|
}
|
|
|
|
|
|
def capability(
|
|
capability_id: str,
|
|
slot: int,
|
|
action_name: str,
|
|
*,
|
|
device_id: str = "toy-light-001",
|
|
enabled: bool = True,
|
|
card_uid: str = "E113B203",
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"id": capability_id,
|
|
"status": "learned",
|
|
"enabled": enabled,
|
|
"device_id": device_id,
|
|
"device_name": "闪烁灯",
|
|
"card_uid": card_uid,
|
|
"action": action_name,
|
|
"slot": slot,
|
|
"risk": "low",
|
|
}
|
|
|
|
|
|
def encoded(payload: dict[str, Any]) -> bytes:
|
|
return json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
|
|
|
|
|
def test_parser_accepts_scene_action_shape_and_strips_compatibility_fields() -> None:
|
|
payload = request(
|
|
action(
|
|
2,
|
|
"打开",
|
|
target={"device": "toy-light-001", "card_uid": "E113B203"},
|
|
operation_ref="toy-light-on",
|
|
)
|
|
)
|
|
|
|
parsed = parse_extension_control_payload(encoded(payload))
|
|
|
|
assert parsed == request(action(2, "打开"))
|
|
assert "target" not in parsed["actions"][0]
|
|
assert "operation_ref" not in parsed["actions"][0]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("payload", "message"),
|
|
[
|
|
(
|
|
{
|
|
"domain": "self_extension",
|
|
"operation_count": 2,
|
|
"actions": [action(2, "打开")],
|
|
},
|
|
"must equal",
|
|
),
|
|
(request({**action(2, "打开"), "domain": "seat"}), "self_extension"),
|
|
(request({**action(2, "打开"), "operation": "set"}), "must be emit"),
|
|
(request(action(2, "打开", target_scope="闪烁灯")), "English device id"),
|
|
],
|
|
)
|
|
def test_parser_rejects_invalid_control_contract(
|
|
payload: dict[str, Any],
|
|
message: str,
|
|
) -> None:
|
|
with pytest.raises(ExtensionControlError, match=message):
|
|
parse_extension_control_payload(encoded(payload))
|
|
|
|
|
|
def test_parser_rejects_duplicate_json_keys() -> None:
|
|
raw = (
|
|
b'{"domain":"self_extension","domain":"self_extension",'
|
|
b'"operation_count":1,"actions":[]}'
|
|
)
|
|
|
|
with pytest.raises(ExtensionControlError, match="duplicate key"):
|
|
parse_extension_control_payload(raw)
|
|
|
|
|
|
def test_authorization_requires_exact_enabled_learned_bound_ability() -> None:
|
|
parsed = parse_extension_control_payload(encoded(request(action(2, "打开"))))
|
|
valid = capability("light-on", 2, "打开")
|
|
|
|
[authorized] = authorize_extension_control_actions(parsed, [valid])
|
|
|
|
assert authorized["capability_id"] == "light-on"
|
|
assert authorized["action"] == action(2, "打开")
|
|
for disallowed in (
|
|
[{**valid, "enabled": False}],
|
|
[{**valid, "status": "draft"}],
|
|
[{**valid, "card_uid": ""}],
|
|
[{**valid, "device_id": "another-device"}],
|
|
[{**valid, "action": "关闭"}],
|
|
[{**valid, "slot": 3}],
|
|
):
|
|
with pytest.raises(ExtensionControlAuthorizationError, match="not authorized"):
|
|
authorize_extension_control_actions(parsed, disallowed)
|
|
|
|
|
|
def test_consumer_validates_entire_request_then_executes_by_at_ms() -> None:
|
|
catalog = [
|
|
capability("light-on", 2, "打开"),
|
|
capability("light-red", 3, "切换为红色"),
|
|
]
|
|
executions: list[tuple[float, dict[str, Any]]] = []
|
|
|
|
def execute(value: dict[str, Any]) -> dict[str, Any]:
|
|
executions.append((time.monotonic(), value))
|
|
return {"success": True, "transport": "tcp", "gateway_acknowledged": True}
|
|
|
|
consumer = ExtensionControlConsumer(
|
|
catalog_provider=lambda: catalog,
|
|
executor=execute,
|
|
adjacent_emit_interval_s=0,
|
|
)
|
|
consumer.start()
|
|
try:
|
|
consumer.submit(
|
|
encoded(
|
|
request(
|
|
action(2, "打开", at_ms=30),
|
|
action(3, "切换为红色", at_ms=0),
|
|
)
|
|
)
|
|
)
|
|
assert wait_until(lambda: consumer.snapshot()["state"] == "completed")
|
|
assert [item[1]["parameters"]["slot"] for item in executions] == [3, 2]
|
|
status = consumer.snapshot()
|
|
assert status["processed_count"] == 1
|
|
assert status["executed_action_count"] == 2
|
|
assert [item["slot"] for item in status["last_results"]] == [3, 2]
|
|
assert [item["slot"] for item in status["action_statuses"]] == [3, 2]
|
|
assert [item["state"] for item in status["action_statuses"]] == [
|
|
"submitted_unverified",
|
|
"submitted_unverified",
|
|
]
|
|
assert all(
|
|
item["state_label"] == "已下发 · 物理效果未确认"
|
|
for item in status["action_statuses"]
|
|
)
|
|
assert status["source_label"] == "车端 MQTT"
|
|
assert status["active_action_index"] is None
|
|
assert status["last_error"] is None
|
|
finally:
|
|
consumer.close()
|
|
|
|
|
|
def test_consumer_rejects_whole_request_before_first_tcp_command() -> None:
|
|
executions: list[dict[str, Any]] = []
|
|
consumer = ExtensionControlConsumer(
|
|
catalog_provider=lambda: [capability("light-on", 2, "打开")],
|
|
executor=lambda value: executions.append(value) or {"success": True},
|
|
adjacent_emit_interval_s=0,
|
|
)
|
|
consumer.start()
|
|
try:
|
|
consumer.submit(
|
|
encoded(
|
|
request(
|
|
action(2, "打开"),
|
|
action(3, "切换为红色"),
|
|
)
|
|
)
|
|
)
|
|
assert wait_until(lambda: consumer.snapshot()["state"] == "rejected")
|
|
assert executions == []
|
|
assert consumer.snapshot()["rejected_count"] == 1
|
|
assert "not authorized" in consumer.snapshot()["last_error"]
|
|
finally:
|
|
consumer.close()
|
|
|
|
|
|
def test_consumer_aborts_remaining_actions_after_tcp_failure() -> None:
|
|
catalog = [
|
|
capability("light-on", 2, "打开"),
|
|
capability("light-red", 3, "切换为红色"),
|
|
capability("light-blue", 5, "切换为蓝色"),
|
|
]
|
|
attempted_slots: list[int] = []
|
|
|
|
def execute(value: dict[str, Any]) -> dict[str, Any]:
|
|
slot = int(value["parameters"]["slot"])
|
|
attempted_slots.append(slot)
|
|
if slot == 3:
|
|
raise RuntimeError("TCP disconnected")
|
|
return {"success": True, "transport": "tcp"}
|
|
|
|
consumer = ExtensionControlConsumer(
|
|
catalog_provider=lambda: catalog,
|
|
executor=execute,
|
|
adjacent_emit_interval_s=0,
|
|
)
|
|
consumer.start()
|
|
try:
|
|
consumer.submit(
|
|
encoded(
|
|
request(
|
|
action(2, "打开"),
|
|
action(3, "切换为红色"),
|
|
action(5, "切换为蓝色"),
|
|
)
|
|
)
|
|
)
|
|
assert wait_until(lambda: consumer.snapshot()["state"] == "failed")
|
|
assert attempted_slots == [2, 3]
|
|
status = consumer.snapshot()
|
|
assert status["failed_count"] == 1
|
|
assert status["executed_action_count"] == 1
|
|
assert [item["state"] for item in status["action_statuses"]] == [
|
|
"submitted_unverified",
|
|
"failed",
|
|
"aborted",
|
|
]
|
|
assert status["last_error"] == "TCP disconnected"
|
|
finally:
|
|
consumer.close()
|
|
|
|
|
|
def test_consumer_projects_mqtt_transport_rejection_without_execution() -> None:
|
|
executions: list[dict[str, Any]] = []
|
|
consumer = ExtensionControlConsumer(
|
|
catalog_provider=list,
|
|
executor=lambda value: executions.append(value) or {"success": True},
|
|
)
|
|
consumer.start()
|
|
try:
|
|
consumer.reject_transport_message(
|
|
"retained extension/ctl messages are not executable",
|
|
"retained",
|
|
)
|
|
|
|
status = consumer.snapshot()
|
|
assert status["state"] == "rejected"
|
|
assert status["request_count"] == 1
|
|
assert status["processed_count"] == 1
|
|
assert status["rejected_count"] == 1
|
|
assert status["last_rejection_kind"] == "retained"
|
|
assert status["last_latency_ms"] == 0
|
|
assert status["action_statuses"] == []
|
|
assert executions == []
|
|
finally:
|
|
consumer.close()
|
|
|
|
|
|
def test_self_extension_reauthorizes_immediately_before_gateway_emit() -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
service = SelfExtensionService(
|
|
SimulationGateway(),
|
|
catalog_db_path=Path(directory, "infrared.sqlite3"),
|
|
)
|
|
service.sync_peripheral_catalog([capability("light-on", 2, "打开")])
|
|
with patch.object(
|
|
service,
|
|
"execute_gateway",
|
|
return_value={
|
|
"success": True,
|
|
"transport": "tcp",
|
|
"gateway_acknowledged": True,
|
|
},
|
|
) as execute_gateway:
|
|
result = service.execute_bound_infrared_action(action(2, "打开"))
|
|
execute_gateway.assert_called_once_with("emit", {"slot": 2})
|
|
assert result["authorization"]["capability_id"] == "light-on"
|
|
|
|
with pytest.raises(ExtensionIntegrationError, match="未通过"):
|
|
service.execute_bound_infrared_action(action(3, "切换为红色"))
|
|
assert execute_gateway.call_count == 1
|
|
service.close()
|
|
|
|
|
|
def test_server_wires_extension_control_subscription_and_health() -> None:
|
|
root = Path(__file__).resolve().parents[2]
|
|
server = (root / "src" / "cognitive_agent" / "server.py").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
seat_service = (
|
|
root / "src" / "cognitive_agent" / "application" / "seat_service.py"
|
|
).read_text(encoding="utf-8")
|
|
console = (root / "apps" / "web-console" / "console-v2.js").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
|
|
assert "extension_control = ExtensionControlConsumer(" in server
|
|
assert "executor=extension.execute_bound_infrared_action" in server
|
|
assert "service.set_extension_control_handler(extension_control.submit)" in server
|
|
assert "extension_control.reject_transport_message" in server
|
|
assert '"extension_control": self.extension_control.snapshot()' in server
|
|
assert 'path == "/api/extension/control/status"' in server
|
|
assert "self.extension_control.wait_for_snapshot(" in server
|
|
assert "SeatDemoHandler.extension_control.start()" in server
|
|
assert "SeatDemoHandler.extension_control.close()" in server
|
|
assert "def set_extension_control_handler" in seat_service
|
|
assert "def set_extension_control_rejection_handler" in seat_service
|
|
assert 'fetch(`/api/extension/control/status?${query}`' in console
|
|
assert "function applyExtensionControlSnapshot" in console
|
|
assert "if (revision < extensionControlSync.revision)" in console
|
|
assert 'vehicle_mqtt_extension: "车端 MQTT"' in console
|
|
assert 'delivery_transport: "车端 MQTT → 红外 TCP"' in console
|
|
assert 'semantic_instruction: `${extensionName} · ${actionName}`' in console
|
|
assert (
|
|
'category: `${extensionName} · target_scope=${targetScope} · at_ms=${atMs}`'
|
|
in console
|
|
)
|