diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/test_cli.py | 25 | ||||
| -rw-r--r-- | tests/test_runtime.py | 82 | ||||
| -rw-r--r-- | tests/test_segmenter.py | 96 | ||||
| -rw-r--r-- | tests/test_service.py | 60 | ||||
| -rw-r--r-- | tests/test_storage.py | 33 | ||||
| -rw-r--r-- | tests/test_vad.py | 19 |
6 files changed, 315 insertions, 0 deletions
diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..09de22c --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,25 @@ +from pathlib import Path + +from mic_clipper.cli import main + + +def test_service_install_dry_run_prints_commands_without_creating_unit(tmp_path, capsys): + unit_path = tmp_path / "mic-clipper.service" + + result = main( + [ + "service", + "install", + "--dry-run", + "--unit-path", + str(unit_path), + "--project-root", + str(tmp_path), + "--python", + "/runtime/python", + ] + ) + + assert result == 0 + assert not unit_path.exists() + assert "systemctl --user enable --now mic-clipper.service" in capsys.readouterr().out diff --git a/tests/test_runtime.py b/tests/test_runtime.py new file mode 100644 index 0000000..d0d45fb --- /dev/null +++ b/tests/test_runtime.py @@ -0,0 +1,82 @@ +from datetime import datetime, timezone + +import numpy as np + +from mic_clipper.audio_input import PulseAudioInput +from mic_clipper.runner import record_frames +from mic_clipper.segmenter import Segmenter + + +class FakeVad: + def __init__(self, probabilities: list[float]) -> None: + self.probabilities = iter(probabilities) + + def probability(self, _samples: np.ndarray) -> float: + return next(self.probabilities) + + +class FakeClip: + def __init__(self) -> None: + self.writes: list[np.ndarray] = [] + self.closed = False + + def write(self, samples: np.ndarray) -> None: + self.writes.append(samples) + + def close(self) -> None: + self.closed = True + + def abort(self) -> None: + self.closed = True + + +class FakeWriter: + def __init__(self) -> None: + self.opened: list[FakeClip] = [] + + def open(self, _started_at: datetime) -> FakeClip: + clip = FakeClip() + self.opened.append(clip) + return clip + + +def samples() -> np.ndarray: + return np.zeros(512, dtype=np.float32) + + +def test_silence_never_opens_an_output_clip(): + writer = FakeWriter() + + record_frames( + [samples(), samples()], + vad=FakeVad([0.1, 0.1]), + writer=writer, + segmenter=Segmenter(pre_roll_seconds=0.5, silence_seconds=1.5), + now=lambda: datetime(2026, 9, 29, tzinfo=timezone.utc), + ) + + assert writer.opened == [] + + +def test_end_event_closes_the_active_encoder(): + writer = FakeWriter() + + record_frames( + [samples()] + [samples()] * 47, + vad=FakeVad([0.9] + [0.1] * 47), + writer=writer, + segmenter=Segmenter(pre_roll_seconds=0.5, silence_seconds=1.5), + now=lambda: datetime(2026, 9, 29, tzinfo=timezone.utc), + ) + + assert len(writer.opened) == 1 + assert writer.opened[0].closed + + +def test_pulse_input_uses_dynamic_default_device_at_16khz_mono(): + command = PulseAudioInput.command() + + assert command[0] == "ffmpeg" + assert command[command.index("-f") + 1] == "pulse" + assert command[command.index("-i") + 1] == "default" + assert command[-2:] == ["f32le", "pipe:1"] diff --git a/tests/test_segmenter.py b/tests/test_segmenter.py new file mode 100644 index 0000000..d1f7ee5 --- /dev/null +++ b/tests/test_segmenter.py @@ -0,0 +1,96 @@ +from datetime import datetime, timedelta, timezone + +import numpy as np + +from mic_clipper.segmenter import Audio, End, Segmenter, Start + + +SAMPLE_RATE = 16_000 +FRAME_SAMPLES = 512 +FRAME_DURATION = timedelta(seconds=FRAME_SAMPLES / SAMPLE_RATE) +START = datetime(2026, 9, 29, 23, 59, 59, tzinfo=timezone.utc) + + +def frame(value: float = 0.0) -> np.ndarray: + return np.full(FRAME_SAMPLES, value, dtype=np.float32) + + +def push(segmenter: Segmenter, index: int, speech: bool, value: float = 0.0): + return segmenter.push(frame(value), speech, START + index * FRAME_DURATION) + + +def test_speech_starts_segment_with_half_second_preroll(): + segmenter = Segmenter(pre_roll_seconds=0.5, silence_seconds=1.5) + + for index in range(16): + assert push(segmenter, index, False) == () + + events = push(segmenter, 16, True, 1.0) + + assert len(events) == 1 + assert isinstance(events[0], Start) + assert events[0].started_at == START + 16 * FRAME_DURATION - timedelta(seconds=0.5) + assert events[0].samples.shape == (8_512,) + assert np.all(events[0].samples[:8_000] == 0.0) + assert np.all(events[0].samples[8_000:] == 1.0) + + +def test_short_silence_is_merged_into_active_segment(): + segmenter = Segmenter(pre_roll_seconds=0.5, silence_seconds=1.5) + + assert isinstance(push(segmenter, 0, True, 1.0)[0], Start) + assert isinstance(push(segmenter, 1, False)[0], Audio) + assert isinstance(push(segmenter, 2, True, 1.0)[0], Audio) + assert segmenter.active + + +def test_one_point_five_seconds_of_silence_closes_segment(): + segmenter = Segmenter(pre_roll_seconds=0.5, silence_seconds=1.5) + + push(segmenter, 0, True, 1.0) + events = () + for index in range(1, 48): + events = push(segmenter, index, False) + + assert isinstance(events[0], Audio) + assert isinstance(events[-1], End) + assert not segmenter.active + + +def test_maximum_segment_rolls_without_losing_audio(): + segmenter = Segmenter( + pre_roll_seconds=0.5, + silence_seconds=1.5, + max_segment_seconds=FRAME_SAMPLES * 2 / SAMPLE_RATE, + ) + + first = push(segmenter, 0, True, 1.0) + second = push(segmenter, 1, True, 2.0) + third = push(segmenter, 2, True, 3.0) + + assert isinstance(first[0], Start) + assert isinstance(second[0], Audio) + assert isinstance(third[0], End) + assert isinstance(third[1], Start) + assert np.all(third[1].samples == 3.0) + + +def test_segment_start_date_is_retained_across_midnight(): + segmenter = Segmenter(pre_roll_seconds=0.5, silence_seconds=1.5) + + events = push(segmenter, 16, True, 1.0) + + assert isinstance(events[0], Start) + assert events[0].started_at.date().isoformat() == "2026-09-29" + + +def test_speech_after_a_closed_segment_keeps_terminal_silence_as_preroll(): + segmenter = Segmenter(pre_roll_seconds=0.5, silence_seconds=1.5) + + push(segmenter, 0, True, 1.0) + for index in range(1, 48): + push(segmenter, index, False) + events = push(segmenter, 48, True, 1.0) + + assert isinstance(events[0], Start) + assert events[0].samples.shape == (8_512,) diff --git a/tests/test_service.py b/tests/test_service.py new file mode 100644 index 0000000..2bda2ab --- /dev/null +++ b/tests/test_service.py @@ -0,0 +1,60 @@ +from pathlib import Path + +from mic_clipper.service import install, render_unit, uninstall + + +def test_rendered_service_waits_for_audio_services_and_restarts_only_on_failure(): + unit = render_unit(Path("/runtime/python"), Path("/project")) + + assert "After=pipewire.service wireplumber.service pipewire-pulse.service" in unit + assert "Restart=on-failure" in unit + assert "RestartMaxDelaySec=60s" in unit + assert "ExecStart=/runtime/python -m mic_clipper run" in unit + assert "UMask=0077" in unit + + +def test_install_and_uninstall_only_manage_the_tool_unit(tmp_path): + unit_path = tmp_path / "systemd" / "mic-clipper.service" + clips = tmp_path / "Mic Clips" / "2026-09-29" / "keep.opus" + clips.parent.mkdir(parents=True) + clips.write_bytes(b"recording") + commands: list[list[str]] = [] + + install( + unit_path=unit_path, + python=Path("/runtime/python"), + project_root=Path("/project"), + run_command=commands.append, + ) + install( + unit_path=unit_path, + python=Path("/runtime/python"), + project_root=Path("/project"), + run_command=commands.append, + ) + uninstall(unit_path=unit_path, run_command=commands.append) + + assert not unit_path.exists() + assert clips.read_bytes() == b"recording" + assert commands == [ + ["systemctl", "--user", "daemon-reload"], + ["systemctl", "--user", "enable", "--now", "mic-clipper.service"], + ["systemctl", "--user", "daemon-reload"], + ["systemctl", "--user", "enable", "--now", "mic-clipper.service"], + ["systemctl", "--user", "disable", "--now", "mic-clipper.service"], + ["systemctl", "--user", "daemon-reload"], + ] + + +def test_dry_run_does_not_write_a_unit(tmp_path): + unit_path = tmp_path / "mic-clipper.service" + + commands = install( + unit_path=unit_path, + python=Path("/runtime/python"), + project_root=Path("/project"), + dry_run=True, + ) + + assert not unit_path.exists() + assert commands[-1] == ["systemctl", "--user", "enable", "--now", "mic-clipper.service"] diff --git a/tests/test_storage.py b/tests/test_storage.py new file mode 100644 index 0000000..389f199 --- /dev/null +++ b/tests/test_storage.py @@ -0,0 +1,33 @@ +import os +import subprocess +from datetime import datetime, timezone + +import numpy as np + +from mic_clipper.storage import ClipWriter + + +def test_writer_creates_private_opus_clip_in_segment_start_date(tmp_path): + started_at = datetime(2026, 9, 29, 23, 59, 59, tzinfo=timezone.utc) + samples = np.sin(np.linspace(0, 100, 16_000, dtype=np.float32)) + writer = ClipWriter(tmp_path) + + clip = writer.open(started_at) + clip.write(samples) + clip.close() + + destination = tmp_path / "2026-09-29" + files = list(destination.glob("*.opus")) + assert len(files) == 1 + assert files[0].name.startswith("23-59-59") + assert os.stat(destination).st_mode & 0o777 == 0o700 + assert os.stat(files[0]).st_mode & 0o777 == 0o600 + assert files[0].read_bytes()[:4] == b"OggS" + assert b"OpusHead" in files[0].read_bytes()[:128] + + decoded = subprocess.run( + ["ffmpeg", "-v", "error", "-i", str(files[0]), "-f", "f32le", "pipe:1"], + check=True, + capture_output=True, + ) + assert len(decoded.stdout) > 0 diff --git a/tests/test_vad.py b/tests/test_vad.py new file mode 100644 index 0000000..98826ae --- /dev/null +++ b/tests/test_vad.py @@ -0,0 +1,19 @@ +import numpy as np +import pytest + +from mic_clipper.vad import FRAME_SAMPLES, SileroVad + + +def test_bundled_silero_vad_accepts_512_samples_and_returns_probability(): + vad = SileroVad() + + probability = vad.probability(np.zeros(FRAME_SAMPLES, dtype=np.float32)) + + assert 0.0 <= probability <= 1.0 + + +def test_bundled_silero_vad_rejects_non_512_sample_frames(): + vad = SileroVad() + + with pytest.raises(ValueError, match="512"): + vad.probability(np.zeros(FRAME_SAMPLES - 1, dtype=np.float32)) |
