335 lines
13 KiB
Python
335 lines
13 KiB
Python
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
|