mirrors/pipewire
1
0
Fork 0
mirror of https://gitlab.freedesktop.org/pipewire/pipewire.git synced 2026-05-12 23:50:35 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

test: improve bluezenv test debug

Browse source
This commit is contained in:
Pauli Virtanen 2026-05-09 14:04:11 +03:00 • committed by Wim Taymans
parent 1fc7ec2e3f
commit ef35aa161b
1 changed files with 36 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

43
test/bluezenv/test_streaming.py
View file

@ -100,22 +100,32 @@ class PipeWire(HostPlugin):
# get versions
res = subprocess.run(
[self.exe_pw, "--version"], stdout=subprocess.PIPE, encoding="utf-8"
[self.exe_pw, "--version"],
stdout=subprocess.PIPE,
encoding="utf-8",
check=True,
)
m = re.search("libpipewire ([0-9.]+)", res.stdout)
if m:
pw_version = tuple(int(x) for x in m.group(1).split("."))
else:
raise ValueError("pipewire version unknown")
raise ValueError(
f"pipewire {self.exe_pw} version unknown: {res.stdout}"
)
res = subprocess.run(
[self.exe_wp, "--version"], stdout=subprocess.PIPE, encoding="utf-8"
[self.exe_wp, "--version"],
stdout=subprocess.PIPE,
encoding="utf-8",
check=True,
)
m = re.search("libwireplumber ([0-9.]+)", res.stdout)
if m:
wp_version = tuple(int(x) for x in m.group(1).split("."))
else:
raise ValueError("wireplumber version unknown")
raise ValueError(
f"wireplumber {self.exe_wp} version unknown: {res.stdout}"
)
# check versions
if pw_version >= (1, 6, 0) and pw_version <= (1, 6, 2):
@ -208,6 +218,8 @@ class PipeWire(HostPlugin):
)
def cond():
self.check_running()
uuids = [str(uuid) for uuid in adapter.Get("org.bluez.Adapter1", "UUIDs")]
return all(uuid in uuids for uuid in self.uuids)
@ -221,6 +233,8 @@ class PipeWire(HostPlugin):
def cond():
nonlocal text
self.check_running()
text = self.pw_dump()
try:
data = json.loads(text)
@ -242,10 +256,23 @@ class PipeWire(HostPlugin):
log.info("PipeWire ready")
def check_running(self):
if self.pw.poll() is not None:
raise RuntimeError("PipeWire process terminated")
if self.wp.poll() is not None:
raise RuntimeError("Wireplumber process terminated")
def pw_dump(self):
ret = subprocess.run(
[self.exe_dump], stdout=subprocess.PIPE, encoding="utf-8", env=self.environ
)
try:
ret = subprocess.run(
[self.exe_dump],
stdout=subprocess.PIPE,
encoding="utf-8",
env=self.environ,
timeout=5,
)
except subprocess.TimeoutExpired:
return "ERROR: timeout"
return ret.stdout
@ -526,6 +553,8 @@ def check_pipewire_devices_exist(host, profile="a2dp-sink"):
def cond():
nonlocal text
host.pipewire.check_running()
text = host.pipewire.pw_dump()
try:
data = json.loads(text)