335 lines
13 KiB
Python
335 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from agent_call.core import (
|
|
AgentCallService,
|
|
ConfigurationError,
|
|
ConflictError,
|
|
InMemoryBroker,
|
|
Store,
|
|
ValidationError,
|
|
iso,
|
|
utcnow,
|
|
)
|
|
from deploy.backup_sqlite import copy_database
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def fixture() -> dict:
|
|
return json.loads(
|
|
(ROOT / "docs/contracts/examples/call.execute.json").read_text(encoding="utf-8")
|
|
)
|
|
|
|
|
|
class RemediationTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.broker = InMemoryBroker()
|
|
self.service = AgentCallService(
|
|
Path(self.temp.name) / "executor.sqlite3",
|
|
object_dir=Path(self.temp.name) / "objects",
|
|
broker=self.broker,
|
|
)
|
|
self.service.profile["limits"]["tenant_cps"] = 100
|
|
self.service.profile["limits"]["global_cps"] = 100
|
|
|
|
def tearDown(self) -> None:
|
|
self.service.stop()
|
|
self.temp.cleanup()
|
|
|
|
def command(
|
|
self,
|
|
command_id: str,
|
|
tenant_id: str = "tenant-demo",
|
|
execution_id: str = "exec",
|
|
) -> dict:
|
|
body = copy.deepcopy(fixture())
|
|
body["command_id"] = command_id
|
|
body["trace_id"] = f"trace-{command_id}"
|
|
body["issued_at"] = iso(utcnow())
|
|
body["tenant_id"] = tenant_id
|
|
tenant = self.service.store.one(
|
|
"SELECT tenant_key FROM tenants WHERE tenant_id=?", (tenant_id,)
|
|
)
|
|
body["tenant_key"] = tenant["tenant_key"]
|
|
body["payload"]["execution_id"] = execution_id
|
|
body["payload"]["task_id"] = (
|
|
"task-demo" if tenant_id == "tenant-demo" else "task-b"
|
|
)
|
|
return body
|
|
|
|
def test_runtime_schema_rejects_naive_time_and_boolean_revision(self) -> None:
|
|
body = self.command("schema-command")
|
|
body["issued_at"] = "2026-09-13T00:00:00"
|
|
with self.assertRaises(ValidationError):
|
|
self.service.publish_execute(body)
|
|
with self.assertRaises(ValidationError):
|
|
self.service.control_task(
|
|
"tenant-demo",
|
|
"task-demo",
|
|
{
|
|
"command_id": "bool-revision",
|
|
"action": "pause",
|
|
"expected_task_revision": True,
|
|
"reason": "x",
|
|
},
|
|
"bool-revision",
|
|
)
|
|
|
|
def test_readiness_requires_persistent_database_write(self) -> None:
|
|
db_path = Path(self.temp.name) / "executor.sqlite3"
|
|
os.chmod(db_path, 0o444)
|
|
try:
|
|
readiness = self.service.readiness()
|
|
finally:
|
|
os.chmod(db_path, 0o644)
|
|
self.assertFalse(readiness["checks"]["database_write"])
|
|
self.assertFalse(readiness["ready"])
|
|
|
|
def test_provider_modes_do_not_fallback(self) -> None:
|
|
profile = json.loads((ROOT / "docs/contracts/mock-profile.json").read_text())
|
|
profile["provider_modes"].pop("llm")
|
|
path = Path(self.temp.name) / "profile.json"
|
|
path.write_text(json.dumps(profile), encoding="utf-8")
|
|
with self.assertRaises(ConfigurationError):
|
|
AgentCallService(
|
|
profile_path=path, object_dir=Path(self.temp.name) / "objects2"
|
|
)
|
|
|
|
def test_same_public_ids_are_tenant_scoped_and_wrong_route_is_dead_lettered(
|
|
self,
|
|
) -> None:
|
|
first = self.command("same-command", "tenant-demo", "same-execution")
|
|
second = self.command("same-command", "tenant-b", "same-execution")
|
|
self.service.publish_execute(first)
|
|
self.service.publish_execute(second)
|
|
self.assertEqual(self.service.ingest_commands(3), 2)
|
|
self.assertEqual(self.service.schedule_once(), 2)
|
|
self.service.wait_for_idle()
|
|
self.assertEqual(
|
|
self.service.store.one("SELECT COUNT(*) AS n FROM calls")["n"], 2
|
|
)
|
|
|
|
wrong = self.command("wrong-route", "tenant-b", "wrong-execution")
|
|
self.broker.publish(
|
|
"agent-call.commands.v1",
|
|
"agent-call.tenant.tenant-demo-key.call.execute",
|
|
wrong,
|
|
"wrong-route",
|
|
)
|
|
self.service._cursor = self.service._tenant_order.index("tenant-demo")
|
|
self.service.ingest_commands(1)
|
|
self.assertEqual(
|
|
self.service.store.one("SELECT COUNT(*) AS n FROM dead_letters")["n"], 1
|
|
)
|
|
self.assertIsNone(
|
|
self.service.store.one(
|
|
"SELECT * FROM calls WHERE command_id=?", ("wrong-route",)
|
|
)
|
|
)
|
|
|
|
def test_over_budget_tenant_is_retained_without_truncation(self) -> None:
|
|
profile = json.loads((ROOT / "docs/contracts/mock-profile.json").read_text())
|
|
key = "x" * 225
|
|
profile["tenants"] = [
|
|
{"tenant_id": "long", "tenant_key": key, "max_concurrency": 1}
|
|
]
|
|
profile["tasks"] = [
|
|
{
|
|
"task_id": "long-task",
|
|
"tenant_id": "long",
|
|
"state": "running",
|
|
"revision": 1,
|
|
}
|
|
]
|
|
path = Path(self.temp.name) / "long-profile.json"
|
|
path.write_text(json.dumps(profile), encoding="utf-8")
|
|
service = AgentCallService(
|
|
Path(self.temp.name) / "long.sqlite3",
|
|
profile_path=path,
|
|
object_dir=Path(self.temp.name) / "long-objects",
|
|
broker=InMemoryBroker(),
|
|
)
|
|
try:
|
|
body = self.command("placeholder")
|
|
body["tenant_id"], body["tenant_key"] = "long", key
|
|
body["payload"]["task_id"] = "long-task"
|
|
body["payload"]["execution_id"] = "long-execution"
|
|
with self.assertRaises(ValidationError) as error:
|
|
service.publish_execute(body)
|
|
self.assertEqual(error.exception.code, "TENANT_KEY_UNSUPPORTED")
|
|
row = service.store.one(
|
|
"SELECT status FROM published_commands WHERE tenant_id=?", ("long",)
|
|
)
|
|
self.assertEqual(row["status"], "retained")
|
|
finally:
|
|
service.stop()
|
|
|
|
def test_control_barrier_blocks_resume_until_all_cells_apply(self) -> None:
|
|
pause = {
|
|
"command_id": "pause-1",
|
|
"action": "pause",
|
|
"expected_task_revision": 1,
|
|
"reason": "maintenance",
|
|
}
|
|
self.service.set_cell_health("cell-a", False)
|
|
self.service.control_task("tenant-demo", "task-demo", pause, "pause-1")
|
|
self.assertEqual(self.service.apply_controls(), 0)
|
|
with self.assertRaises(ConflictError) as error:
|
|
self.service.control_task(
|
|
"tenant-demo",
|
|
"task-demo",
|
|
{
|
|
"command_id": "resume-1",
|
|
"action": "resume",
|
|
"expected_task_revision": 2,
|
|
"reason": "resume",
|
|
},
|
|
"resume-1",
|
|
)
|
|
self.assertEqual(error.exception.code, "CONTROL_IN_PROGRESS")
|
|
self.service.set_cell_health("cell-a", True)
|
|
self.assertEqual(self.service.apply_controls(), 1)
|
|
self.assertEqual(
|
|
self.service.store.one(
|
|
"SELECT COUNT(*) AS n FROM control_barriers WHERE command_id=? AND status='applied'",
|
|
("pause-1",),
|
|
)["n"],
|
|
2,
|
|
)
|
|
|
|
def test_final_transcript_is_idempotent_and_conflicts_are_preserved(self) -> None:
|
|
self.service.profile["limits"]["hold_ms"] = 1000
|
|
body = self.command("transcript-command", "tenant-demo", "transcript-execution")
|
|
self.service.publish_execute(body)
|
|
self.service.ingest_commands()
|
|
self.service.schedule_once()
|
|
deadline = time.monotonic() + 2
|
|
call_id = None
|
|
while time.monotonic() < deadline:
|
|
row = self.service.store.one(
|
|
"SELECT * FROM calls WHERE command_id=?", (body["command_id"],)
|
|
)
|
|
if row and row["started_at"]:
|
|
call_id = row["call_id"]
|
|
break
|
|
time.sleep(0.01)
|
|
self.assertIsNotNone(call_id)
|
|
assert call_id is not None
|
|
self.service._add_transcript(call_id, 99, "agent", "final text", "confirmed")
|
|
before = self.service.store.one(
|
|
"SELECT COUNT(*) AS n FROM events WHERE event_type='transcript.updated'"
|
|
)["n"]
|
|
self.service._add_transcript(call_id, 99, "agent", "final text", "confirmed")
|
|
after = self.service.store.one(
|
|
"SELECT COUNT(*) AS n FROM events WHERE event_type='transcript.updated'"
|
|
)["n"]
|
|
self.assertEqual(before, after)
|
|
with self.assertRaises(ConflictError):
|
|
self.service._add_transcript(
|
|
call_id, 99, "agent", "different", "confirmed", revision=2
|
|
)
|
|
with self.assertRaises(ConflictError):
|
|
self.service._add_transcript(
|
|
call_id, 99, "agent", "different", "confirmed", revision=1
|
|
)
|
|
self.service._add_transcript(
|
|
call_id, 99, "agent", "final text", "sent", revision=3, is_final=False
|
|
)
|
|
segment = self.service.store.one(
|
|
"SELECT text, is_final, playback_state FROM transcripts WHERE segment_id=?",
|
|
(f"seg_{call_id}_99_agent",),
|
|
)
|
|
self.assertEqual(
|
|
(segment["text"], segment["is_final"], segment["playback_state"]),
|
|
("final text", 1, "confirmed"),
|
|
)
|
|
self.service._finish_call(call_id, "failed", "TEST_FINISHED")
|
|
|
|
def test_upload_rejects_metadata_changes(self) -> None:
|
|
body = self.command(
|
|
"recording-metadata-command", "tenant-demo", "recording-metadata-execution"
|
|
)
|
|
self.service.publish_execute(body)
|
|
self.service.wait_for_idle()
|
|
call_id = self.service.get_command("tenant-demo", body["command_id"])["call_id"]
|
|
recording = self.service.get_call("tenant-demo", call_id)["recordings"][0]
|
|
request = {
|
|
"recording_id": recording["recording_id"],
|
|
"call_id": call_id,
|
|
"content_type": "audio/wav",
|
|
"size_bytes": recording["size_bytes"],
|
|
"checksum_algorithm": "SHA-256",
|
|
"checksum": recording["checksum"],
|
|
"channels": 1,
|
|
"sample_rate_hz": recording["sample_rate_hz"],
|
|
"duration_ms": recording["duration_ms"] + 1,
|
|
}
|
|
with self.assertRaises(ConflictError) as error:
|
|
self.service.create_upload("tenant-demo", request, "metadata-change", "api")
|
|
self.assertEqual(error.exception.code, "RECORDING_METADATA_CONFLICT")
|
|
|
|
def test_recording_is_non_silent_and_upload_completion_is_idempotent(self) -> None:
|
|
body = self.command("recording-command", "tenant-demo", "recording-execution")
|
|
self.service.publish_execute(body)
|
|
self.service.wait_for_idle()
|
|
call = self.service.get_call(
|
|
"tenant-demo",
|
|
self.service.get_command("tenant-demo", body["command_id"])["call_id"],
|
|
)
|
|
recording = call["recordings"][0]
|
|
upload = self.service.store.one(
|
|
"SELECT upload_id FROM uploads WHERE tenant_id=? AND recording_id=? ORDER BY created_at DESC LIMIT 1",
|
|
("tenant-demo", recording["recording_id"]),
|
|
)
|
|
self.assertIsNotNone(upload)
|
|
assert upload is not None
|
|
path = self.service.object_store.final_path_for(upload["upload_id"])
|
|
self.assertTrue(any(path.read_bytes()))
|
|
request = {
|
|
"recording_id": recording["recording_id"],
|
|
"call_id": call["call_id"],
|
|
"content_type": "audio/wav",
|
|
"size_bytes": recording["size_bytes"],
|
|
"checksum_algorithm": "SHA-256",
|
|
"checksum": recording["checksum"],
|
|
"channels": 1,
|
|
"sample_rate_hz": recording["sample_rate_hz"],
|
|
"duration_ms": recording["duration_ms"],
|
|
}
|
|
status, first = self.service.create_upload(
|
|
"tenant-demo", request, "api-create", "api"
|
|
)
|
|
status2, second = self.service.create_upload(
|
|
"tenant-demo", request, "api-create", "api"
|
|
)
|
|
self.assertEqual((status, status2), (200, 200))
|
|
self.assertEqual(first["upload_id"], second["upload_id"])
|
|
self.assertEqual(
|
|
self.service.store.one(
|
|
"SELECT COUNT(*) AS n FROM recordings WHERE call_id=?",
|
|
(call["call_id"],),
|
|
)["n"],
|
|
1,
|
|
)
|
|
|
|
def test_backup_refuses_to_replace_existing_database(self) -> None:
|
|
source = Path(self.temp.name) / "source.sqlite3"
|
|
destination = Path(self.temp.name) / "backup.sqlite3"
|
|
Store(source)
|
|
copy_database(source, destination)
|
|
with self.assertRaises(FileExistsError):
|
|
copy_database(source, destination)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|