345 lines
12 KiB
Python
345 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import threading
|
|
import unittest
|
|
from http.client import HTTPConnection
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
from typing import Any
|
|
from urllib.parse import quote
|
|
|
|
from agent_call.cell_agent import (
|
|
AsteriskConfigApplier,
|
|
CellAgentError,
|
|
CellAgentService,
|
|
make_server,
|
|
)
|
|
from agent_call.sip_management import (
|
|
SipManagementError,
|
|
SipManagementService,
|
|
_validate_trunk,
|
|
)
|
|
|
|
|
|
class FakeApplier:
|
|
def __init__(self) -> None:
|
|
self.calls: list[tuple[str, dict[str, Any]]] = []
|
|
self.fail = False
|
|
|
|
def apply(self, trunk_id: str, config: dict[str, Any]) -> None:
|
|
if self.fail:
|
|
raise CellAgentError(502, "ASTERISK_RELOAD_FAILED", "reload failed")
|
|
self.calls.append((trunk_id, config))
|
|
|
|
|
|
class CellAgentTests(unittest.TestCase):
|
|
def test_revision_hash_idempotency_and_gap_protection(self) -> None:
|
|
with TemporaryDirectory() as directory:
|
|
applier = FakeApplier()
|
|
service = CellAgentService(
|
|
Path(directory) / "cell.sqlite3", "cell-a", applier
|
|
)
|
|
try:
|
|
body = self._publication("cell-a", "trunk:a", 1)
|
|
first = service.apply(body, "request-1")
|
|
second = service.apply(body, "request-1-retry")
|
|
self.assertFalse(first["idempotent"])
|
|
self.assertTrue(second["idempotent"])
|
|
self.assertEqual(len(applier.calls), 1)
|
|
with self.assertRaisesRegex(CellAgentError, "next revision"):
|
|
service.apply(
|
|
self._publication("cell-a", "trunk:a", 3), "request-3"
|
|
)
|
|
bad = self._publication("cell-a", "trunk:a", 2)
|
|
bad["config_sha256"] = "0" * 64
|
|
with self.assertRaisesRegex(CellAgentError, "does not match"):
|
|
service.apply(bad, "request-bad")
|
|
finally:
|
|
service.close()
|
|
|
|
def test_http_endpoint_unquotes_path_and_rejects_mismatch(self) -> None:
|
|
with TemporaryDirectory() as directory:
|
|
service = CellAgentService(
|
|
Path(directory) / "cell.sqlite3", "cell-a", FakeApplier()
|
|
)
|
|
server = make_server(service, "127.0.0.1", 0)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
body = self._publication("cell-a", "trunk:a", 1)
|
|
response = self._request(
|
|
server.server_port,
|
|
f"/v1/sip/trunks/{quote('trunk:a', safe='')}/apply",
|
|
body,
|
|
)
|
|
self.assertEqual(response.status, 200)
|
|
mismatch = self._publication("cell-a", "other", 2)
|
|
response = self._request(
|
|
server.server_port,
|
|
"/v1/sip/trunks/trunk%3Aa/apply",
|
|
mismatch,
|
|
)
|
|
self.assertEqual(response.status, 400)
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join(timeout=2)
|
|
service.close()
|
|
|
|
def test_asterisk_applier_writes_codec_and_restores_on_reload_failure(self) -> None:
|
|
with TemporaryDirectory() as directory:
|
|
|
|
class RecordingApplier(AsteriskConfigApplier):
|
|
def __init__(self, config_dir: str) -> None:
|
|
super().__init__(config_dir)
|
|
self.fail = False
|
|
|
|
def _reload(self) -> None:
|
|
if self.fail:
|
|
raise CellAgentError(
|
|
502, "ASTERISK_RELOAD_FAILED", "reload failed"
|
|
)
|
|
|
|
applier = RecordingApplier(directory)
|
|
config = self._config()
|
|
applier.apply("trunk:a", config)
|
|
target = next(Path(directory).glob("*.conf"))
|
|
original = target.read_text(encoding="utf-8")
|
|
self.assertIn("allow=alaw", original)
|
|
self.assertIn("from_user=BD93205882", original)
|
|
applier.fail = True
|
|
with self.assertRaises(CellAgentError):
|
|
applier.apply(
|
|
"trunk:a",
|
|
{
|
|
**config,
|
|
"codec_profile": {"allowed": ["PCMU"], "preferred": "PCMU"},
|
|
},
|
|
)
|
|
self.assertEqual(target.read_text(encoding="utf-8"), original)
|
|
|
|
def test_real_management_activates_only_after_cell_ack(self) -> None:
|
|
class Publisher:
|
|
def __init__(self) -> None:
|
|
self.fail = False
|
|
self.calls: list[int] = []
|
|
|
|
def publish(
|
|
self,
|
|
cell: dict[str, Any],
|
|
trunk_id: str,
|
|
revision: int,
|
|
config: dict[str, Any],
|
|
request_id: str,
|
|
) -> dict[str, Any]:
|
|
self.calls.append(revision)
|
|
if self.fail:
|
|
raise OSError("offline")
|
|
digest = hashlib.sha256(
|
|
json.dumps(
|
|
config,
|
|
ensure_ascii=False,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
).encode()
|
|
).hexdigest()
|
|
return {
|
|
"mode": "real",
|
|
"cell_id": cell["cell_id"],
|
|
"trunk_id": trunk_id,
|
|
"revision": revision,
|
|
"config_sha256": digest,
|
|
"status": "applied",
|
|
}
|
|
|
|
with TemporaryDirectory() as directory:
|
|
publisher = Publisher()
|
|
service = SipManagementService(
|
|
Path(directory) / "sip.sqlite3", mode="real", publisher=publisher
|
|
)
|
|
try:
|
|
service.upsert_cell(
|
|
"cell-a",
|
|
{
|
|
"egress_pool_id": "egress-a",
|
|
"codec_capabilities": ["PCMA"],
|
|
"status": "healthy",
|
|
"max_concurrency": 10,
|
|
"management_url": "https://cell-a.internal:9443",
|
|
},
|
|
expected_revision=0,
|
|
actor="ops",
|
|
request_id="cell-1",
|
|
)
|
|
service.upsert_trunk(
|
|
"trunk:a",
|
|
self._config(),
|
|
expected_revision=0,
|
|
actor="ops",
|
|
request_id="trunk-1",
|
|
)
|
|
published = service.publish_trunk(
|
|
"trunk:a", expected_revision=1, actor="ops", request_id="pub-1"
|
|
)
|
|
self.assertEqual(published["active_revision"], 1)
|
|
self.assertEqual(
|
|
service.list_publications("trunk:a")[0]["status"], "applied"
|
|
)
|
|
publisher.fail = True
|
|
service.upsert_trunk(
|
|
"trunk:a",
|
|
{**self._config(), "display_name": "Provider B"},
|
|
expected_revision=1,
|
|
actor="ops",
|
|
request_id="trunk-2",
|
|
)
|
|
with self.assertRaisesRegex(SipManagementError, "every Cell"):
|
|
service.publish_trunk(
|
|
"trunk:a", expected_revision=2, actor="ops", request_id="pub-2"
|
|
)
|
|
self.assertEqual(service.get_trunk("trunk:a")["active_revision"], 1)
|
|
self.assertEqual(
|
|
service.list_publications("trunk:a")[0]["status"], "applied"
|
|
)
|
|
self.assertEqual(
|
|
service.list_publications("trunk:a")[1]["status"], "failed"
|
|
)
|
|
finally:
|
|
service.close()
|
|
|
|
def test_real_publication_resumes_after_process_crash(self) -> None:
|
|
class CrashOncePublisher:
|
|
def __init__(self) -> None:
|
|
self.crashed = False
|
|
|
|
def publish(
|
|
self,
|
|
cell: dict[str, Any],
|
|
trunk_id: str,
|
|
revision: int,
|
|
config: dict[str, Any],
|
|
request_id: str,
|
|
) -> dict[str, Any]:
|
|
if not self.crashed:
|
|
self.crashed = True
|
|
raise KeyboardInterrupt
|
|
digest = hashlib.sha256(
|
|
json.dumps(
|
|
config,
|
|
ensure_ascii=False,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
).encode()
|
|
).hexdigest()
|
|
return {
|
|
"mode": "real",
|
|
"cell_id": cell["cell_id"],
|
|
"trunk_id": trunk_id,
|
|
"revision": revision,
|
|
"config_sha256": digest,
|
|
"status": "applied",
|
|
}
|
|
|
|
with TemporaryDirectory() as directory:
|
|
publisher = CrashOncePublisher()
|
|
service = SipManagementService(
|
|
Path(directory) / "sip.sqlite3", mode="real", publisher=publisher
|
|
)
|
|
try:
|
|
service.upsert_cell(
|
|
"cell-a",
|
|
{
|
|
"egress_pool_id": "egress-a",
|
|
"codec_capabilities": ["PCMA"],
|
|
"status": "healthy",
|
|
"max_concurrency": 10,
|
|
"management_url": "https://cell-a.internal:9443",
|
|
},
|
|
expected_revision=0,
|
|
actor="ops",
|
|
request_id="cell-1",
|
|
)
|
|
service.upsert_trunk(
|
|
"trunk:a",
|
|
self._config(),
|
|
expected_revision=0,
|
|
actor="ops",
|
|
request_id="trunk-1",
|
|
)
|
|
with self.assertRaises(KeyboardInterrupt):
|
|
service.publish_trunk(
|
|
"trunk:a", expected_revision=1, actor="ops", request_id="pub-1"
|
|
)
|
|
self.assertEqual(service.get_trunk("trunk:a")["active_revision"], 0)
|
|
resumed = service.publish_trunk(
|
|
"trunk:a", expected_revision=1, actor="ops", request_id="pub-2"
|
|
)
|
|
self.assertEqual(resumed["active_revision"], 1)
|
|
self.assertEqual(
|
|
service.list_publications("trunk:a")[0]["status"], "applied"
|
|
)
|
|
finally:
|
|
service.close()
|
|
|
|
@staticmethod
|
|
def _config() -> dict[str, Any]:
|
|
return {
|
|
"display_name": "Provider A",
|
|
"enabled": True,
|
|
"sip": {
|
|
"host": "61.132.228.221",
|
|
"port": 5060,
|
|
"transport": "udp",
|
|
"auth_mode": "ip",
|
|
"register": False,
|
|
},
|
|
"codec_profile": {"allowed": ["PCMA"], "preferred": "PCMA"},
|
|
"caller_ids": ["BD93205882"],
|
|
"dial_prefix": "7089",
|
|
"egress_pool_id": "egress-a",
|
|
"max_concurrency": 100,
|
|
"max_cps": 10,
|
|
}
|
|
|
|
@classmethod
|
|
def _publication(cls, cell_id: str, trunk_id: str, revision: int) -> dict[str, Any]:
|
|
config = _validate_trunk(trunk_id, cls._config())
|
|
digest = hashlib.sha256(
|
|
json.dumps(
|
|
config, ensure_ascii=False, sort_keys=True, separators=(",", ":")
|
|
).encode()
|
|
).hexdigest()
|
|
return {
|
|
"mode": "real",
|
|
"cell_id": cell_id,
|
|
"trunk_id": trunk_id,
|
|
"revision": revision,
|
|
"config": config,
|
|
"config_sha256": digest,
|
|
}
|
|
|
|
@staticmethod
|
|
def _request(port: int, path: str, body: dict[str, Any]):
|
|
encoded = json.dumps(body, ensure_ascii=False).encode()
|
|
connection = HTTPConnection("127.0.0.1", port, timeout=3)
|
|
connection.request(
|
|
"POST",
|
|
path,
|
|
body=encoded,
|
|
headers={
|
|
"Content-Type": "application/json",
|
|
"Content-Length": str(len(encoded)),
|
|
"X-Request-ID": "http-test",
|
|
},
|
|
)
|
|
response = connection.getresponse()
|
|
response.read()
|
|
connection.close()
|
|
return response
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|