"""Tests for the audio/video streaming pipeline module. Tests cover: - Platform profiles and presets - GStreamer pipeline builder - Streamer lifecycle (state machine, configuration) - Scene management - Stream API endpoints - Camera detection (mocked) These tests do NOT require actual camera hardware or GStreamer. They validate the logic, configuration, and state management. """ from __future__ import annotations import copy import threading import time from unittest.mock import patch, MagicMock, PropertyMock import pytest from src.streaming.platforms import ( PlatformProfile, PRESET_PLATFORMS, get_platform_profile, list_platforms, YOUTUBE_RTMP, TWITCH_RTMP, ) from src.streaming.gst_pipeline import ( GstPipelineBuilder, GstPipelineConfig, VideoSource, AudioSource, EncoderType, build_pipeline_string, ) from src.streaming.streamer import ( Streamer, StreamerConfig, StreamState, StreamScene, StreamStats, create_streamer, ) from src.streaming.camera import ( CameraInfo, CameraType, CameraResolution, detect_cameras, get_default_camera, ) from src.streaming.controls import ( StreamKeyboardController, StreamHotkey, HotkeyMapping, DEFAULT_HOTKEYS, start_keyboard_controller, ) # ═══════════════════════════════════════════════════════════════════════════════ # Platform profiles # ═══════════════════════════════════════════════════════════════════════════════ class TestPlatformProfiles: """Tests for streaming platform profiles and presets.""" def test_all_presets_have_required_fields(self): """Every preset should have RTMP URL and encode settings.""" required = ["name", "platform_id", "rtmp_url", "width", "height", "video_bitrate_kbps", "audio_bitrate_kbps"] for pid, profile in PRESET_PLATFORMS.items(): for field in required: assert getattr(profile, field, None) is not None, \ f"Preset '{pid}' missing field '{field}'" assert profile.width > 0 assert profile.height > 0 assert profile.video_bitrate_kbps > 0 assert profile.audio_bitrate_kbps > 0 def test_youtube_preset(self): profile = PRESET_PLATFORMS["youtube"] assert profile.platform_id == "youtube" assert "youtube" in profile.rtmp_url assert profile.width == 1280 assert profile.height == 720 assert profile.framerate == 30 assert profile.low_latency is True def test_twitch_preset(self): profile = PRESET_PLATFORMS["twitch"] assert profile.platform_id == "twitch" assert "twitch" in profile.rtmp_url assert profile.encoder == "h264" def test_audio_only_preset(self): profile = PRESET_PLATFORMS["audio_only"] assert profile.platform_id == "audio_only" assert profile.audio_bitrate_kbps == 192 assert profile.video_bitrate_kbps < 200 # Very low video assert profile.framerate <= 5 def test_get_platform_profile_found(self): for pid in ["youtube", "twitch", "custom", "audio_only"]: assert get_platform_profile(pid) is not None def test_get_platform_profile_not_found(self): assert get_platform_profile("nonexistent") is None def test_list_platforms(self): platforms = list_platforms() assert len(platforms) >= 5 for p in platforms: assert "id" in p assert "name" in p def test_platform_to_dict(self): profile = PRESET_PLATFORMS["youtube"] d = profile.to_dict() assert d["platform_id"] == "youtube" assert d["width"] == 1280 assert d["encoder"] == "h264" def test_platform_from_dict(self): data = PRESET_PLATFORMS["youtube"].to_dict() profile = PlatformProfile.from_dict(data) assert profile.platform_id == "youtube" assert profile.width == 1280 # ═══════════════════════════════════════════════════════════════════════════════ # GStreamer pipeline builder # ═══════════════════════════════════════════════════════════════════════════════ class TestGstPipelineBuilder: """Tests for GStreamer pipeline string construction.""" def test_build_basic_pipeline(self): config = GstPipelineConfig( video_source=VideoSource.TEST, audio_source=AudioSource.TEST, video_encoder=EncoderType.X264, rtmp_url="rtmp://localhost/live", rtmp_stream_key="test-key", ) builder = GstPipelineBuilder(config) pipeline = builder.build() assert "videotestsrc" in pipeline assert "audiotestsrc" in pipeline assert "x264enc" in pipeline assert "flvmux" in pipeline def test_rtmp_url_construction(self): config = GstPipelineConfig( rtmp_url="rtmp://example.com/live", rtmp_stream_key="abc123", ) assert config.full_rtmp_url == "rtmp://example.com/live/abc123" def test_rtmp_url_no_key(self): config = GstPipelineConfig( rtmp_url="rtmp://example.com/live", rtmp_stream_key="", ) assert config.full_rtmp_url == "rtmp://example.com/live" def test_build_v4l2_source(self): config = GstPipelineConfig( video_source=VideoSource.V4L2, video_device="/dev/video0", ) builder = GstPipelineBuilder(config) pipeline = builder.build() assert "v4l2src" in pipeline assert "device=/dev/video0" in pipeline def test_build_libcamera_source(self): config = GstPipelineConfig( video_source=VideoSource.LIBCAMERA, video_device="/dev/video0", ) builder = GstPipelineBuilder(config) pipeline = builder.build() assert "libcamerasrc" in pipeline def test_build_rpicam_source(self): config = GstPipelineConfig( video_source=VideoSource.RASPICAM, video_bitrate_kbps=3000, ) builder = GstPipelineBuilder(config) pipeline = builder.build() assert "rpicamsrc" in pipeline def test_build_alias_audio_source(self): config = GstPipelineConfig(audio_source=AudioSource.ALSA) builder = GstPipelineBuilder(config) pipeline = builder.build() assert "alsasrc" in pipeline def test_build_jack_audio_source(self): config = GstPipelineConfig(audio_source=AudioSource.JACK) builder = GstPipelineBuilder(config) pipeline = builder.build() assert "jackaudiosrc" in pipeline def test_build_pulse_audio_source(self): config = GstPipelineConfig(audio_source=AudioSource.PULSE) builder = GstPipelineBuilder(config) pipeline = builder.build() assert "pulsesrc" in pipeline def test_v4l2_h264_encoder(self): config = GstPipelineConfig( video_source=VideoSource.TEST, audio_source=AudioSource.TEST, video_encoder=EncoderType.V4L2_H264, ) builder = GstPipelineBuilder(config) pipeline = builder.build() assert "v4l2h264enc" in pipeline def test_omx_h264_encoder(self): config = GstPipelineConfig( video_source=VideoSource.TEST, audio_source=AudioSource.TEST, video_encoder=EncoderType.OMX_H264, ) builder = GstPipelineBuilder(config) pipeline = builder.build() assert "omxh264enc" in pipeline def test_x264_encoder(self): config = GstPipelineConfig( video_source=VideoSource.TEST, audio_source=AudioSource.TEST, video_encoder=EncoderType.X264, ) builder = GstPipelineBuilder(config) pipeline = builder.build() assert "x264enc" in pipeline def test_low_latency_queue_elements(self): config = GstPipelineConfig( video_source=VideoSource.TEST, audio_source=AudioSource.TEST, low_latency=True, video_queue_max_time_ms=500, ) builder = GstPipelineBuilder(config) pipeline = builder.build() assert "max-size-time" in pipeline def test_high_latency_no_queue_limits(self): config = GstPipelineConfig( video_source=VideoSource.TEST, audio_source=AudioSource.TEST, low_latency=False, ) builder = GstPipelineBuilder(config) pipeline = builder.build() # Without low latency, queues should be simpler assert "queue" in pipeline def test_pipeline_includes_h264parse(self): config = GstPipelineConfig( video_source=VideoSource.TEST, audio_source=AudioSource.TEST, video_encoder=EncoderType.X264, ) builder = GstPipelineBuilder(config) pipeline = builder.build() assert "h264parse" in pipeline def test_pipeline_includes_audio_encoder(self): config = GstPipelineConfig( video_source=VideoSource.TEST, audio_source=AudioSource.TEST, audio_encoder="lamemp3enc", audio_bitrate_kbps=192, ) builder = GstPipelineBuilder(config) pipeline = builder.build() assert "lamemp3enc" in pipeline assert "bitrate=192000" in pipeline def test_build_pipeline_string_helper(self): pipeline = build_pipeline_string( video_device="/dev/video0", audio_device="hw:0", rtmp_url="rtmp://example.com/live/key", ) assert "v4l2src" in pipeline assert "alsasrc" in pipeline assert "rtmp" in pipeline def test_test_mode_pipeline(self): pipeline = build_pipeline_string( video_device="test", audio_device="test", rtmp_url="rtmp://example.com/live/key", ) assert "videotestsrc" in pipeline assert "audiotestsrc" in pipeline def test_build_args_list(self): config = GstPipelineConfig( video_source=VideoSource.TEST, audio_source=AudioSource.TEST, video_encoder=EncoderType.X264, ) builder = GstPipelineBuilder(config) args = builder.build_args() assert isinstance(args, list) assert len(args) > 5 def test_pipeline_includes_resolution(self): config = GstPipelineConfig( video_source=VideoSource.TEST, audio_source=AudioSource.TEST, video_width=1920, video_height=1080, ) builder = GstPipelineBuilder(config) pipeline = builder.build() assert "1920" in pipeline assert "1080" in pipeline # ═══════════════════════════════════════════════════════════════════════════════ # Streamer lifecycle # ═══════════════════════════════════════════════════════════════════════════════ class TestStreamerLifecycle: """Tests for Streamer state machine and lifecycle.""" def test_initial_state_is_idle(self): streamer = Streamer() assert streamer.state == StreamState.IDLE assert streamer.is_live is False def test_configure_sets_state(self): streamer = Streamer() streamer.configure(platform_id="youtube") assert streamer.state == StreamState.CONFIGURED def test_configure_applies_platform_preset(self): streamer = Streamer() streamer.configure(platform_id="youtube", stream_key="test-key") assert streamer.config.platform_id == "youtube" assert streamer.config.stream_key == "test-key" assert streamer.config.rtmp_url == YOUTUBE_RTMP def test_configure_twitch_preset(self): streamer = Streamer() streamer.configure(platform_id="twitch", stream_key="live_xxxx") assert streamer.config.rtmp_url == TWITCH_RTMP def test_configure_custom_devices(self): streamer = Streamer() streamer.configure( platform_id="youtube", video_device="/dev/video2", audio_device="hw:1", ) assert streamer.config.video_device == "/dev/video2" assert streamer.config.audio_device == "hw:1" def test_configure_partial_updates(self): streamer = Streamer() streamer.configure(platform_id="youtube") # Partial update — only change stream key streamer.configure(stream_key="new-key") assert streamer.config.stream_key == "new-key" assert streamer.config.platform_id == "youtube" # Unchanged def test_start_requires_rtmp_url(self): streamer = Streamer() # No platform configured, no RTMP URL, no stream key success = streamer.start() assert success is False assert streamer.state == StreamState.ERROR def test_stats_initial_state(self): streamer = Streamer() stats = streamer.stats assert stats.state == StreamState.IDLE assert stats.uptime_seconds == 0.0 assert stats.reconnect_count == 0 def test_get_stats_dict(self): streamer = Streamer() d = streamer.get_stats_dict() assert d["state"] == "idle" assert "uptime_seconds" in d assert "reconnect_count" in d def test_config_from_preset(self): config = StreamerConfig.from_platform("youtube", "test-key") assert config.platform_id == "youtube" assert config.stream_key == "test-key" assert config.video_width == 1280 assert config.video_height == 720 assert config.low_latency is True def test_config_from_unknown_preset(self): config = StreamerConfig.from_platform("nonexistent", "test-key") assert config.platform_id == "custom" # Falls back to custom def test_config_full_rtmp_url(self): config = StreamerConfig( rtmp_url=YOUTUBE_RTMP, stream_key="abcd-efgh-ijkl", ) assert config.full_rtmp_url == f"{YOUTUBE_RTMP}/abcd-efgh-ijkl" def test_config_full_rtmp_url_no_key(self): config = StreamerConfig(rtmp_url=YOUTUBE_RTMP, stream_key="") assert config.full_rtmp_url == YOUTUBE_RTMP def test_config_full_rtmp_url_trailing_slash(self): config = StreamerConfig( rtmp_url="rtmp://example.com/live/", stream_key="key123", ) assert config.full_rtmp_url == "rtmp://example.com/live/key123" # ═══════════════════════════════════════════════════════════════════════════════ # Scene management # ═══════════════════════════════════════════════════════════════════════════════ class TestSceneManagement: """Tests for streaming scene management.""" def test_default_scene_exists(self): streamer = Streamer() scenes = streamer.get_scenes() assert len(scenes) == 1 assert scenes[0]["name"] == "default" def test_add_scene(self): streamer = Streamer() scene = StreamScene( name="closeup", camera_device="/dev/video1", video_width=640, video_height=480, description="Close-up camera", ) streamer.add_scene(scene) scenes = streamer.get_scenes() assert len(scenes) == 2 names = [s["name"] for s in scenes] assert "closeup" in names def test_add_duplicate_scene_name(self): streamer = Streamer() scene = StreamScene(name="default") streamer.add_scene(scene) # Adding with same name should overwrite scenes = streamer.get_scenes() assert len(scenes) == 1 def test_remove_scene(self): streamer = Streamer() scene = StreamScene(name="test") streamer.add_scene(scene) assert streamer.remove_scene("test") is True assert streamer.remove_scene("test") is False def test_cannot_remove_default_scene(self): streamer = Streamer() assert streamer.remove_scene("default") is False def test_switch_scene_not_found(self): streamer = Streamer() assert streamer.switch_scene("nonexistent") is False def test_switch_scene_updates_config(self): streamer = Streamer() scene = StreamScene( name="wide", camera_device="/dev/video2", video_width=1920, video_height=1080, video_framerate=24, ) streamer.add_scene(scene) streamer.switch_scene("wide") assert streamer.config.video_device == "/dev/video2" assert streamer.config.video_width == 1920 assert streamer.config.video_height == 1080 def test_scene_to_dict(self): scene = StreamScene( name="test", camera_device="/dev/video0", video_width=1280, video_height=720, video_framerate=30, audio_device="hw:0", overlay_text="Live!", description="Test scene", ) d = scene.to_dict() assert d["name"] == "test" assert d["camera_device"] == "/dev/video0" assert d["overlay_text"] == "Live!" # ═══════════════════════════════════════════════════════════════════════════════ # Streamer configuration # ═══════════════════════════════════════════════════════════════════════════════ class TestStreamerConfig: """Tests for StreamerConfig behavior.""" def test_auto_reconnect_defaults(self): streamer = Streamer() assert streamer.config.auto_reconnect is True assert streamer.config.max_reconnect_attempts == 5 assert streamer.config.reconnect_delay_seconds == 3.0 def test_encoder_type_default(self): streamer = Streamer() assert streamer.config.encoder_type == EncoderType.V4L2_H264 def test_low_latency_default(self): streamer = Streamer() assert streamer.config.low_latency is True def test_scenes_default(self): streamer = Streamer() assert streamer.config.active_scene == "default" assert len(streamer.get_scenes()) == 1 def test_create_streamer_factory(self): streamer = create_streamer( platform_id="youtube", stream_key="test-key", video_device="/dev/video0", audio_device="hw:0", ) assert streamer.state == StreamState.CONFIGURED assert streamer.config.platform_id == "youtube" # ═══════════════════════════════════════════════════════════════════════════════ # Stream statistics # ═══════════════════════════════════════════════════════════════════════════════ class TestStreamStats: """Tests for StreamStats data class.""" def test_stats_to_dict(self): stats = StreamStats( state=StreamState.LIVE, uptime_seconds=120.5, video_bitrate_kbps=2500.0, camera_name="Test Cam", platform_name="youtube", ) d = stats.to_dict() assert d["state"] == "live" assert d["uptime_seconds"] == 120.5 assert d["video_bitrate_kbps"] == 2500.0 assert d["camera_name"] == "Test Cam" def test_stats_copy_is_independent(self): streamer = Streamer() stats1 = streamer.stats stats2 = streamer.stats assert stats1 is not stats2 # Different objects def test_state_change_callback(self): streamer = Streamer() received_states = [] def callback(state, stats): received_states.append(state) streamer.on_state_change(callback) streamer.configure(platform_id="youtube") assert len(received_states) >= 1 assert StreamState.CONFIGURED in received_states # ═══════════════════════════════════════════════════════════════════════════════ # Camera detection # ═══════════════════════════════════════════════════════════════════════════════ class TestCameraDetection: """Tests for camera detection logic (mocked).""" def test_camera_info_to_dict(self): cam = CameraInfo( device_path="/dev/video0", camera_type=CameraType.USB_WEBCAM, name="Test Webcam", driver="uvcvideo", resolutions=[ CameraResolution(640, 480, 30), CameraResolution(1280, 720, 30), ], pixel_formats=["YUYV", "MJPG"], ) d = cam.to_dict() assert d["device_path"] == "/dev/video0" assert d["camera_type"] == "usb_webcam" assert len(d["resolutions"]) == 2 def test_camera_best_resolution(self): cam = CameraInfo( device_path="/dev/video0", camera_type=CameraType.USB_WEBCAM, name="Test", resolutions=[ CameraResolution(640, 480, 30), CameraResolution(1920, 1080, 30), CameraResolution(1280, 720, 60), ], ) best = cam.best_resolution assert best is not None assert best.width == 1920 assert best.height == 1080 def test_camera_no_resolutions(self): cam = CameraInfo( device_path="/dev/video0", camera_type=CameraType.USB_WEBCAM, name="Test", ) assert cam.best_resolution is None def test_camera_supports_h264(self): cam = CameraInfo( device_path="/dev/video0", camera_type=CameraType.PI_CAMERA, name="Pi Cam", pixel_formats=["YUYV", "H264"], ) assert cam.supports_h264 is True def test_camera_no_h264(self): cam = CameraInfo( device_path="/dev/video0", camera_type=CameraType.USB_WEBCAM, name="Webcam", pixel_formats=["YUYV"], ) assert cam.supports_h264 is False def test_classify_pi_camera(self): cam = CameraInfo( device_path="/dev/video0", camera_type=CameraType.PI_CAMERA, name="Pi Camera (imx219)", driver="unicam", ) assert cam.camera_type == CameraType.PI_CAMERA def test_classify_usb_webcam(self): cam = CameraInfo( device_path="/dev/video1", camera_type=CameraType.USB_WEBCAM, name="USB Camera", driver="uvcvideo", ) assert cam.camera_type == CameraType.USB_WEBCAM # ═══════════════════════════════════════════════════════════════════════════════ # Keyboard controls # ═══════════════════════════════════════════════════════════════════════════════ class TestKeyboardControls: """Tests for keyboard controller logic.""" def test_default_hotkeys_defined(self): assert len(DEFAULT_HOTKEYS) >= 10 def test_start_stop_hotkey_exists(self): actions = [hk.hotkey for hk in DEFAULT_HOTKEYS] assert StreamHotkey.START_STOP in actions def test_scene_hotkeys_exist(self): actions = [hk.hotkey for hk in DEFAULT_HOTKEYS] for i in range(1, 10): scene_action = StreamHotkey(f"scene_{i}") assert scene_action in actions, f"Missing scene hotkey: scene_{i}" def test_hotkey_descriptions(self): for hk in DEFAULT_HOTKEYS: assert hk.description, f"Hotkey {hk.hotkey} missing description" assert hk.key assert hk.modifiers def test_controller_init(self): streamer = Streamer() controller = StreamKeyboardController(streamer) assert controller._running is False def test_controller_get_hotkeys(self): streamer = Streamer() controller = StreamKeyboardController(streamer) hotkeys = controller.get_hotkeys() assert len(hotkeys) == len(DEFAULT_HOTKEYS) for hk in hotkeys: assert "action" in hk assert "modifiers" in hk assert "key" in hk assert "description" in hk def test_controller_custom_hotkeys(self): streamer = Streamer() custom = [ HotkeyMapping( StreamHotkey.START_STOP, {"ctrl"}, "x", "Custom start/stop", ), ] controller = StreamKeyboardController(streamer, hotkeys=custom) hotkeys = controller.get_hotkeys() assert len(hotkeys) == 1 assert hotkeys[0]["key"] == "x" def test_hotkey_mapping_to_dict(self): hk = HotkeyMapping( StreamHotkey.START_STOP, {"ctrl", "shift"}, "s", "Start/Stop", ) assert hk.hotkey == StreamHotkey.START_STOP assert hk.key == "s" assert "ctrl" in hk.modifiers assert "shift" in hk.modifiers # ═══════════════════════════════════════════════════════════════════════════════ # API endpoint integration (unit-level) # ═══════════════════════════════════════════════════════════════════════════════ class TestStreamAPIProvider: """Tests for StreamStateProvider and stream route adapter.""" def test_provider_initial_state(self): from src.network.stream_routes import StreamStateProvider sp = StreamStateProvider() assert sp.get_status is None assert sp.start_stream is None assert sp.list_scenes is None def test_provider_can_be_wired(self): from src.network.stream_routes import StreamStateProvider sp = StreamStateProvider() def dummy_status() -> dict: return {"state": "idle"} sp.get_status = dummy_status assert sp.get_status() == {"state": "idle"} # ═══════════════════════════════════════════════════════════════════════════════ # Pipeline edge cases # ═══════════════════════════════════════════════════════════════════════════════ class TestPipelineEdgeCases: """Edge case tests for GStreamer pipeline builder.""" def test_mono_audio_channel(self): config = GstPipelineConfig( video_source=VideoSource.TEST, audio_source=AudioSource.TEST, audio_channels=1, ) builder = GstPipelineBuilder(config) pipeline = builder.build() assert "channels=1" in pipeline def test_custom_audio_encoder(self): config = GstPipelineConfig( video_source=VideoSource.TEST, audio_source=AudioSource.TEST, audio_encoder="voaacenc", audio_bitrate_kbps=96, ) builder = GstPipelineBuilder(config) pipeline = builder.build() assert "voaacenc" in pipeline def test_high_bitrate_config(self): config = GstPipelineConfig( video_source=VideoSource.TEST, audio_source=AudioSource.TEST, video_bitrate_kbps=6000, video_bitrate_max_kbps=10000, ) assert config.video_bitrate_kbps == 6000 def test_libcamera_h264_encoder(self): config = GstPipelineConfig( video_source=VideoSource.LIBCAMERA, audio_source=AudioSource.TEST, video_encoder=EncoderType.LIBCAMERA_H264, ) builder = GstPipelineBuilder(config) pipeline = builder.build() # libcamera h264 encoder shouldn't add separate encoder element assert "libcamerasrc" in pipeline def test_empty_rtmp_url_pipeline(self): config = GstPipelineConfig( video_source=VideoSource.TEST, audio_source=AudioSource.TEST, rtmp_url="", ) builder = GstPipelineBuilder(config) pipeline = builder.build() assert "fakesink" in pipeline def test_full_rtmp_url_with_trailing_slash(self): config = GstPipelineConfig( rtmp_url="rtmp://example.com/live/", rtmp_stream_key="stream-key", ) assert config.full_rtmp_url == "rtmp://example.com/live/stream-key" def test_video_bitrate_max(self): config = GstPipelineConfig( video_bitrate_kbps=3000, video_bitrate_max_kbps=5000, ) assert config.video_bitrate_max_kbps == 5000 def test_keyframe_interval(self): config = GstPipelineConfig(keyframe_interval=120) assert config.keyframe_interval == 120 def test_h264_profile_and_level(self): config = GstPipelineConfig( h264_profile="high", h264_level="4.1", ) assert config.h264_profile == "high" assert config.h264_level == "4.1" def test_zero_latency_tuning(self): config = GstPipelineConfig( low_latency=True, zerolatency_tune=True, b_frames=0, ) assert config.b_frames == 0 assert config.zerolatency_tune is True def test_audio_buffer_time(self): config = GstPipelineConfig( audio_source=AudioSource.ALSA, audio_buffer_time_us=10000, ) assert config.audio_buffer_time_us == 10000 def test_test_video_pattern(self): config = GstPipelineConfig( video_source=VideoSource.TEST, test_video_pattern=18, # SMPTE color bars ) builder = GstPipelineBuilder(config) pipeline = builder.build() assert "pattern=18" in pipeline def test_test_audio_frequency(self): config = GstPipelineConfig( audio_source=AudioSource.TEST, test_audio_freq=1000.0, # 1kHz test tone ) builder = GstPipelineBuilder(config) pipeline = builder.build() assert "freq=1000.0" in pipeline