File size: 3,924 Bytes
1719c2a
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
from __future__ import annotations

from pathlib import Path

import pytest

from echocoach.recording import (
    ServerRecordingError,
    recording_level_warning,
    select_capture_backend,
    start_server_recording,
    stop_server_recording,
)


class _FakeProcess:
    def __init__(self, out_path: Path) -> None:
        self._running = True
        self.returncode: int | None = None
        self.stderr = None
        self._out_path = out_path

    def poll(self) -> int | None:
        return None if self._running else self.returncode

    def send_signal(self, sig: int) -> None:
        self._running = False
        self.returncode = 1  # pw-record exits 1 on SIGINT
        self._out_path.write_bytes(b"RIFF" + b"x" * 100)

    def wait(self, timeout: float | None = None) -> int:
        return 0


def test_select_capture_backend_prefers_pw_record(monkeypatch):
    monkeypatch.setattr("echocoach.recording._pw_record_available", lambda: True)
    monkeypatch.setattr("echocoach.recording._sounddevice_available", lambda: True)
    monkeypatch.setattr("echocoach.recording._arecord_available", lambda: True)
    assert select_capture_backend() == "pw-record"


class _FakeTimer:
    def __init__(self, *_args, **_kwargs) -> None:
        self.daemon = False

    def start(self) -> None:
        return None

    def cancel(self) -> None:
        return None


def test_start_stop_session(monkeypatch, tmp_path):
    import echocoach.recording as rec

    rec._session = None

    def fake_spawn(backend, path: Path):
        return _FakeProcess(path)

    monkeypatch.setattr("echocoach.recording.select_capture_backend", lambda: "pw-record")
    monkeypatch.setattr("echocoach.recording.outputs_dir", lambda: tmp_path)
    monkeypatch.setattr("echocoach.recording._spawn_capture_process", fake_spawn)
    monkeypatch.setattr("echocoach.recording.threading.Timer", _FakeTimer)

    start_server_recording(10)
    path = stop_server_recording()
    assert path.is_file()
    rec._session = None


def test_start_while_recording_raises(monkeypatch, tmp_path):
    import echocoach.recording as rec

    rec._session = None

    def fake_spawn(backend, path: Path):
        return _FakeProcess(path)

    monkeypatch.setattr("echocoach.recording.select_capture_backend", lambda: "pw-record")
    monkeypatch.setattr("echocoach.recording.outputs_dir", lambda: tmp_path)
    monkeypatch.setattr("echocoach.recording._spawn_capture_process", fake_spawn)
    monkeypatch.setattr("echocoach.recording.threading.Timer", _FakeTimer)

    start_server_recording(10)
    with pytest.raises(ServerRecordingError, match="Already recording"):
        start_server_recording(10)
    stop_server_recording()
    rec._session = None


def test_stop_without_start_raises():
    import echocoach.recording as rec

    rec._session = None
    rec._last_recording_path = None
    with pytest.raises(ServerRecordingError, match="Not recording"):
        stop_server_recording()


def test_recording_level_warning_detects_silence(tmp_path):
    import numpy as np
    import soundfile as sf

    silent = tmp_path / "silent.wav"
    sf.write(silent, np.zeros(16_000, dtype=np.float32), 16_000)
    assert "silent" in (recording_level_warning(silent) or "").lower()


def test_finalize_accepts_pw_record_exit_code_one(tmp_path, monkeypatch):
    import echocoach.recording as rec

    rec._session = None
    rec._last_recording_path = None

    out_path = tmp_path / "recordings" / "server_pw.wav"
    out_path.parent.mkdir(parents=True)
    fake_proc = _FakeProcess(out_path)

    session = rec._RecordingSession(
        process=fake_proc,
        out_path=out_path,
        backend="pw-record",
        max_seconds=10,
        started_at=0.0,
        watchdog=None,
    )
    rec._session = session
    path = rec.stop_server_recording()
    assert path == out_path
    assert path.stat().st_size > 44
    rec._session = None