vee
77 lines
2.6 KiB
Python
77 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
import threading
|
|
import unittest
|
|
|
|
from cognitive_agent.application.mode_execution import ModeExecutionStore
|
|
|
|
|
|
class ModeExecutionStoreTests(unittest.TestCase):
|
|
def test_service_failure_status_is_normalized_for_remote_clients(self) -> None:
|
|
store = ModeExecutionStore()
|
|
finished = threading.Event()
|
|
|
|
def run() -> dict[str, object]:
|
|
finished.set()
|
|
return {"status": "execution_failed", "request_id": "req-failed"}
|
|
|
|
execution, created = store.create(
|
|
vehicle_id="denza_n7",
|
|
mode={"id": "nap", "name": "小憩模式", "domains": ["seat"]},
|
|
execution_target="dry_run",
|
|
source="android",
|
|
source_device_id="n7-cockpit-test",
|
|
hardware_client_id=None,
|
|
idempotency_key="failure-normalization",
|
|
runner=run,
|
|
)
|
|
self.assertTrue(created)
|
|
self.assertTrue(finished.wait(1))
|
|
terminal = self._wait_terminal(store, execution["execution_id"])
|
|
self.assertEqual(terminal["status"], "failed")
|
|
self.assertEqual(terminal["result"]["status"], "execution_failed")
|
|
|
|
def test_idempotency_key_returns_the_original_execution(self) -> None:
|
|
store = ModeExecutionStore()
|
|
release = threading.Event()
|
|
|
|
def run() -> dict[str, object]:
|
|
release.wait(1)
|
|
return {"status": "succeeded", "request_id": "req-ok"}
|
|
|
|
arguments = {
|
|
"vehicle_id": "denza_n7",
|
|
"mode": {"id": "nap", "name": "小憩模式", "domains": ["seat"]},
|
|
"execution_target": "dry_run",
|
|
"source": "android",
|
|
"source_device_id": "n7-cockpit-test",
|
|
"hardware_client_id": None,
|
|
"idempotency_key": "same-tap",
|
|
"runner": run,
|
|
}
|
|
first, first_created = store.create(**arguments)
|
|
second, second_created = store.create(**arguments)
|
|
release.set()
|
|
|
|
self.assertTrue(first_created)
|
|
self.assertFalse(second_created)
|
|
self.assertEqual(first["execution_id"], second["execution_id"])
|
|
|
|
@staticmethod
|
|
def _wait_terminal(
|
|
store: ModeExecutionStore,
|
|
execution_id: str,
|
|
) -> dict[str, object]:
|
|
for _ in range(100):
|
|
execution = store.get(execution_id)
|
|
if execution["status"] in {
|
|
"succeeded", "partially_succeeded", "failed", "cancelled", "timed_out"
|
|
}:
|
|
return execution
|
|
threading.Event().wait(0.01)
|
|
raise AssertionError("mode execution did not reach a terminal status")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|