Name the PipeWire the check is for, and keep it out of the tests

--raw arrived in 1.4, the same release that stopped reading a bare "-" as
raw on its own. The split is there rather than at 1.0: Ubuntu 24.10 and
anything else on 1.2 refuses the option too, and 1.4 onwards writes a
container around the stream without it.

Asking the installed binary meant the command tests ran pw-record --help
four times for real, on a machine whose answer decides what they see, which
the module docstring promises they never do. They pin the answer in setUp,
and the reading itself gets its own class, the empty help nobody covered
included.
This commit is contained in:
yusufipk
2026-08-05 18:41:23 +03:00
parent 49b940bcb9
commit f6754ebf43
2 changed files with 50 additions and 22 deletions
+6 -3
View File
@@ -458,9 +458,12 @@ def _pulse_record(target):
def _pw_record_raw_option(): def _pw_record_raw_option():
"""Use --raw only on pw-record releases that provide it. """Use --raw only on pw-record releases that provide it.
PipeWire 1.0, including Ubuntu 24.04's build, writes raw PCM to stdout but PipeWire gained --raw in 1.4, and in the same release stopped treating a
rejects the newer --raw option. A rejected option ends the recorder before filename of "-" as raw on its own: before it, the option is refused and the
it receives sound, so ask the installed binary which form it understands. recorder dies before any sound arrives; after it, leaving the option out
wraps the stream in a container the rest of this file would read as noise.
Ubuntu 24.04 and anything else still on 1.0 or 1.2 sit on the near side of
that line, so ask the installed binary which form it understands.
""" """
try: try:
result = subprocess.run( result = subprocess.run(
+44 -19
View File
@@ -238,6 +238,14 @@ class FakeProcess:
class RecordingCommand(OnLinux, DikteTest): class RecordingCommand(OnLinux, DikteTest):
"""Which program captures the microphone, and how it is asked to.""" """Which program captures the microphone, and how it is asked to."""
def setUp(self):
super().setUp()
# Whether pw-record takes --raw is read off the installed binary, and
# what is being tested here is the command rather than the machine the
# test is running on. PwRecordRawOption covers the reading itself.
self.enterContext(mock.patch.object(
audio, "_pw_record_raw_option", return_value=["--raw"]))
def test_parec_is_preferred(self): def test_parec_is_preferred(self):
"""It speaks to PulseAudio and to PipeWire's compatibility service, so """It speaks to PulseAudio and to PipeWire's compatibility service, so
it is the one that works on both desktops.""" it is the one that works on both desktops."""
@@ -248,23 +256,6 @@ class RecordingCommand(OnLinux, DikteTest):
with only_these_tools("pw-record"): with only_these_tools("pw-record"):
self.assertEqual(audio.recording_command()[0], "pw-record") self.assertEqual(audio.recording_command()[0], "pw-record")
def test_pw_record_uses_raw_when_the_installed_version_supports_it(self):
help_result = FakeCompleted(stdout=" --raw Write raw samples\n")
with only_these_tools("pw-record"), \
mock.patch.object(audio.subprocess, "run", return_value=help_result):
self.assertIn("--raw", audio.recording_command())
def test_pw_record_omits_raw_when_pipewire_1_0_rejects_it(self):
help_result = FakeCompleted(stdout=" --rate Sample rate\n")
with only_these_tools("pw-record"), \
mock.patch.object(audio.subprocess, "run", return_value=help_result):
self.assertNotIn("--raw", audio.recording_command())
def test_pw_record_help_failure_keeps_the_existing_command(self):
with only_these_tools("pw-record"), \
mock.patch.object(audio.subprocess, "run", side_effect=OSError):
self.assertIn("--raw", audio.recording_command())
def test_neither_is_installed(self): def test_neither_is_installed(self):
with only_these_tools(): with only_these_tools():
self.assertEqual(audio.recording_command(), []) self.assertEqual(audio.recording_command(), [])
@@ -305,9 +296,45 @@ class RecordingCommand(OnLinux, DikteTest):
if arg.startswith(flag)]) if arg.startswith(flag)])
class PwRecordRawOption(DikteTest):
"""Two pw-record generations want opposite commands for the same stream.
PipeWire 1.4 added --raw and stopped treating a filename of "-" as raw on
its own, so the option is refused by everything older and needed by
everything newer. The help text is the only thing that tells them apart.
"""
def option(self, **run):
with mock.patch.object(audio.subprocess, "run", **run):
return audio._pw_record_raw_option()
def test_a_version_that_offers_raw_is_asked_for_it(self):
self.assertEqual(["--raw"], self.option(
return_value=FakeCompleted(stdout=" -a, --raw RAW mode\n")))
def test_a_version_without_it_is_not(self):
self.assertEqual([], self.option(
return_value=FakeCompleted(stdout=" --rate Sample rate\n")))
def test_help_that_could_not_be_read_keeps_the_option(self):
"""Whatever is installed, the command that worked before this check
existed is the safer guess."""
self.assertEqual(["--raw"], self.option(side_effect=OSError))
self.assertEqual(["--raw"], self.option(
side_effect=subprocess.TimeoutExpired("pw-record", 2)))
def test_help_that_said_nothing_keeps_it_too(self):
self.assertEqual(["--raw"], self.option(return_value=FakeCompleted()))
class RecorderChain(OnLinux, DikteTest): class RecorderChain(OnLinux, DikteTest):
"""Start to WAV, with pw-record faked out.""" """Start to WAV, with pw-record faked out."""
def setUp(self):
super().setUp()
self.enterContext(mock.patch.object(
audio, "_pw_record_raw_option", return_value=["--raw"]))
def record(self, data, target="", max_seconds=300): def record(self, data, target="", max_seconds=300):
recorder = audio.Recorder() recorder = audio.Recorder()
results = [] results = []
@@ -316,7 +343,6 @@ class RecorderChain(OnLinux, DikteTest):
recorder.failed.connect(failures.append) recorder.failed.connect(failures.append)
proc = FakeProcess(data) proc = FakeProcess(data)
with only_these_tools("pw-record"), \ with only_these_tools("pw-record"), \
mock.patch.object(audio, "_pw_record_raw_option", return_value=[]), \
mock.patch.object(subprocess, "Popen", return_value=proc) as popen: mock.patch.object(subprocess, "Popen", return_value=proc) as popen:
recorder.start(target=target, max_seconds=max_seconds) recorder.start(target=target, max_seconds=max_seconds)
recorder._thread.join(timeout=5) recorder._thread.join(timeout=5)
@@ -363,7 +389,6 @@ class RecorderChain(OnLinux, DikteTest):
recorder.stopped.connect(lambda *args: results.append(args)) recorder.stopped.connect(lambda *args: results.append(args))
proc = FakeProcess(tone(1.0)) proc = FakeProcess(tone(1.0))
with only_these_tools("pw-record"), \ with only_these_tools("pw-record"), \
mock.patch.object(audio, "_pw_record_raw_option", return_value=[]), \
mock.patch.object(subprocess, "Popen", return_value=proc): mock.patch.object(subprocess, "Popen", return_value=proc):
recorder.start() recorder.start()
recorder._thread.join(timeout=5) recorder._thread.join(timeout=5)