thomas.kopp
52e55e112f
The null sink and its two loopbacks were loaded at startup and stayed loaded forever. While loaded they link the microphone and the speaker device into a single PipeWire driver group: the microphone drives the graph and the speaker device runs as a clock follower, so every playback stream pays for continuous cross-device resampling. Nothing needs the chain while idle — recording is triggered manually. Build it in toggle_recording() and drop it as soon as the audio is captured, before transcription runs for minutes. Module ids are now derived from pactl instead of tracked in a state file, so a crashed process cannot leave stale ids behind. The state file only holds the device selection. Startup tears down any leftover chain. Loopbacks get latency_msec=200; transcription is offline, and a small buffer would put these nodes on the graph's realtime deadline.
146 lines
5.1 KiB
Python
146 lines
5.1 KiB
Python
import json
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def state_path(tmp_path, monkeypatch):
|
|
import pipewire
|
|
p = tmp_path / "pipewire-modules.json"
|
|
monkeypatch.setattr(pipewire, "STATE_PATH", p)
|
|
return p
|
|
|
|
|
|
def test_save_devices_persists_selection(state_path):
|
|
import pipewire
|
|
pipewire.save_devices("mic-src", "monitor-src")
|
|
data = json.loads(state_path.read_text())
|
|
assert data["mic"] == "mic-src"
|
|
assert data["monitor"] == "monitor-src"
|
|
|
|
|
|
def test_save_devices_does_not_load_modules(state_path):
|
|
import pipewire
|
|
with patch("subprocess.check_output") as run:
|
|
pipewire.save_devices("mic-src", "monitor-src")
|
|
run.assert_not_called()
|
|
|
|
|
|
def test_load_devices_returns_none_without_state(state_path):
|
|
import pipewire
|
|
assert pipewire.load_devices() is None
|
|
|
|
|
|
def test_load_devices_returns_none_on_incomplete_state(state_path):
|
|
import pipewire
|
|
state_path.write_text(json.dumps({"mic": "mic-src"}))
|
|
assert pipewire.load_devices() is None
|
|
|
|
|
|
def test_load_devices_roundtrip(state_path):
|
|
import pipewire
|
|
pipewire.save_devices("mic-src", "monitor-src")
|
|
assert pipewire.load_devices() == {"mic": "mic-src", "monitor": "monitor-src"}
|
|
|
|
|
|
def test_setup_without_configured_devices_does_nothing(state_path):
|
|
import pipewire
|
|
with patch("subprocess.check_output") as run:
|
|
assert pipewire.setup() == []
|
|
run.assert_not_called()
|
|
|
|
|
|
def test_setup_loads_sink_and_two_loopbacks(state_path):
|
|
import pipewire
|
|
pipewire.save_devices("mic-src", "monitor-src")
|
|
with patch.object(pipewire, "is_loaded", side_effect=[False, True]), \
|
|
patch("subprocess.check_output", side_effect=[b"11\n", b"12\n", b"13\n"]) as run:
|
|
ids = pipewire.setup()
|
|
assert ids == [11, 12, 13]
|
|
cmds = [c.args[0] for c in run.call_args_list]
|
|
assert cmds[0][:3] == ["pactl", "load-module", "module-null-sink"]
|
|
assert cmds[1][:3] == ["pactl", "load-module", "module-loopback"]
|
|
assert cmds[2][:3] == ["pactl", "load-module", "module-loopback"]
|
|
|
|
|
|
def test_setup_uses_configured_sources(state_path):
|
|
import pipewire
|
|
pipewire.save_devices("mic-src", "monitor-src")
|
|
with patch.object(pipewire, "is_loaded", side_effect=[False, True]), \
|
|
patch("subprocess.check_output", side_effect=[b"11\n", b"12\n", b"13\n"]) as run:
|
|
pipewire.setup()
|
|
cmds = [c.args[0] for c in run.call_args_list]
|
|
assert "source=mic-src" in cmds[1]
|
|
assert "source=monitor-src" in cmds[2]
|
|
|
|
|
|
def test_setup_applies_large_loopback_latency(state_path):
|
|
"""A small buffer would put the loopbacks on the graph's realtime deadline."""
|
|
import pipewire
|
|
pipewire.save_devices("mic-src", "monitor-src")
|
|
with patch.object(pipewire, "is_loaded", side_effect=[False, True]), \
|
|
patch("subprocess.check_output", side_effect=[b"11\n", b"12\n", b"13\n"]) as run:
|
|
pipewire.setup()
|
|
expected = f"latency_msec={pipewire.LOOPBACK_LATENCY_MSEC}"
|
|
for cmd in [c.args[0] for c in run.call_args_list][1:]:
|
|
assert expected in cmd
|
|
assert pipewire.LOOPBACK_LATENCY_MSEC >= 100
|
|
|
|
|
|
def test_setup_is_noop_when_already_loaded(state_path):
|
|
import pipewire
|
|
pipewire.save_devices("mic-src", "monitor-src")
|
|
with patch.object(pipewire, "is_loaded", return_value=True), \
|
|
patch("subprocess.check_output") as run:
|
|
assert pipewire.setup() == []
|
|
run.assert_not_called()
|
|
|
|
|
|
def test_teardown_unloads_chain_modules_highest_id_first():
|
|
import pipewire
|
|
listing = (
|
|
b"7\tmodule-something\tunrelated\n"
|
|
b"11\tmodule-null-sink\tsink_name=transkriptor-combined\n"
|
|
b"12\tmodule-loopback\tsource=mic-src sink=transkriptor-combined\n"
|
|
b"13\tmodule-loopback\tsource=monitor-src sink=transkriptor-combined\n"
|
|
)
|
|
with patch("subprocess.check_output", return_value=listing), \
|
|
patch("subprocess.run") as run:
|
|
pipewire.teardown()
|
|
unloaded = [c.args[0][-1] for c in run.call_args_list]
|
|
assert unloaded == ["13", "12", "11"]
|
|
|
|
|
|
def test_teardown_ignores_unrelated_modules():
|
|
import pipewire
|
|
listing = b"7\tmodule-something\tunrelated\n"
|
|
with patch("subprocess.check_output", return_value=listing), \
|
|
patch("subprocess.run") as run:
|
|
pipewire.teardown()
|
|
run.assert_not_called()
|
|
|
|
|
|
def test_teardown_survives_pactl_failure():
|
|
import pipewire
|
|
import subprocess as sp
|
|
with patch("subprocess.check_output", side_effect=sp.SubprocessError("boom")):
|
|
pipewire.teardown() # must not raise
|
|
|
|
|
|
def test_setup_survives_pactl_failure(state_path):
|
|
import pipewire
|
|
import subprocess as sp
|
|
pipewire.save_devices("mic-src", "monitor-src")
|
|
with patch.object(pipewire, "is_loaded", return_value=False), \
|
|
patch("subprocess.check_output", side_effect=sp.SubprocessError("boom")):
|
|
assert pipewire.setup() == []
|
|
|
|
|
|
def test_is_loaded_detects_sink():
|
|
import pipewire
|
|
with patch("subprocess.check_output", return_value=b"42\ttranskriptor-combined.monitor\n"):
|
|
assert pipewire.is_loaded() is True
|
|
with patch("subprocess.check_output", return_value=b"42\tsomething-else.monitor\n"):
|
|
assert pipewire.is_loaded() is False
|