231 lines
8.7 KiB
Python
231 lines
8.7 KiB
Python
"""Tests for PartialAudioBuffer class."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import numpy as np
|
|
|
|
from noteflow.config.constants import DEFAULT_SAMPLE_RATE
|
|
from noteflow.infrastructure.audio.partial_buffer import PartialAudioBuffer
|
|
|
|
# Audio constants for test assertions
|
|
SAMPLE_RATE_16K = DEFAULT_SAMPLE_RATE
|
|
SAMPLE_RATE_48K = 48000
|
|
CHUNK_100MS = 1600 # 100ms at 16kHz
|
|
CHUNK_50MS = 800 # 50ms at 16kHz
|
|
ONE_SECOND_16K = DEFAULT_SAMPLE_RATE # 1 second at 16kHz
|
|
|
|
|
|
class TestPartialAudioBufferInit:
|
|
"""Tests for buffer initialization."""
|
|
|
|
def test_default_capacity(self) -> None:
|
|
"""Buffer should have 10 seconds capacity by default at 16kHz."""
|
|
buffer = PartialAudioBuffer()
|
|
expected_capacity = int(PartialAudioBuffer.DEFAULT_MAX_DURATION * SAMPLE_RATE_16K)
|
|
assert buffer.capacity_samples == expected_capacity, (
|
|
"Default capacity should match DEFAULT_MAX_DURATION"
|
|
)
|
|
|
|
def test_custom_duration(self) -> None:
|
|
"""Buffer should respect custom max duration."""
|
|
buffer = PartialAudioBuffer(max_duration_seconds=3.0, sample_rate=SAMPLE_RATE_16K)
|
|
assert buffer.capacity_samples == 3 * SAMPLE_RATE_16K, (
|
|
"Capacity should match custom duration"
|
|
)
|
|
|
|
def test_custom_sample_rate(self) -> None:
|
|
"""Buffer should respect custom sample rate."""
|
|
buffer = PartialAudioBuffer(max_duration_seconds=1.0, sample_rate=SAMPLE_RATE_48K)
|
|
assert buffer.capacity_samples == SAMPLE_RATE_48K, "Capacity should match sample rate"
|
|
assert buffer.sample_rate == SAMPLE_RATE_48K, "Sample rate property should match"
|
|
|
|
def test_starts_empty(self) -> None:
|
|
"""Buffer should start empty."""
|
|
buffer = PartialAudioBuffer()
|
|
assert buffer.is_empty, "New buffer should be empty"
|
|
assert buffer.samples_buffered == 0, "No samples should be buffered"
|
|
assert buffer.duration_seconds == 0.0, "Duration should be zero"
|
|
|
|
|
|
class TestPartialAudioBufferAppend:
|
|
"""Tests for buffer append operations."""
|
|
|
|
def test_append_single_chunk(self) -> None:
|
|
"""Appending a chunk should increase buffer size."""
|
|
buffer = PartialAudioBuffer(sample_rate=SAMPLE_RATE_16K)
|
|
chunk = np.ones(CHUNK_100MS, dtype=np.float32)
|
|
|
|
result = buffer.append(chunk)
|
|
|
|
assert result is True, "Append should succeed"
|
|
assert buffer.samples_buffered == CHUNK_100MS, "Should have 1600 samples"
|
|
assert not buffer.is_empty, "Buffer should not be empty"
|
|
|
|
def test_append_multiple_chunks(self) -> None:
|
|
"""Appending multiple chunks should accumulate."""
|
|
buffer = PartialAudioBuffer(sample_rate=SAMPLE_RATE_16K)
|
|
chunk = np.ones(CHUNK_100MS, dtype=np.float32)
|
|
|
|
for _ in range(10):
|
|
buffer.append(chunk)
|
|
|
|
assert buffer.samples_buffered == ONE_SECOND_16K, "Should have 10 chunks"
|
|
assert buffer.duration_seconds == 1.0, "Should be 1 second of audio"
|
|
|
|
def test_append_copies_data(self) -> None:
|
|
"""Append should copy data, not hold reference."""
|
|
buffer = PartialAudioBuffer(sample_rate=SAMPLE_RATE_16K)
|
|
chunk = np.ones(CHUNK_100MS, dtype=np.float32) * 0.5
|
|
|
|
buffer.append(chunk)
|
|
|
|
# Modify original - should not affect buffer
|
|
chunk[:] = 0.0
|
|
|
|
retrieved = buffer.get_audio()
|
|
assert np.allclose(retrieved, 0.5), "Buffer should have copied data"
|
|
|
|
def test_append_overflow_returns_false(self) -> None:
|
|
"""Append should return False on overflow."""
|
|
buffer = PartialAudioBuffer(max_duration_seconds=0.1, sample_rate=SAMPLE_RATE_16K)
|
|
# 0.1s = 1600 samples, try to add 2000
|
|
chunk = np.ones(2000, dtype=np.float32)
|
|
|
|
result = buffer.append(chunk)
|
|
|
|
assert result is False, "Overflow should return False"
|
|
assert buffer.is_empty, "Buffer should remain empty on overflow"
|
|
|
|
|
|
class TestPartialAudioBufferGetAudio:
|
|
"""Tests for audio retrieval."""
|
|
|
|
def test_get_audio_empty_buffer(self) -> None:
|
|
"""Getting audio from empty buffer should return empty array."""
|
|
buffer = PartialAudioBuffer()
|
|
|
|
audio = buffer.get_audio()
|
|
|
|
assert len(audio) == 0, "Empty buffer should return empty array"
|
|
assert audio.dtype == np.float32, "Should be float32"
|
|
|
|
def test_get_audio_returns_copy(self) -> None:
|
|
"""get_audio should return a copy, not a view."""
|
|
buffer = PartialAudioBuffer(sample_rate=SAMPLE_RATE_16K)
|
|
buffer.append(np.ones(CHUNK_100MS, dtype=np.float32))
|
|
|
|
audio1 = buffer.get_audio()
|
|
audio2 = buffer.get_audio()
|
|
|
|
# Modify first copy
|
|
audio1[:] = 0.0
|
|
|
|
# Second copy should be unaffected
|
|
assert np.allclose(audio2, 1.0), "Second copy should be independent"
|
|
|
|
def test_get_audio_view_is_view(self) -> None:
|
|
"""get_audio_view should return a view into buffer."""
|
|
buffer = PartialAudioBuffer(sample_rate=SAMPLE_RATE_16K)
|
|
buffer.append(np.ones(CHUNK_100MS, dtype=np.float32))
|
|
|
|
view = buffer.get_audio_view()
|
|
|
|
# Verify it's the same data
|
|
assert len(view) == CHUNK_100MS, "View should have correct length"
|
|
assert np.allclose(view, 1.0), "View should have correct values"
|
|
|
|
def test_get_audio_preserves_values(self) -> None:
|
|
"""Retrieved audio should match appended audio."""
|
|
buffer = PartialAudioBuffer(sample_rate=SAMPLE_RATE_16K)
|
|
chunk1 = np.full(CHUNK_100MS, 0.5, dtype=np.float32)
|
|
chunk2 = np.full(CHUNK_100MS, -0.5, dtype=np.float32)
|
|
|
|
buffer.append(chunk1)
|
|
buffer.append(chunk2)
|
|
|
|
audio = buffer.get_audio()
|
|
|
|
assert np.allclose(audio[:CHUNK_100MS], 0.5), "First chunk should match"
|
|
assert np.allclose(audio[CHUNK_100MS:], -0.5), "Second chunk should match"
|
|
|
|
|
|
class TestPartialAudioBufferClear:
|
|
"""Tests for buffer clearing."""
|
|
|
|
def test_clear_empties_buffer(self) -> None:
|
|
"""Clear should empty the buffer."""
|
|
buffer = PartialAudioBuffer(sample_rate=SAMPLE_RATE_16K)
|
|
buffer.append(np.ones(CHUNK_100MS, dtype=np.float32))
|
|
|
|
buffer.clear()
|
|
|
|
assert buffer.is_empty, "Buffer should be empty after clear"
|
|
assert buffer.samples_buffered == 0, "Sample count should be zero"
|
|
|
|
def test_clear_allows_reuse(self) -> None:
|
|
"""Buffer should be reusable after clear."""
|
|
buffer = PartialAudioBuffer(sample_rate=SAMPLE_RATE_16K)
|
|
buffer.append(np.ones(CHUNK_100MS, dtype=np.float32))
|
|
buffer.clear()
|
|
|
|
# Should be able to append again
|
|
buffer.append(np.full(CHUNK_50MS, 0.5, dtype=np.float32))
|
|
|
|
assert buffer.samples_buffered == CHUNK_50MS, "Should have new samples"
|
|
audio = buffer.get_audio()
|
|
assert np.allclose(audio, 0.5), "New data should be correct"
|
|
|
|
def test_clear_is_idempotent(self) -> None:
|
|
"""Multiple clears should be safe."""
|
|
buffer = PartialAudioBuffer()
|
|
|
|
buffer.clear()
|
|
buffer.clear()
|
|
buffer.clear()
|
|
|
|
assert buffer.is_empty, "Buffer should be empty"
|
|
|
|
|
|
class TestPartialAudioBufferDunderMethods:
|
|
"""Tests for special methods (__len__, __bool__)."""
|
|
|
|
def test_len_empty(self) -> None:
|
|
"""len() on empty buffer should return 0."""
|
|
buffer = PartialAudioBuffer()
|
|
assert len(buffer) == 0, "Empty buffer should have len 0"
|
|
|
|
def test_len_with_samples(self) -> None:
|
|
"""len() should return sample count."""
|
|
buffer = PartialAudioBuffer(sample_rate=SAMPLE_RATE_16K)
|
|
buffer.append(np.ones(CHUNK_100MS, dtype=np.float32))
|
|
assert len(buffer) == CHUNK_100MS, "len should match sample count"
|
|
|
|
def test_bool_empty(self) -> None:
|
|
"""Empty buffer should be falsy."""
|
|
buffer = PartialAudioBuffer()
|
|
assert not buffer, "Empty buffer should be falsy"
|
|
|
|
def test_bool_with_samples(self) -> None:
|
|
"""Buffer with samples should be truthy."""
|
|
buffer = PartialAudioBuffer(sample_rate=SAMPLE_RATE_16K)
|
|
buffer.append(np.ones(100, dtype=np.float32))
|
|
assert buffer, "Non-empty buffer should be truthy"
|
|
|
|
|
|
class TestPartialAudioBufferDuration:
|
|
"""Tests for duration calculation."""
|
|
|
|
def test_duration_calculation(self) -> None:
|
|
"""Duration should be calculated from samples and sample rate."""
|
|
buffer = PartialAudioBuffer(sample_rate=SAMPLE_RATE_16K)
|
|
buffer.append(np.ones(8000, dtype=np.float32))
|
|
|
|
assert buffer.duration_seconds == 0.5, "8000 samples at 16kHz = 0.5 seconds"
|
|
|
|
def test_duration_different_sample_rate(self) -> None:
|
|
"""Duration should respect sample rate."""
|
|
buffer = PartialAudioBuffer(sample_rate=SAMPLE_RATE_48K)
|
|
buffer.append(np.ones(48000, dtype=np.float32))
|
|
|
|
assert buffer.duration_seconds == 1.0, "48000 samples at 48kHz = 1 second"
|