121 lines
3.7 KiB
Python
121 lines
3.7 KiB
Python
import unittest
|
|
|
|
from frigate_camera_control_bridge.app import CameraControlBridge
|
|
|
|
|
|
class DiscoveryMappingTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.bridge = CameraControlBridge
|
|
|
|
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.camera_candidate_urls("camera-a", camera_cfg, streams)
|
|
|
|
self.assertEqual(urls, ["rtsp://camera-a.example.test:554/Streaming/Channels/101"])
|
|
self.assertEqual(self.bridge.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.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.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.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.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.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.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.camera_candidate_urls("camera-a", camera_cfg, streams)
|
|
|
|
self.assertEqual(
|
|
urls,
|
|
[
|
|
"rtsp://direct-camera.example.test:554/Streaming/Channels/102",
|
|
],
|
|
)
|
|
self.assertEqual(
|
|
self.bridge.hosts_from_urls(urls),
|
|
[
|
|
"direct-camera.example.test",
|
|
],
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|