104 lines
4.3 KiB
Python
104 lines
4.3 KiB
Python
from django.test import TestCase
|
|
|
|
from apps.core.settings_service import SettingsService
|
|
from apps.live_monitoring.models import Camera
|
|
from apps.live_monitoring.protocols import PROTOCOL_KEYS, detect_protocol_from_url
|
|
from apps.live_monitoring.services import StreamAccessService
|
|
|
|
|
|
class StreamAccessTests(TestCase):
|
|
def setUp(self):
|
|
self.camera = Camera.objects.create(
|
|
name="Cam1",
|
|
protocol="hls",
|
|
stream_url="https://example.com/live.m3u8",
|
|
)
|
|
self.service = StreamAccessService()
|
|
|
|
def test_issue_and_verify_token(self):
|
|
token, exp = self.service.issue_token(self.camera)
|
|
payload = self.service.verify_token(token)
|
|
self.assertIsNotNone(payload)
|
|
self.assertEqual(payload["camera"], str(self.camera.id))
|
|
self.assertGreater(payload["exp"], exp - 1)
|
|
|
|
def test_tampered_token_rejected(self):
|
|
token, _ = self.service.issue_token(self.camera)
|
|
raw, _sig = token.split(".", 1)
|
|
self.assertIsNone(self.service.verify_token(raw + ".deadbeef"))
|
|
|
|
def test_expired_token_rejected(self):
|
|
token, _ = self.service.issue_token(self.camera, ttl=-1)
|
|
self.assertIsNone(self.service.verify_token(token))
|
|
|
|
def test_ttl_comes_from_settings(self):
|
|
SettingsService().set("camera.stream_ttl_seconds", 120)
|
|
token, exp = self.service.issue_token(self.camera)
|
|
payload = self.service.verify_token(token)
|
|
self.assertAlmostEqual(payload["exp"], exp, delta=1)
|
|
# Token should expire roughly 120s from now (not the old 3600 default).
|
|
import time
|
|
|
|
self.assertLess(payload["exp"] - int(time.time()), 130)
|
|
|
|
def test_resolve_hls_stream(self):
|
|
result = self.service.resolve_stream(self.camera)
|
|
self.assertEqual(result["protocol"], "hls")
|
|
self.assertEqual(result["stream_url"], "https://example.com/live.m3u8")
|
|
self.assertTrue(result["browser_native"])
|
|
self.assertEqual(result["player"], "hls")
|
|
|
|
def test_rtsp_requires_gateway_or_errors(self):
|
|
cam = Camera.objects.create(
|
|
name="RTSP Cam",
|
|
protocol="rtsp",
|
|
stream_url="rtsp://user:secret@192.168.1.10:554/stream1",
|
|
)
|
|
result = self.service.resolve_stream(cam)
|
|
self.assertIsNone(result["stream_url"])
|
|
self.assertIn("media gateway", result["error"].lower())
|
|
|
|
def test_rtsp_rewrites_through_gateway(self):
|
|
SettingsService().set("camera.media_gateway_url", "https://media.example.com")
|
|
cam = Camera.objects.create(
|
|
name="RTSP Cam",
|
|
protocol="rtsp",
|
|
stream_url="rtsp://user:secret@192.168.1.10:554/stream1",
|
|
)
|
|
result = self.service.resolve_stream(cam)
|
|
self.assertEqual(result["protocol"], "hls")
|
|
self.assertEqual(result["stream_url"], f"https://media.example.com/{cam.id}/index.m3u8")
|
|
self.assertNotIn("secret", result["stream_url"])
|
|
|
|
def test_strips_embedded_credentials(self):
|
|
cam = Camera.objects.create(
|
|
name="Cred Cam",
|
|
protocol="http",
|
|
stream_url="https://user:pass@cdn.example.com/live.mp4",
|
|
)
|
|
result = self.service.resolve_stream(cam)
|
|
self.assertEqual(result["stream_url"], "https://cdn.example.com/live.mp4")
|
|
|
|
def test_inherits_default_protocol_from_settings(self):
|
|
SettingsService().set("camera.default_protocol", "mjpeg")
|
|
cam = Camera.objects.create(name="Inherit", stream_url="https://cam/video.mjpg")
|
|
self.assertEqual(cam.resolved_protocol(), "mjpeg")
|
|
|
|
def test_monitoring_disabled(self):
|
|
SettingsService().set("camera.enabled", False)
|
|
result = self.service.resolve_stream(self.camera)
|
|
self.assertIsNone(result["stream_url"])
|
|
self.assertIn("disabled", result["error"].lower())
|
|
|
|
|
|
class ProtocolCatalogTests(TestCase):
|
|
def test_all_standard_protocols_registered(self):
|
|
for key in ("rtsp", "rtmp", "hls", "dash", "webrtc", "mjpeg", "http", "onvif", "srt"):
|
|
self.assertIn(key, PROTOCOL_KEYS)
|
|
|
|
def test_detect_from_url(self):
|
|
self.assertEqual(detect_protocol_from_url("rtsp://x/y"), "rtsp")
|
|
self.assertEqual(detect_protocol_from_url("https://x/a.m3u8"), "hls")
|
|
self.assertEqual(detect_protocol_from_url("https://x/a.mpd"), "dash")
|
|
self.assertEqual(detect_protocol_from_url("srt://1.2.3.4:9000"), "srt")
|