linux observer
activity,screencast: replace silent-degrade DBus catches with observable probes
activity.py and screencast.py hid DBus probe failures behind broad
`except Exception: pass` and `return <literal>` fallbacks. Parser bugs,
bus-daemon dropouts, and timeouts all looked identical to "service not
present", which dropped operator signal and could silently flip
probe_activity_services() into always-capture mode when a real backend
was present but introspection broke.
Fix: inherit the doctor.check_portal pattern from 1629914. Service
presence probing now asks the bus daemon (org.freedesktop.DBus at
/org/freedesktop/DBus) whether a well-known name is owned via
NameHasOwner instead of introspecting each target service path. The
bus-daemon XML has no hyphenated members, so the parser bug does not
apply, and the code can distinguish "not registered" from "probe
broke".
This fixes seven sites: activity.probe_activity_services now uses
NameHasOwner against the bus daemon; activity.is_screen_locked narrows
both the FDO and GNOME fallback branches; activity.is_power_save_active
does the same for Mutter and KDE Solid; activity.get_monitor_geometries_kscreen
narrows its catch to concrete DBus/introspection failures; and
screencast.Screencaster._close_session now warns with the portal session
path instead of silently swallowing close failures.
One design choice is intentional: backend branches in the lock/power
fallback chains and KScreen suppress WARN logs when DBusError.type is
org.freedesktop.DBus.Error.ServiceUnknown or
org.freedesktop.DBus.Error.NameHasNoOwner. Without that, asymmetric
desktops such as vanilla GNOME (no KDE Solid) or KDE (no Mutter
DisplayConfig) would emit thousands of warnings per day from the 5s poll
loop. Any other probe failure stays loud: parser bugs, bus failures,
non-missing DBusError values, and close-session failures still warn.
_close_session and _name_has_owner log unconditionally because closing a
known session should work, and NameHasOwner returns False rather than
raising for truly missing services.
This closes the follow-up called out in the footer of 1629914, and adds
regression coverage for parser failures, broken backends, silent missing
services, and close-session logging.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
+415
-36
+120
-12
src/solstone_linux/activity.py
+120
-12
src/solstone_linux/activity.py
···
9
9
running regardless of desktop environment.
10
10
"""
11
11
12
+
import asyncio
12
13
import logging
13
14
import os
14
15
15
16
from dbus_next import Variant
16
17
from dbus_next.aio import MessageBus
18
+
from dbus_next.errors import (
19
+
DBusError,
20
+
InvalidIntrospectionError,
21
+
InvalidMemberNameError,
22
+
)
17
23
18
24
logger = logging.getLogger(__name__)
25
+
26
+
_DBUS_PROBE_TIMEOUT_SEC = 2.0
27
+
28
+
_SERVICE_MISSING_ERRORS = (
29
+
"org.freedesktop.DBus.Error.ServiceUnknown",
30
+
"org.freedesktop.DBus.Error.NameHasNoOwner",
31
+
)
19
32
20
33
# GTK4/GDK4 — optional, only needed for monitor geometry detection.
21
34
# On systems without GTK4, get_monitor_geometries() will raise RuntimeError
···
55
68
KSCREEN_IFACE = "org.kde.kscreen.Backend"
56
69
57
70
71
+
def _is_service_missing(exc: BaseException) -> bool:
72
+
"""True if exc is a DBusError meaning the bus name is not currently owned."""
73
+
return (
74
+
isinstance(exc, DBusError)
75
+
and getattr(exc, "type", "") in _SERVICE_MISSING_ERRORS
76
+
)
77
+
78
+
79
+
async def _name_has_owner(bus: MessageBus, bus_name: str) -> bool:
80
+
"""Ask the bus daemon whether a well-known name is currently owned.
81
+
82
+
Returns False on any probe failure (daemon unreachable, timeout, parser
83
+
error) after logging a warning — the service is treated as absent.
84
+
"""
85
+
86
+
async def _probe() -> bool:
87
+
intro = await bus.introspect("org.freedesktop.DBus", "/org/freedesktop/DBus")
88
+
obj = bus.get_proxy_object(
89
+
"org.freedesktop.DBus", "/org/freedesktop/DBus", intro
90
+
)
91
+
iface = obj.get_interface("org.freedesktop.DBus")
92
+
return bool(await iface.call_name_has_owner(bus_name))
93
+
94
+
try:
95
+
return await asyncio.wait_for(_probe(), timeout=_DBUS_PROBE_TIMEOUT_SEC)
96
+
except (DBusError, InvalidMemberNameError, OSError, asyncio.TimeoutError) as exc:
97
+
logger.warning(
98
+
"NameHasOwner probe failed: service=%s path=%s: %s: %s",
99
+
bus_name,
100
+
"/org/freedesktop/DBus",
101
+
type(exc).__name__,
102
+
exc,
103
+
)
104
+
return False
105
+
106
+
58
107
async def probe_activity_services(bus: MessageBus) -> dict[str, bool]:
59
108
"""Check which activity DBus services are reachable."""
60
109
services = {
···
66
115
}
67
116
results = {}
68
117
for name, bus_name in services.items():
69
-
try:
70
-
await bus.introspect(bus_name, "/")
71
-
results[name] = True
72
-
except Exception:
73
-
results[name] = False
118
+
results[name] = await _name_has_owner(bus, bus_name)
74
119
75
120
# Log grouped by function
76
121
lock_backends = ["fdo_screensaver", "gnome_screensaver"]
···
110
155
obj = bus.get_proxy_object(FDO_SCREENSAVER_BUS, FDO_SCREENSAVER_PATH, intro)
111
156
iface = obj.get_interface(FDO_SCREENSAVER_IFACE)
112
157
return bool(await iface.call_get_active())
113
-
except Exception:
114
-
pass
158
+
except (
159
+
DBusError,
160
+
InvalidMemberNameError,
161
+
InvalidIntrospectionError,
162
+
OSError,
163
+
) as exc:
164
+
if not _is_service_missing(exc):
165
+
logger.warning(
166
+
"is_screen_locked FDO backend failed: service=%s path=%s: %s: %s",
167
+
FDO_SCREENSAVER_BUS,
168
+
FDO_SCREENSAVER_PATH,
169
+
type(exc).__name__,
170
+
exc,
171
+
)
115
172
116
173
# Fall back to GNOME ScreenSaver
117
174
try:
···
119
176
obj = bus.get_proxy_object(GNOME_SCREENSAVER_BUS, GNOME_SCREENSAVER_PATH, intro)
120
177
iface = obj.get_interface(GNOME_SCREENSAVER_IFACE)
121
178
return bool(await iface.call_get_active())
122
-
except Exception:
179
+
except (
180
+
DBusError,
181
+
InvalidMemberNameError,
182
+
InvalidIntrospectionError,
183
+
OSError,
184
+
) as exc:
185
+
if not _is_service_missing(exc):
186
+
logger.warning(
187
+
"is_screen_locked GNOME backend failed: service=%s path=%s: %s: %s",
188
+
GNOME_SCREENSAVER_BUS,
189
+
GNOME_SCREENSAVER_PATH,
190
+
type(exc).__name__,
191
+
exc,
192
+
)
123
193
return False
124
194
125
195
···
136
206
mode_variant = await iface.call_get(DISPLAY_CONFIG_IFACE, "PowerSaveMode")
137
207
mode = int(mode_variant.value)
138
208
return mode != 0
139
-
except Exception:
140
-
pass
209
+
except (
210
+
DBusError,
211
+
InvalidMemberNameError,
212
+
InvalidIntrospectionError,
213
+
OSError,
214
+
) as exc:
215
+
if not _is_service_missing(exc):
216
+
logger.warning(
217
+
"is_power_save_active Mutter backend failed: service=%s path=%s: %s: %s",
218
+
DISPLAY_CONFIG_BUS,
219
+
DISPLAY_CONFIG_PATH,
220
+
type(exc).__name__,
221
+
exc,
222
+
)
141
223
142
224
# Fall back to KDE Solid PowerManagement
143
225
try:
···
145
227
obj = bus.get_proxy_object(KDE_POWER_BUS, KDE_POWER_PATH, intro)
146
228
iface = obj.get_interface(KDE_POWER_IFACE)
147
229
return bool(await iface.call_is_lid_closed())
148
-
except Exception:
230
+
except (
231
+
DBusError,
232
+
InvalidMemberNameError,
233
+
InvalidIntrospectionError,
234
+
OSError,
235
+
) as exc:
236
+
if not _is_service_missing(exc):
237
+
logger.warning(
238
+
"is_power_save_active KDE backend failed: service=%s path=%s: %s: %s",
239
+
KDE_POWER_BUS,
240
+
KDE_POWER_PATH,
241
+
type(exc).__name__,
242
+
exc,
243
+
)
149
244
return False
150
245
151
246
···
258
353
monitors = assign_monitor_positions(geometries)
259
354
logger.debug("KScreen monitor geometries found: %d", len(monitors))
260
355
return monitors
261
-
except Exception:
356
+
except (
357
+
DBusError,
358
+
InvalidMemberNameError,
359
+
InvalidIntrospectionError,
360
+
OSError,
361
+
) as exc:
362
+
if not _is_service_missing(exc):
363
+
logger.warning(
364
+
"get_monitor_geometries_kscreen failed: service=%s path=%s: %s: %s",
365
+
KSCREEN_BUS,
366
+
KSCREEN_PATH,
367
+
type(exc).__name__,
368
+
exc,
369
+
)
262
370
return []
+18
-2
src/solstone_linux/screencast.py
+18
-2
src/solstone_linux/screencast.py
···
32
32
from dbus_next import Variant, introspection
33
33
from dbus_next.aio import MessageBus
34
34
from dbus_next.constants import BusType
35
+
from dbus_next.errors import (
36
+
DBusError,
37
+
InvalidIntrospectionError,
38
+
InvalidMemberNameError,
39
+
)
35
40
36
41
# Workaround for dbus-next issue #122: portal has properties with hyphens
37
42
# (e.g., "power-saver-enabled") which violate strict D-Bus naming validation.
···
543
548
)
544
549
session_iface = session_obj.get_interface(SESSION_IFACE)
545
550
await session_iface.call_close()
546
-
except Exception:
547
-
pass
551
+
except (
552
+
DBusError,
553
+
InvalidMemberNameError,
554
+
InvalidIntrospectionError,
555
+
OSError,
556
+
) as exc:
557
+
logger.warning(
558
+
"_close_session failed: service=%s path=%s: %s: %s",
559
+
PORTAL_BUS,
560
+
self.session_handle,
561
+
type(exc).__name__,
562
+
exc,
563
+
)
548
564
self.session_handle = None
549
565
550
566
def is_healthy(self) -> bool:
+243
-21
tests/test_activity.py
+243
-21
tests/test_activity.py
···
7
7
from unittest.mock import AsyncMock, MagicMock, call
8
8
9
9
import pytest
10
+
from dbus_next.errors import DBusError, InvalidMemberNameError
10
11
11
12
from solstone_linux import activity
12
13
···
23
24
return variant
24
25
25
26
27
+
def _service_unknown(detail: str) -> DBusError:
28
+
return DBusError("org.freedesktop.DBus.Error.ServiceUnknown", detail)
29
+
30
+
31
+
def _no_reply(detail: str) -> DBusError:
32
+
return DBusError("org.freedesktop.DBus.Error.NoReply", detail)
33
+
34
+
35
+
def _make_name_has_owner_bus(
36
+
*, return_value: bool | None = None, side_effect=None
37
+
) -> tuple[MagicMock, MagicMock]:
38
+
bus = MagicMock()
39
+
bus.introspect = AsyncMock(return_value=object())
40
+
iface = MagicMock()
41
+
if side_effect is not None:
42
+
iface.call_name_has_owner = AsyncMock(side_effect=side_effect)
43
+
else:
44
+
iface.call_name_has_owner = AsyncMock(return_value=return_value)
45
+
bus.get_proxy_object.return_value = _make_proxy_with_interface(iface)
46
+
return bus, iface
47
+
48
+
26
49
class TestIsScreenLocked:
27
50
"""Test screen lock fallback order."""
28
51
···
61
84
@pytest.mark.asyncio
62
85
async def test_fdo_failure_gnome_returns_true(self):
63
86
bus = MagicMock()
64
-
bus.introspect = AsyncMock(side_effect=[Exception("fdo unavailable"), object()])
87
+
bus.introspect = AsyncMock(
88
+
side_effect=[_service_unknown("fdo unavailable"), object()]
89
+
)
65
90
gnome_iface = MagicMock()
66
91
gnome_iface.call_get_active = AsyncMock(return_value=True)
67
92
bus.get_proxy_object.return_value = _make_proxy_with_interface(gnome_iface)
···
77
102
@pytest.mark.asyncio
78
103
async def test_fdo_failure_gnome_returns_false(self):
79
104
bus = MagicMock()
80
-
bus.introspect = AsyncMock(side_effect=[Exception("fdo unavailable"), object()])
105
+
bus.introspect = AsyncMock(
106
+
side_effect=[_service_unknown("fdo unavailable"), object()]
107
+
)
81
108
gnome_iface = MagicMock()
82
109
gnome_iface.call_get_active = AsyncMock(return_value=False)
83
110
bus.get_proxy_object.return_value = _make_proxy_with_interface(gnome_iface)
···
94
121
async def test_both_backends_fail_returns_false(self):
95
122
bus = MagicMock()
96
123
bus.introspect = AsyncMock(
97
-
side_effect=[Exception("fdo unavailable"), Exception("gnome unavailable")]
124
+
side_effect=[
125
+
_service_unknown("fdo unavailable"),
126
+
_service_unknown("gnome unavailable"),
127
+
]
98
128
)
99
129
100
130
result = await activity.is_screen_locked(bus)
···
105
135
call(activity.GNOME_SCREENSAVER_BUS, activity.GNOME_SCREENSAVER_PATH),
106
136
]
107
137
138
+
@pytest.mark.asyncio
139
+
async def test_is_screen_locked_fdo_parser_error_falls_through_to_gnome(
140
+
self, caplog
141
+
):
142
+
bus = MagicMock()
143
+
bus.introspect = AsyncMock(
144
+
side_effect=[InvalidMemberNameError("bad"), object()]
145
+
)
146
+
gnome_iface = MagicMock()
147
+
gnome_iface.call_get_active = AsyncMock(return_value=True)
148
+
bus.get_proxy_object.return_value = _make_proxy_with_interface(gnome_iface)
149
+
150
+
with caplog.at_level(logging.WARNING):
151
+
result = await activity.is_screen_locked(bus)
152
+
153
+
assert result is True
154
+
assert [record.message for record in caplog.records] == [
155
+
"is_screen_locked FDO backend failed: "
156
+
"service=org.freedesktop.ScreenSaver path=/ScreenSaver: "
157
+
"InvalidMemberNameError: invalid member name: bad"
158
+
]
159
+
160
+
@pytest.mark.parametrize(
161
+
"error_name",
162
+
[
163
+
"org.freedesktop.DBus.Error.ServiceUnknown",
164
+
"org.freedesktop.DBus.Error.NameHasNoOwner",
165
+
],
166
+
)
167
+
@pytest.mark.asyncio
168
+
async def test_is_screen_locked_service_missing_does_not_log(
169
+
self, caplog, error_name
170
+
):
171
+
bus = MagicMock()
172
+
bus.introspect = AsyncMock(
173
+
side_effect=[
174
+
DBusError(error_name, "missing"),
175
+
DBusError(error_name, "missing"),
176
+
]
177
+
)
178
+
179
+
with caplog.at_level(logging.WARNING):
180
+
result = await activity.is_screen_locked(bus)
181
+
182
+
assert result is False
183
+
assert caplog.records == []
184
+
185
+
@pytest.mark.asyncio
186
+
async def test_is_screen_locked_both_backends_broken_logs_both_warnings(
187
+
self, caplog
188
+
):
189
+
bus = MagicMock()
190
+
bus.introspect = AsyncMock(side_effect=[_no_reply("broke"), _no_reply("broke")])
191
+
192
+
with caplog.at_level(logging.WARNING):
193
+
result = await activity.is_screen_locked(bus)
194
+
195
+
assert result is False
196
+
assert [record.message for record in caplog.records] == [
197
+
"is_screen_locked FDO backend failed: "
198
+
"service=org.freedesktop.ScreenSaver path=/ScreenSaver: "
199
+
"DBusError: broke",
200
+
"is_screen_locked GNOME backend failed: "
201
+
"service=org.gnome.ScreenSaver path=/org/gnome/ScreenSaver: "
202
+
"DBusError: broke",
203
+
]
204
+
108
205
109
206
class TestIsPowerSaveActive:
110
207
"""Test power save fallback order."""
···
143
240
async def test_gnome_failure_kde_lid_closed_returns_true(self):
144
241
bus = MagicMock()
145
242
bus.introspect = AsyncMock(
146
-
side_effect=[Exception("gnome unavailable"), object()]
243
+
side_effect=[_service_unknown("gnome unavailable"), object()]
147
244
)
148
245
kde_iface = MagicMock()
149
246
kde_iface.call_is_lid_closed = AsyncMock(return_value=True)
···
161
258
async def test_gnome_failure_kde_lid_open_returns_false(self):
162
259
bus = MagicMock()
163
260
bus.introspect = AsyncMock(
164
-
side_effect=[Exception("gnome unavailable"), object()]
261
+
side_effect=[_service_unknown("gnome unavailable"), object()]
165
262
)
166
263
kde_iface = MagicMock()
167
264
kde_iface.call_is_lid_closed = AsyncMock(return_value=False)
···
179
276
async def test_both_backends_fail_returns_false(self):
180
277
bus = MagicMock()
181
278
bus.introspect = AsyncMock(
182
-
side_effect=[Exception("gnome unavailable"), Exception("kde unavailable")]
279
+
side_effect=[
280
+
_service_unknown("gnome unavailable"),
281
+
_service_unknown("kde unavailable"),
282
+
]
183
283
)
184
284
185
285
result = await activity.is_power_save_active(bus)
···
190
290
call(activity.KDE_POWER_BUS, activity.KDE_POWER_PATH),
191
291
]
192
292
293
+
@pytest.mark.asyncio
294
+
async def test_is_power_save_active_mutter_parser_error_falls_through_to_kde(
295
+
self, caplog
296
+
):
297
+
bus = MagicMock()
298
+
bus.introspect = AsyncMock(
299
+
side_effect=[InvalidMemberNameError("bad"), object()]
300
+
)
301
+
kde_iface = MagicMock()
302
+
kde_iface.call_is_lid_closed = AsyncMock(return_value=True)
303
+
bus.get_proxy_object.return_value = _make_proxy_with_interface(kde_iface)
304
+
305
+
with caplog.at_level(logging.WARNING):
306
+
result = await activity.is_power_save_active(bus)
307
+
308
+
assert result is True
309
+
assert [record.message for record in caplog.records] == [
310
+
"is_power_save_active Mutter backend failed: "
311
+
"service=org.gnome.Mutter.DisplayConfig "
312
+
"path=/org/gnome/Mutter/DisplayConfig: "
313
+
"InvalidMemberNameError: invalid member name: bad"
314
+
]
315
+
316
+
@pytest.mark.parametrize(
317
+
"error_name",
318
+
[
319
+
"org.freedesktop.DBus.Error.ServiceUnknown",
320
+
"org.freedesktop.DBus.Error.NameHasNoOwner",
321
+
],
322
+
)
323
+
@pytest.mark.asyncio
324
+
async def test_is_power_save_active_service_missing_does_not_log(
325
+
self, caplog, error_name
326
+
):
327
+
bus = MagicMock()
328
+
bus.introspect = AsyncMock(
329
+
side_effect=[
330
+
DBusError(error_name, "missing"),
331
+
DBusError(error_name, "missing"),
332
+
]
333
+
)
334
+
335
+
with caplog.at_level(logging.WARNING):
336
+
result = await activity.is_power_save_active(bus)
337
+
338
+
assert result is False
339
+
assert caplog.records == []
340
+
341
+
@pytest.mark.asyncio
342
+
async def test_is_power_save_active_both_backends_broken_logs_both_warnings(
343
+
self, caplog
344
+
):
345
+
bus = MagicMock()
346
+
bus.introspect = AsyncMock(side_effect=[_no_reply("broke"), _no_reply("broke")])
347
+
348
+
with caplog.at_level(logging.WARNING):
349
+
result = await activity.is_power_save_active(bus)
350
+
351
+
assert result is False
352
+
assert [record.message for record in caplog.records] == [
353
+
"is_power_save_active Mutter backend failed: "
354
+
"service=org.gnome.Mutter.DisplayConfig "
355
+
"path=/org/gnome/Mutter/DisplayConfig: DBusError: broke",
356
+
"is_power_save_active KDE backend failed: "
357
+
"service=org.kde.Solid.PowerManagement "
358
+
"path=/org/kde/Solid/PowerManagement: DBusError: broke",
359
+
]
360
+
193
361
194
362
class TestProbeActivityServices:
195
363
"""Test activity backend probing and logging."""
196
364
197
365
@pytest.mark.asyncio
198
366
async def test_all_services_available_returns_true_results(self):
199
-
bus = MagicMock()
200
-
bus.introspect = AsyncMock(return_value=object())
367
+
bus, _ = _make_name_has_owner_bus(return_value=True)
201
368
202
369
results = await activity.probe_activity_services(bus)
203
370
···
210
377
211
378
@pytest.mark.asyncio
212
379
async def test_no_services_available_logs_warning(self, caplog):
213
-
bus = MagicMock()
214
-
bus.introspect = AsyncMock(side_effect=Exception("missing"))
380
+
bus, _ = _make_name_has_owner_bus(return_value=False)
215
381
216
382
with caplog.at_level(logging.WARNING):
217
383
results = await activity.probe_activity_services(bus)
···
225
391
226
392
@pytest.mark.asyncio
227
393
async def test_mixed_service_availability_returns_correct_results(self):
228
-
bus = MagicMock()
229
-
bus.introspect = AsyncMock(
230
-
side_effect=[
231
-
object(),
232
-
Exception("missing"),
233
-
object(),
234
-
Exception("missing"),
235
-
object(),
236
-
]
237
-
)
394
+
bus, _ = _make_name_has_owner_bus(side_effect=[True, False, True, False, True])
238
395
239
396
results = await activity.probe_activity_services(bus)
240
397
···
244
401
assert results["kde_power"] is False
245
402
assert results["kscreen"] is True
246
403
404
+
@pytest.mark.asyncio
405
+
async def test_probe_activity_services_parser_error_on_one_service_logs_and_continues(
406
+
self, caplog
407
+
):
408
+
bus, _ = _make_name_has_owner_bus(
409
+
side_effect=[True, InvalidMemberNameError("bad"), True, True, True]
410
+
)
411
+
412
+
with caplog.at_level(logging.INFO):
413
+
results = await activity.probe_activity_services(bus)
414
+
415
+
assert results == {
416
+
"fdo_screensaver": True,
417
+
"gnome_screensaver": False,
418
+
"gnome_display_config": True,
419
+
"kde_power": True,
420
+
"kscreen": True,
421
+
"gtk4": activity._HAS_GTK,
422
+
}
423
+
messages = [record.message for record in caplog.records]
424
+
assert (
425
+
"NameHasOwner probe failed: service=org.gnome.ScreenSaver "
426
+
"path=/org/freedesktop/DBus: "
427
+
"InvalidMemberNameError: invalid member name: bad"
428
+
) in messages
429
+
assert any(message.startswith("Screen lock backends:") for message in messages)
430
+
assert any(message.startswith("Power save backends:") for message in messages)
431
+
assert any(message.startswith("Monitor backends:") for message in messages)
432
+
247
433
248
434
class TestGetMonitorGeometriesKscreen:
249
435
"""Test KDE KScreen monitor geometry detection."""
···
325
511
@pytest.mark.asyncio
326
512
async def test_returns_empty_on_dbus_failure(self):
327
513
bus = MagicMock()
328
-
bus.introspect = AsyncMock(side_effect=Exception("missing"))
514
+
bus.introspect = AsyncMock(side_effect=_service_unknown("missing"))
329
515
330
516
result = await activity.get_monitor_geometries_kscreen(bus)
331
517
···
357
543
assert result == [
358
544
{"id": "DP-1", "box": [0, 0, 1920, 1080], "position": "center"}
359
545
]
546
+
547
+
@pytest.mark.asyncio
548
+
async def test_get_monitor_geometries_kscreen_dbus_error_logs_and_returns_empty(
549
+
self, caplog
550
+
):
551
+
bus = MagicMock()
552
+
bus.introspect = AsyncMock(side_effect=_no_reply("broke"))
553
+
554
+
with caplog.at_level(logging.WARNING):
555
+
result = await activity.get_monitor_geometries_kscreen(bus)
556
+
557
+
assert result == []
558
+
assert [record.message for record in caplog.records] == [
559
+
"get_monitor_geometries_kscreen failed: "
560
+
"service=org.kde.KScreen path=/backend: DBusError: broke"
561
+
]
562
+
563
+
@pytest.mark.parametrize(
564
+
"error_name",
565
+
[
566
+
"org.freedesktop.DBus.Error.ServiceUnknown",
567
+
"org.freedesktop.DBus.Error.NameHasNoOwner",
568
+
],
569
+
)
570
+
@pytest.mark.asyncio
571
+
async def test_get_monitor_geometries_kscreen_service_missing_does_not_log(
572
+
self, caplog, error_name
573
+
):
574
+
bus = MagicMock()
575
+
bus.introspect = AsyncMock(side_effect=DBusError(error_name, "missing"))
576
+
577
+
with caplog.at_level(logging.WARNING):
578
+
result = await activity.get_monitor_geometries_kscreen(bus)
579
+
580
+
assert result == []
581
+
assert caplog.records == []
+34
-1
tests/test_screencast.py
+34
-1
tests/test_screencast.py
···
3
3
4
4
"""Tests for portal screencast stream matching."""
5
5
6
-
from solstone_linux.screencast import _match_streams_to_monitors
6
+
import logging
7
+
from pathlib import Path
8
+
from unittest.mock import AsyncMock, MagicMock
9
+
10
+
import pytest
11
+
from dbus_next.errors import DBusError
12
+
13
+
from solstone_linux.screencast import Screencaster, _match_streams_to_monitors
7
14
8
15
9
16
class TestMatchStreamsToMonitors:
···
168
175
assert result[0]["position_label"] == "left"
169
176
assert result[1]["connector"] == "DP-2"
170
177
assert result[1]["position_label"] == "right"
178
+
179
+
180
+
@pytest.mark.asyncio
181
+
async def test_close_session_call_close_failure_logs_and_clears_handle(caplog):
182
+
screencaster = Screencaster(restore_token_path=Path("/tmp/fake"))
183
+
mock_bus = MagicMock()
184
+
session_iface = MagicMock()
185
+
session_iface.call_close = AsyncMock(
186
+
side_effect=DBusError("org.freedesktop.DBus.Error.NoReply", "broke")
187
+
)
188
+
189
+
mock_bus.introspect = AsyncMock(return_value=object())
190
+
mock_bus.get_proxy_object.return_value.get_interface.return_value = session_iface
191
+
screencaster.bus = mock_bus
192
+
screencaster.session_handle = "/org/freedesktop/portal/desktop/session/fake"
193
+
194
+
with caplog.at_level(logging.WARNING):
195
+
await screencaster._close_session()
196
+
197
+
assert [record.message for record in caplog.records] == [
198
+
"_close_session failed: "
199
+
"service=org.freedesktop.portal.Desktop "
200
+
"path=/org/freedesktop/portal/desktop/session/fake: "
201
+
"DBusError: broke"
202
+
]
203
+
assert screencaster.session_handle is None