150 lines
5.4 KiB
Python
150 lines
5.4 KiB
Python
"""
|
|
Tests for Chunker — time-based segmentation, chunk counts, sequence numbers, generator behavior.
|
|
|
|
Demonstrates: TDD (Interview Topic 8) — parametrized tests, edge cases, mocking.
|
|
"""
|
|
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
import pytest
|
|
|
|
from core.chunker import Chunker
|
|
from core.chunker.exceptions import ChunkReadError
|
|
|
|
|
|
def mock_probe(duration):
|
|
"""Create a mock probe_file that returns the given duration."""
|
|
result = MagicMock()
|
|
result.duration = duration
|
|
return result
|
|
|
|
|
|
class TestChunker:
|
|
@patch("core.chunker.chunker.probe_file")
|
|
def test_basic_chunking(self, mock_pf, temp_file):
|
|
"""File splits into expected number of time-based chunks."""
|
|
path = temp_file(b"x" * 1000)
|
|
mock_pf.return_value = mock_probe(30.0)
|
|
|
|
chunker = Chunker(path, chunk_duration=10.0)
|
|
chunks = list(chunker.chunks())
|
|
|
|
assert len(chunks) == 3
|
|
assert chunks[0].start_time == 0.0
|
|
assert chunks[0].end_time == 10.0
|
|
assert chunks[0].duration == 10.0
|
|
assert chunks[1].start_time == 10.0
|
|
assert chunks[2].start_time == 20.0
|
|
|
|
@patch("core.chunker.chunker.probe_file")
|
|
def test_sequence_numbers(self, mock_pf, temp_file):
|
|
"""Chunks have sequential sequence numbers starting at 0."""
|
|
path = temp_file(b"x" * 100)
|
|
mock_pf.return_value = mock_probe(40.0)
|
|
|
|
chunker = Chunker(path, chunk_duration=10.0)
|
|
chunks = list(chunker.chunks())
|
|
sequences = [c.sequence for c in chunks]
|
|
|
|
assert sequences == [0, 1, 2, 3]
|
|
|
|
@patch("core.chunker.chunker.probe_file")
|
|
def test_time_ranges(self, mock_pf, temp_file):
|
|
"""Each chunk has correct start_time and end_time."""
|
|
path = temp_file(b"x" * 100)
|
|
mock_pf.return_value = mock_probe(25.0)
|
|
|
|
chunker = Chunker(path, chunk_duration=10.0)
|
|
chunks = list(chunker.chunks())
|
|
|
|
assert chunks[0].start_time == 0.0
|
|
assert chunks[0].end_time == 10.0
|
|
assert chunks[1].start_time == 10.0
|
|
assert chunks[1].end_time == 20.0
|
|
assert chunks[2].start_time == 20.0
|
|
assert chunks[2].end_time == 25.0 # last chunk shorter
|
|
assert chunks[2].duration == 5.0
|
|
|
|
@patch("core.chunker.chunker.probe_file")
|
|
def test_expected_chunks_property(self, mock_pf, temp_file):
|
|
"""expected_chunks calculates correctly before iteration."""
|
|
path = temp_file(b"x" * 100)
|
|
mock_pf.return_value = mock_probe(25.0)
|
|
|
|
chunker = Chunker(path, chunk_duration=10.0)
|
|
assert chunker.expected_chunks == 3 # ceil(25/10)
|
|
|
|
@patch("core.chunker.chunker.probe_file")
|
|
def test_source_path_on_chunks(self, mock_pf, temp_file):
|
|
"""Each chunk carries the source file path."""
|
|
path = temp_file(b"x" * 100)
|
|
mock_pf.return_value = mock_probe(10.0)
|
|
|
|
chunker = Chunker(path, chunk_duration=10.0)
|
|
chunks = list(chunker.chunks())
|
|
|
|
assert all(c.source_path == path for c in chunks)
|
|
|
|
def test_file_not_found(self):
|
|
"""Non-existent file raises ChunkReadError."""
|
|
with pytest.raises(ChunkReadError, match="File not found"):
|
|
Chunker("/nonexistent/file.mp4")
|
|
|
|
@patch("core.chunker.chunker.probe_file")
|
|
def test_invalid_chunk_duration(self, mock_pf, temp_file):
|
|
"""Zero or negative chunk_duration raises ValueError."""
|
|
path = temp_file(b"x" * 100)
|
|
|
|
with pytest.raises(ValueError, match="chunk_duration must be positive"):
|
|
Chunker(path, chunk_duration=0)
|
|
|
|
with pytest.raises(ValueError, match="chunk_duration must be positive"):
|
|
Chunker(path, chunk_duration=-1)
|
|
|
|
@patch("core.chunker.chunker.probe_file")
|
|
def test_generator_laziness(self, mock_pf, temp_file):
|
|
"""Chunks are yielded lazily, not pre-loaded."""
|
|
path = temp_file(b"x" * 100)
|
|
mock_pf.return_value = mock_probe(30.0)
|
|
|
|
chunker = Chunker(path, chunk_duration=10.0)
|
|
gen = chunker.chunks()
|
|
first = next(gen)
|
|
assert first.sequence == 0
|
|
# Generator is not exhausted — remaining chunks still pending
|
|
|
|
@pytest.mark.parametrize("duration,chunk_dur,expected", [
|
|
(10.0, 10.0, 1),
|
|
(10.1, 10.0, 2),
|
|
(1.0, 1.0, 1),
|
|
(100.0, 1.0, 100),
|
|
(5.0, 100.0, 1),
|
|
])
|
|
@patch("core.chunker.chunker.probe_file")
|
|
def test_expected_chunks_parametrized(self, mock_pf, temp_file, duration, chunk_dur, expected):
|
|
"""Parametrized: various duration/chunk_duration combos."""
|
|
path = temp_file(b"x" * 100)
|
|
mock_pf.return_value = mock_probe(duration)
|
|
chunker = Chunker(path, chunk_duration=chunk_dur)
|
|
assert chunker.expected_chunks == expected
|
|
|
|
@patch("core.chunker.chunker.probe_file")
|
|
def test_exact_multiple(self, mock_pf, temp_file):
|
|
"""Duration exactly divisible by chunk_duration."""
|
|
path = temp_file(b"x" * 100)
|
|
mock_pf.return_value = mock_probe(30.0)
|
|
|
|
chunker = Chunker(path, chunk_duration=10.0)
|
|
chunks = list(chunker.chunks())
|
|
assert len(chunks) == 3
|
|
assert all(c.duration == 10.0 for c in chunks)
|
|
|
|
@patch("core.chunker.chunker.probe_file")
|
|
def test_probe_failure(self, mock_pf, temp_file):
|
|
"""Probe failure raises ChunkReadError."""
|
|
path = temp_file(b"x" * 100)
|
|
mock_pf.side_effect = Exception("ffprobe failed")
|
|
|
|
with pytest.raises(ChunkReadError, match="Failed to probe"):
|
|
Chunker(path, chunk_duration=10.0)
|