import json from pathlib import Path import tempfile import unittest from unittest.mock import patch from tools.rtsp_test_manager import Manager, SenderError, LABEL, atomic_json, mount, video_info, wait_ready class SenderTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.root = Path(self.tmp.name) self.video = self.root / "video #1, name.mp4" self.video.write_bytes(b"fixture") self.manager = Manager(self.root / "registry", run=lambda args: "") def tearDown(self): self.tmp.cleanup() def probe(self): return patch("tools.rtsp_test_manager.command", return_value=json.dumps({ "streams": [{"codec_name": "mpeg4", "width": 1280, "height": 720}]})) def info(self): with self.probe(): return video_info(str(self.video)) def container(self, name="sample01", running=True): return {"Id": "a"*64, "Config": {"Labels": { f"{LABEL}.owner": self.manager.owner, f"{LABEL}.source": name}}, "State": {"Running": running}} def test_absolute_relative_and_special_path(self): with self.probe(): self.assertEqual(video_info(self.video.name, str(self.root))["path"], str(self.video)) self.assertEqual(video_info(str(self.video))["codec_name"], "mpeg4") with self.assertRaises(SenderError): video_info(self.video.name) with self.assertRaises(FileNotFoundError): video_info(str(self.root / "missing")) import csv self.assertIn(f"src={self.video}", next(csv.reader([mount(self.video, "/input/media")]))) def test_register_stable_uri_and_save_failure(self): with self.probe(): first = self.manager.register("sample01", str(self.video)) second = self.manager.register("sample01", self.video.name, str(self.root)) self.assertEqual(first["uri"], second["uri"]) before = (self.manager.root / "registry.json").read_bytes() with patch.object(self.manager, "save", side_effect=OSError("disk full")): with self.assertRaises(OSError): self.manager.register("sample02", str(self.video)) self.assertEqual(before, (self.manager.root / "registry.json").read_bytes()) def test_running_replace_and_duplicate_start_rejected(self): with self.probe(): self.manager.register("sample01", str(self.video)) with patch.object(self.manager, "containers", return_value=[self.container()]): with self.assertRaisesRegex(SenderError, "running"): self.manager.register("sample01", str(self.video)) with self.assertRaisesRegex(SenderError, "Already running"): self.manager.start("sample01", "once") def test_limit_preserves_existing(self): with self.probe(): self.manager.register("sample05", str(self.video)) existing = [self.container(f"sample{i:02}") for i in range(1,5)] with patch.object(self.manager, "containers", return_value=existing): with self.assertRaisesRegex(SenderError, "limit4"): self.manager.start("sample05") def test_stop_only_owned_and_idempotent(self): other = self.container("other") other["Config"]["Labels"][f"{LABEL}.owner"] = "not-owned" with patch.object(self.manager, "containers", return_value=[other]), patch.object(self.manager, "remove") as remove: self.manager.stop() remove.assert_not_called() with patch.object(self.manager, "containers", return_value=[]): self.assertEqual(self.manager.stop()["stopped"], []) def test_inspect_identity_before_stop(self): container = self.container() bad = self.container() bad["Config"]["Labels"][f"{LABEL}.owner"] = "another-owner" with patch.object(self.manager, "run", return_value=json.dumps([bad])) as run: with self.assertRaisesRegex(SenderError, "ownership"): self.manager.remove(container) self.assertEqual(run.call_count, 1) def test_invalid_codec_and_name(self): with patch("tools.rtsp_test_manager.command", return_value='{"streams": [{"codec_name": "vp9"}]}'): with self.assertRaises(SenderError): video_info(str(self.video)) with self.assertRaises(SenderError): self.manager.register("../bad", str(self.video)) with self.assertRaises(SenderError): self.manager.start("sample01", "remembered") def test_atomic_failure_keeps_previous(self): target = self.root / "state.json" atomic_json(target, {"old": True}) with patch("tools.rtsp_test_manager.os.replace", side_effect=OSError("disk")): with self.assertRaises(OSError): atomic_json(target, {"new": True}) self.assertEqual(json.loads(target.read_text()), {"old": True}) self.assertFalse(list(self.root.glob(".write-*"))) def test_readiness_is_not_docker_start(self): atomic_json(self.root / "status.json", {"state": "LISTENING"}) wait_ready(self.root, 1) atomic_json(self.root / "status.json", {"state": "STREAMING"}) wait_ready(self.root, 1) atomic_json(self.root / "status.json", {"state": "FAILED", "error": "prepare error"}) with self.assertRaisesRegex(SenderError, "prepare error"): wait_ready(self.root, 1) with self.assertRaisesRegex(SenderError, "timed out"): wait_ready(self.root / "missing", 0) def test_port_conflict_does_not_create_container(self): with self.probe(): self.manager.register("sample01", str(self.video)) with patch("tools.rtsp_test_manager.socket.socket") as sock, patch.object(self.manager, "run") as run: sock.return_value.__enter__.return_value.bind.side_effect = OSError("busy") with patch.object(self.manager, "containers", return_value=[]): with self.assertRaisesRegex(SenderError, "Port already"): self.manager.start("sample01") self.assertTrue(all("create" not in call.args[0] for call in run.call_args_list)) def test_start_failure_removes_only_new_container(self): with self.probe(): self.manager.register("sample01", str(self.video)) calls = [] def run(args): calls.append(args) if args[1] == "create": return "a"*64 if args[1] == "start": raise SenderError("start failure") if args[1] == "inspect": return json.dumps([self.container()]) return "" with patch.object(self.manager, "run", side_effect=run), patch.object(self.manager, "containers", return_value=[]), patch("tools.rtsp_test_manager.socket.socket"): with self.assertRaisesRegex(SenderError, "start failure"): self.manager.start("sample01") self.assertIn(["docker", "rm", "a"*64], calls) create = next(c for c in calls if c[1] == "create") self.assertEqual(create[create.index("--mode") + 1], "continuous") def test_bind_address_validation_and_atomic_failure(self): with patch("tools.rtsp_test_manager.socket.socket"): self.manager.configure("192.168.10.20") with self.probe(): row = self.manager.register("sample01", str(self.video)) self.assertEqual(row["uri"], "rtsp://192.168.10.20:18554/sample01") for address in ("0.0.0.0", "8.8.8.8", "224.0.0.1"): with self.assertRaises(SenderError): self.manager.configure(address) before = (self.manager.root / "registry.json").read_bytes() with patch.object(self.manager, "containers", return_value=[self.container()]): with self.assertRaises(SenderError): self.manager.configure("192.168.10.21") with patch("tools.rtsp_test_manager.socket.socket"), patch.object(self.manager, "save", side_effect=OSError("disk")): with self.assertRaises(OSError): self.manager.configure("192.168.10.21") self.assertEqual(before, (self.manager.root / "registry.json").read_bytes()) def test_status_uses_execution_bundle_without_connecting(self): with self.probe(): self.manager.register("sample01", str(self.video)) container = self.container() labels = container["Config"]["Labels"] labels.update({f"{LABEL}.status": str(self.root), f"{LABEL}.mode": "continuous", f"{LABEL}.execution": "new", f"{LABEL}.uri": "rtsp://192.168.10.20:18554/sample01"}) atomic_json(self.root / "status.json", {"state": "STREAMING", "executionId": "old", "uri": labels[f"{LABEL}.uri"], "anchor": {"bad": True}}) with patch.object(self.manager, "containers", return_value=[container]), patch("tools.rtsp_test_manager.socket.socket") as socket: self.assertIsNone(self.manager.status()[0]["anchor"]) socket.assert_not_called() atomic_json(self.root / "status.json", {"state": "STREAMING", "executionId": "new", "uri": labels[f"{LABEL}.uri"], "anchor": {"valid": True}}) with patch.object(self.manager, "containers", return_value=[container]): self.assertEqual(self.manager.status()[0]["anchor"], {"valid": True}) container["State"].update(Running=False, ExitCode=0) self.assertIsNone(self.manager.status()[0]["anchor"]) def test_stored_wildcard_cannot_bypass_configure(self): with self.probe(): self.manager.register("sample01", str(self.video)) with self.manager.registry() as data: data["bind_ip"] = "0.0.0.0" self.manager.save(data) with patch.object(self.manager, "containers") as containers: with self.assertRaises(SenderError): self.manager.start("sample01") containers.assert_not_called() if __name__ == "__main__": unittest.main()