From 9fd23fabb69ad53782935abee1519390e6a48048 Mon Sep 17 00:00:00 2001 From: Zac Gaetano Date: Tue, 14 Apr 2026 09:21:13 -0400 Subject: [PATCH] Add backend/tests/test_scte35.py --- backend/tests/test_scte35.py | 335 +++++++++++++++++++++++++++++++++++ 1 file changed, 335 insertions(+) create mode 100644 backend/tests/test_scte35.py diff --git a/backend/tests/test_scte35.py b/backend/tests/test_scte35.py new file mode 100644 index 0000000..f8d85bd --- /dev/null +++ b/backend/tests/test_scte35.py @@ -0,0 +1,335 @@ +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