130 lines
4.2 KiB
Python
130 lines
4.2 KiB
Python
import importlib.util
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
def load_bridge_module():
|
|
root = Path(__file__).resolve().parents[1]
|
|
path = root / "frigate-camera-control-bridge.py"
|
|
spec = importlib.util.spec_from_file_location("bridge", path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
class DiscoveryMappingTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.bridge = load_bridge_module()
|
|
|
|
def test_loopback_ffmpeg_path_resolves_through_go2rtc(self):
|
|
camera_cfg = {
|
|
"ffmpeg": {
|
|
"inputs": [
|
|
{"path": "rtsp://127.0.0.1:8554/stream-a?video=copy"},
|
|
]
|
|
}
|
|
}
|
|
streams = {
|
|
"stream-a": [
|
|
"rtsp://camera-a.example.test:554/Streaming/Channels/101",
|
|
]
|
|
}
|
|
|
|
urls = self.bridge.CameraControlBridge.camera_candidate_urls("camera-a", camera_cfg, streams)
|
|
|
|
self.assertEqual(urls, ["rtsp://camera-a.example.test:554/Streaming/Channels/101"])
|
|
self.assertEqual(self.bridge.CameraControlBridge.hosts_from_urls(urls), ["camera-a.example.test"])
|
|
|
|
def test_wrapped_go2rtc_stream_resolves_inner_camera_url(self):
|
|
camera_cfg = {
|
|
"ffmpeg": {
|
|
"inputs": [
|
|
{"path": "rtsp://127.0.0.1:8554/stream-a"},
|
|
]
|
|
}
|
|
}
|
|
streams = {
|
|
"stream-a": [
|
|
"ffmpeg:rtsp://camera-a.example.test:554/Streaming/Channels/101?profile=main#hardware",
|
|
]
|
|
}
|
|
|
|
urls = self.bridge.CameraControlBridge.camera_candidate_urls("camera-a", camera_cfg, streams)
|
|
|
|
self.assertEqual(
|
|
urls,
|
|
[
|
|
"rtsp://camera-a.example.test:554/Streaming/Channels/101?profile=main#hardware",
|
|
],
|
|
)
|
|
self.assertEqual(self.bridge.CameraControlBridge.hosts_from_urls(urls), ["camera-a.example.test"])
|
|
|
|
def test_direct_ffmpeg_path_is_used_without_go2rtc(self):
|
|
camera_cfg = {
|
|
"ffmpeg": {
|
|
"inputs": [
|
|
{"path": "rtsp://camera-b.example.test:554/cam/realmonitor?channel=1&subtype=0"},
|
|
]
|
|
}
|
|
}
|
|
|
|
urls = self.bridge.CameraControlBridge.camera_candidate_urls("camera-b", camera_cfg, {})
|
|
|
|
self.assertEqual(urls, ["rtsp://camera-b.example.test:554/cam/realmonitor?channel=1&subtype=0"])
|
|
self.assertEqual(self.bridge.CameraControlBridge.hosts_from_urls(urls), ["camera-b.example.test"])
|
|
|
|
def test_wrapped_direct_ffmpeg_path_is_used_without_go2rtc(self):
|
|
camera_cfg = {
|
|
"ffmpeg": {
|
|
"inputs": [
|
|
{"path": "ffmpeg:rtsp://camera-b.example.test:554/cam/realmonitor?channel=1&subtype=0#video=copy"},
|
|
]
|
|
}
|
|
}
|
|
|
|
urls = self.bridge.CameraControlBridge.camera_candidate_urls("camera-b", camera_cfg, {})
|
|
|
|
self.assertEqual(
|
|
urls,
|
|
[
|
|
"rtsp://camera-b.example.test:554/cam/realmonitor?channel=1&subtype=0#video=copy",
|
|
],
|
|
)
|
|
self.assertEqual(self.bridge.CameraControlBridge.hosts_from_urls(urls), ["camera-b.example.test"])
|
|
|
|
def test_direct_path_wins_when_loopback_is_also_present(self):
|
|
camera_cfg = {
|
|
"ffmpeg": {
|
|
"inputs": [
|
|
{"path": "rtsp://127.0.0.1:8554/stream-a"},
|
|
{"path": "rtsp://direct-camera.example.test:554/Streaming/Channels/102"},
|
|
]
|
|
}
|
|
}
|
|
streams = {
|
|
"stream-a": {
|
|
"streams": [
|
|
"rtsp://camera-a.example.test:554/Streaming/Channels/101",
|
|
]
|
|
}
|
|
}
|
|
|
|
urls = self.bridge.CameraControlBridge.camera_candidate_urls("camera-a", camera_cfg, streams)
|
|
|
|
self.assertEqual(
|
|
urls,
|
|
[
|
|
"rtsp://direct-camera.example.test:554/Streaming/Channels/102",
|
|
],
|
|
)
|
|
self.assertEqual(
|
|
self.bridge.CameraControlBridge.hosts_from_urls(urls),
|
|
[
|
|
"direct-camera.example.test",
|
|
],
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|