import pytest import struct import asyncio from datetime import datetime from unittest.mock import Mock, AsyncMock, patch, MagicMock from backend.app.models import SCTE35Marker from backend.app.recorders.scte35 import SCTE35Manager class TestSCTE35Encoding: """Test SCTE35 binary encoding""" def test_encode_splice_insert_returns_bytes(self): """encode_splice_insert returns bytes""" manager = SCTE35Manager() result = manager.encode_splice_insert( event_id=1, duration_seconds=30 ) assert isinstance(result, bytes) assert len(result) > 0 def test_encode_splice_insert_starts_with_table_id(self): """Binary output starts with 0xFC (SCTE35 table ID)""" manager = SCTE35Manager() result = manager.encode_splice_insert( event_id=1, duration_seconds=30 ) assert result[0] == 0xFC def test_encode_splice_insert_event_id_present(self): """Event ID is encoded in the binary output""" manager = SCTE35Manager() event_id = 0x12345678 result = manager.encode_splice_insert( event_id=event_id, duration_seconds=30 ) # Event ID should be somewhere in the bytes (after headers) assert len(result) >= 20 # Minimum SCTE35 section length def test_encode_splice_insert_duration_flag_set(self): """Duration > 0 sets duration_flag in flags byte""" manager = SCTE35Manager() result = manager.encode_splice_insert( event_id=1, duration_seconds=30 ) # Find the flags byte (after event_id in the command) # This is a basic check that encoding succeeded assert len(result) > 0 def test_encode_splice_insert_no_duration(self): """Duration = 0 encodes without duration flag""" manager = SCTE35Manager() result = manager.encode_splice_insert( event_id=1, duration_seconds=0 ) assert isinstance(result, bytes) assert len(result) > 0 def test_encode_splice_insert_out_of_network_flag(self): """Out of network flag is set correctly""" manager = SCTE35Manager() result_oon = manager.encode_splice_insert( event_id=1, duration_seconds=30, out_of_network=True ) result_in = manager.encode_splice_insert( event_id=1, duration_seconds=30, out_of_network=False ) # Both should be bytes, potentially different lengths or content assert isinstance(result_oon, bytes) assert isinstance(result_in, bytes) def test_encode_splice_insert_splice_immediate(self): """splice_immediate flag prevents PTS encoding""" manager = SCTE35Manager() result = manager.encode_splice_insert( event_id=1, duration_seconds=30, splice_immediate=True ) assert isinstance(result, bytes) def test_encode_splice_insert_first_byte_is_table_id(self): """Binary output first byte is exactly 0xFC (SCTE35 table ID)""" manager = SCTE35Manager() result = manager.encode_splice_insert(event_id=1, duration_seconds=30.0) assert result[0] == 0xFC def test_encode_splice_insert_event_id_bytes(self): """Event ID is encoded as 4 big-endian bytes somewhere in the output""" manager = SCTE35Manager() result = manager.encode_splice_insert(event_id=0x12345678, duration_seconds=30.0) event_id_bytes = struct.pack('>I', 0x12345678) assert event_id_bytes in result def test_encode_splice_insert_out_of_network_different_bytes(self): """Out of network flag affects the binary output""" manager = SCTE35Manager() result_onn = manager.encode_splice_insert( event_id=1, duration_seconds=30.0, out_of_network=True ) result_not_onn = manager.encode_splice_insert( event_id=1, duration_seconds=30.0, out_of_network=False ) # The results should differ assert result_onn != result_not_onn def test_input_validation_negative_duration(self): """Negative duration raises ValueError""" manager = SCTE35Manager() with pytest.raises(ValueError, match="non-negative"): manager.encode_splice_insert(event_id=1, duration_seconds=-1.0) def test_input_validation_invalid_event_id_too_large(self): """Event ID out of 32-bit range raises ValueError""" manager = SCTE35Manager() with pytest.raises(ValueError, match="32-bit"): manager.encode_splice_insert(event_id=0xFFFFFFFF + 1, duration_seconds=30.0) def test_input_validation_invalid_event_id_negative(self): """Negative event ID raises ValueError""" manager = SCTE35Manager() with pytest.raises(ValueError, match="32-bit"): manager.encode_splice_insert(event_id=-1, duration_seconds=30.0) class TestSCTE35Manager: """Test SCTE35Manager functionality""" def test_manager_initialization(self): """SCTE35Manager initializes with empty history""" manager = SCTE35Manager() assert manager.get_marker_history() == [] @pytest.mark.asyncio async def test_inject_marker_returns_marker(self): """inject_marker returns SCTE35Marker with correct fields""" manager = SCTE35Manager() marker = await manager.inject_marker( event_id=1, duration_seconds=30, webhook_url=None, out_of_network=True, splice_immediate=False ) assert isinstance(marker, SCTE35Marker) assert marker.event_id == "1" # event_id is stored as string assert marker.duration_seconds == 30 assert marker.out_of_network is True assert marker.splice_immediate is False assert isinstance(marker.timestamp, datetime) @pytest.mark.asyncio async def test_inject_marker_adds_to_history(self): """inject_marker adds SCTE35Marker to history""" manager = SCTE35Manager() marker = await manager.inject_marker( event_id=1, duration_seconds=30 ) history = manager.get_marker_history() assert len(history) == 1 assert history[0].event_id == marker.event_id @pytest.mark.asyncio async def test_inject_marker_multiple_adds(self): """Multiple inject_marker calls add to history""" manager = SCTE35Manager() for i in range(3): await manager.inject_marker( event_id=i, duration_seconds=30 ) history = manager.get_marker_history() assert len(history) == 3 @pytest.mark.asyncio async def test_inject_marker_no_webhook_when_url_none(self): """No HTTP request made when webhook_url is None""" manager = SCTE35Manager() with patch('aiohttp.ClientSession.post') as mock_post: marker = await manager.inject_marker( event_id=1, duration_seconds=30, webhook_url=None ) # Should not attempt to post assert not mock_post.called @pytest.mark.asyncio async def test_inject_marker_triggers_webhook(self): """HTTP POST is made when webhook_url is provided""" manager = SCTE35Manager() webhook_url = "https://example.com/webhook" # Mock the aiohttp session with patch('aiohttp.ClientSession') as mock_session_class: mock_session = AsyncMock() mock_response = AsyncMock() mock_response.status = 200 # Setup the context managers mock_session.post.return_value = AsyncMock() mock_session.post.return_value.__aenter__ = AsyncMock(return_value=mock_response) mock_session.post.return_value.__aexit__ = AsyncMock(return_value=None) mock_session_class.return_value = AsyncMock() mock_session_class.return_value.__aenter__ = AsyncMock(return_value=mock_session) mock_session_class.return_value.__aexit__ = AsyncMock(return_value=None) marker = await manager.inject_marker( event_id=1, duration_seconds=30, webhook_url=webhook_url ) # Verify POST was called mock_session.post.assert_called_once() call_args = mock_session.post.call_args assert webhook_url in call_args[0] or webhook_url == call_args[1].get('url', '') @pytest.mark.asyncio async def test_webhook_payload_contains_required_fields(self): """Webhook JSON payload has event_id, duration_seconds, timestamp, cue_hex""" manager = SCTE35Manager() webhook_url = "https://example.com/webhook" with patch('aiohttp.ClientSession') as mock_session_class: mock_session = AsyncMock() mock_response = AsyncMock() mock_response.status = 200 # Setup the context managers properly mock_session.post.return_value = AsyncMock() mock_session.post.return_value.__aenter__ = AsyncMock(return_value=mock_response) mock_session.post.return_value.__aexit__ = AsyncMock(return_value=None) mock_session_class.return_value = AsyncMock() mock_session_class.return_value.__aenter__ = AsyncMock(return_value=mock_session) mock_session_class.return_value.__aexit__ = AsyncMock(return_value=None) marker = await manager.inject_marker( event_id=42, duration_seconds=60, webhook_url=webhook_url, out_of_network=True, splice_immediate=False ) # Extract the json argument from the POST call call_args = mock_session.post.call_args json_data = call_args[1].get('json', {}) assert json_data.get('event_id') == "42" # event_id is a string assert json_data.get('duration_seconds') == 60 assert json_data.get('out_of_network') is True assert json_data.get('splice_immediate') is False assert 'timestamp' in json_data assert 'cue_hex' in json_data def test_get_marker_history_returns_list(self): """get_marker_history returns list of markers""" manager = SCTE35Manager() history = manager.get_marker_history() assert isinstance(history, list) def test_get_marker_history_returns_copy(self): """get_marker_history returns a copy, not reference""" manager = SCTE35Manager() history1 = manager.get_marker_history() history2 = manager.get_marker_history() # Should be equal but different objects assert history1 == history2 @pytest.mark.asyncio async def test_clear_history(self): """clear_history empties the history list""" manager = SCTE35Manager() # Add some markers for i in range(3): await manager.inject_marker( event_id=i, duration_seconds=30 ) assert len(manager.get_marker_history()) == 3 # Clear manager.clear_history() assert len(manager.get_marker_history()) == 0 @pytest.mark.asyncio async def test_marker_timestamp_set(self): """Marker timestamp is set when created""" manager = SCTE35Manager() before = datetime.utcnow() marker = await manager.inject_marker( event_id=1, duration_seconds=30 ) after = datetime.utcnow() assert before <= marker.timestamp <= after @pytest.mark.asyncio async def test_inject_marker_webhook_exception_handling(self): """inject_marker handles webhook errors gracefully""" manager = SCTE35Manager() webhook_url = "https://example.com/webhook" with patch('aiohttp.ClientSession') as mock_session_class: mock_session = AsyncMock() # Setup post to raise an exception mock_session.post.side_effect = Exception("Connection failed") mock_session_class.return_value = AsyncMock() mock_session_class.return_value.__aenter__ = AsyncMock(return_value=mock_session) mock_session_class.return_value.__aexit__ = AsyncMock(return_value=None) # Should not raise, should still add marker to history marker = await manager.inject_marker( event_id=1, duration_seconds=30, webhook_url=webhook_url ) # Marker should still be added despite webhook failure assert len(manager.get_marker_history()) == 1