+2 -2

AGENTS.md

reviewed

··· 20 20 audio_recorder.py Stereo audio recording (mic + system via soundcard) 21 21 audio_detect.py Audio device detection via ultrasonic tone 22 22 audio_mute.py PulseAudio mute state detection 23 23 - activity.py GNOME-specific activity detection (idle, screen lock, power save) 23 23 + activity.py Cross-desktop activity detection (screen lock, power save) via DBus 24 24 monitor_positions.py Monitor position assignment from geometry 25 25 session_env.py Desktop session environment checks and recovery 26 26 streams.py Stream name derivation (hostname-based) ··· 101 101 102 102 ## Key Patterns 103 103 104 104 - - **Activity detection is GNOME-specific.** Uses Mutter IdleMonitor, GNOME ScreenSaver, and Mutter DisplayConfig DBus interfaces. Other desktops capture screen and audio but won't get activity-based segment boundaries. 104 104 + - **Activity detection is cross-desktop.** Uses ordered DBus fallback chains for screen lock (freedesktop.org ScreenSaver → GNOME ScreenSaver) and power save (Mutter DisplayConfig → KDE Solid PowerManagement). All backends degrade gracefully to safe defaults. 105 105 - **Audio is stereo-interleaved.** Left channel = microphone, right channel = system audio. When muted, channels are split into separate mono FLAC files. 106 106 - **Screencast uses xdg-desktop-portal.** Session persistence via restore tokens avoids re-prompting the user. GStreamer subprocess (`gst-launch-1.0`) handles the actual PipeWire recording. 107 107 - **Crash recovery runs on startup.** `recovery.py` scans for orphaned `.incomplete` directories older than 2 minutes and finalizes or marks them as failed.

+25 -21

src/solstone_linux/activity.py

reviewed

··· 3 3 4 4 """Activity detection using DBus APIs. 5 5 6 6 - Detects screen lock and power save state across GNOME and KDE desktops 7 7 - using freedesktop, GNOME, and KDE DBus interfaces with ordered fallback 8 8 - chains. Every function degrades gracefully — returning a safe default — 9 9 - so the observer keeps running regardless of desktop environment. 6 6 + Detects screen lock and display power-save state via DBus, with ordered 7 7 + fallback chains that cover GNOME and KDE desktops. Every function 8 8 + degrades gracefully — returning a safe default — so the observer keeps 9 9 + running regardless of desktop environment. 10 10 """ 11 11 12 12 import logging ··· 30 30 except (ImportError, ValueError): 31 31 _HAS_GTK = False 32 32 33 33 - # DBus service constants 33 33 + # DBus service constants — screen lock 34 34 FDO_SCREENSAVER_BUS = "org.freedesktop.ScreenSaver" 35 35 FDO_SCREENSAVER_PATH = "/ScreenSaver" 36 36 FDO_SCREENSAVER_IFACE = "org.freedesktop.ScreenSaver" ··· 39 39 GNOME_SCREENSAVER_PATH = "/org/gnome/ScreenSaver" 40 40 GNOME_SCREENSAVER_IFACE = "org.gnome.ScreenSaver" 41 41 42 42 + # DBus service constants — power save 42 43 DISPLAY_CONFIG_BUS = "org.gnome.Mutter.DisplayConfig" 43 44 DISPLAY_CONFIG_PATH = "/org/gnome/Mutter/DisplayConfig" 44 45 DISPLAY_CONFIG_IFACE = "org.gnome.Mutter.DisplayConfig" ··· 64 65 except Exception: 65 66 results[name] = False 66 67 68 68 + # Log grouped by function 67 69 lock_backends = ["fdo_screensaver", "gnome_screensaver"] 68 70 power_backends = ["gnome_display_config", "kde_power"] 69 71 70 70 - lock_available = [name for name in lock_backends if results.get(name)] 71 71 - power_available = [name for name in power_backends if results.get(name)] 72 72 + def _status(keys): 73 73 + return ", ".join(f"{k} [{'ok' if results[k] else 'missing'}]" for k in keys) 72 74 73 73 - if lock_available: 74 74 - logger.info("Screen lock backends: %s", ", ".join(lock_available)) 75 75 - else: 76 76 - logger.warning("No screen lock backends available — will assume unlocked") 75 75 + logger.info("Screen lock backends: %s", _status(lock_backends)) 76 76 + logger.info("Power save backends: %s", _status(power_backends)) 77 77 78 78 - if power_available: 79 79 - logger.info("Power save backends: %s", ", ".join(power_available)) 80 80 - else: 81 81 - logger.warning("No power save backends available — will assume active display") 78 78 + any_lock = any(results[k] for k in lock_backends) 79 79 + any_power = any(results[k] for k in power_backends) 80 80 + if not any_lock and not any_power: 81 81 + logger.warning( 82 82 + "No activity backends available — running in always-capture mode" 83 83 + ) 82 84 83 85 results["gtk4"] = _HAS_GTK 84 86 if not _HAS_GTK: ··· 88 90 89 91 90 92 async def is_screen_locked(bus: MessageBus) -> bool: 91 91 - """Check if the screen is currently locked. 93 93 + """Check if the screen is locked via FDO ScreenSaver, then GNOME ScreenSaver. 92 94 93 93 - Tries freedesktop.ScreenSaver first (works on KDE, GNOME, and others), 94 94 - then falls back to GNOME ScreenSaver. 95 95 + Returns True if locked, False if unlocked or all backends unavailable. 95 96 """ 97 97 + # Try freedesktop.org ScreenSaver first (KDE kwin + GNOME) 96 98 try: 97 99 intro = await bus.introspect(FDO_SCREENSAVER_BUS, FDO_SCREENSAVER_PATH) 98 100 obj = bus.get_proxy_object(FDO_SCREENSAVER_BUS, FDO_SCREENSAVER_PATH, intro) ··· 101 103 except Exception: 102 104 pass 103 105 106 106 + # Fall back to GNOME ScreenSaver 104 107 try: 105 108 intro = await bus.introspect(GNOME_SCREENSAVER_BUS, GNOME_SCREENSAVER_PATH) 106 109 obj = bus.get_proxy_object(GNOME_SCREENSAVER_BUS, GNOME_SCREENSAVER_PATH, intro) ··· 111 114 112 115 113 116 async def is_power_save_active(bus: MessageBus) -> bool: 114 114 - """Check if display power save mode is active. 117 117 + """Check display power save via GNOME Mutter, then KDE Solid. 115 118 116 116 - Tries GNOME Mutter DisplayConfig first (DPMS state), then falls back 117 117 - to KDE Solid PowerManagement lid state. 119 119 + Returns True if power save is active, False otherwise. 118 120 """ 121 121 + # Try GNOME Mutter DisplayConfig first 119 122 try: 120 123 intro = await bus.introspect(DISPLAY_CONFIG_BUS, DISPLAY_CONFIG_PATH) 121 124 obj = bus.get_proxy_object(DISPLAY_CONFIG_BUS, DISPLAY_CONFIG_PATH, intro) ··· 126 129 except Exception: 127 130 pass 128 131 132 132 + # Fall back to KDE Solid PowerManagement 129 133 try: 130 134 intro = await bus.introspect(KDE_POWER_BUS, KDE_POWER_PATH) 131 135 obj = bus.get_proxy_object(KDE_POWER_BUS, KDE_POWER_PATH, intro)

+180 -166

tests/test_activity.py

reviewed

··· 1 1 # SPDX-License-Identifier: AGPL-3.0-only 2 2 # Copyright (c) 2026 sol pbc 3 3 4 4 - """Tests for desktop activity detection fallbacks.""" 4 4 + """Tests for cross-desktop activity detection backends.""" 5 5 6 6 - from types import SimpleNamespace 7 7 - from unittest.mock import AsyncMock, MagicMock 6 6 + import logging 7 7 + from unittest.mock import AsyncMock, MagicMock, call 8 8 9 9 import pytest 10 10 11 11 from solstone_linux import activity 12 12 13 13 14 14 - def _make_proxy(interface_name: str, iface: object) -> MagicMock: 14 14 + def _make_proxy_with_interface(interface: MagicMock) -> MagicMock: 15 15 proxy = MagicMock() 16 16 - proxy.get_interface.side_effect = lambda name: ( 17 17 - iface if name == interface_name else None 18 18 - ) 16 16 + proxy.get_interface.return_value = interface 19 17 return proxy 20 18 21 19 22 22 - @pytest.mark.asyncio 23 23 - async def test_is_screen_locked_prefers_fdo_true(): 24 24 - bus = MagicMock() 25 25 - fdo_iface = AsyncMock() 26 26 - fdo_iface.call_get_active.return_value = True 27 27 - bus.introspect = AsyncMock(return_value="fdo-intro") 28 28 - bus.get_proxy_object.return_value = _make_proxy( 29 29 - activity.FDO_SCREENSAVER_IFACE, fdo_iface 30 30 - ) 20 20 + def _make_variant(value: int) -> MagicMock: 21 21 + variant = MagicMock() 22 22 + variant.value = value 23 23 + return variant 31 24 32 32 - result = await activity.is_screen_locked(bus) 33 25 34 34 - assert result is True 35 35 - bus.introspect.assert_awaited_once_with( 36 36 - activity.FDO_SCREENSAVER_BUS, activity.FDO_SCREENSAVER_PATH 37 37 - ) 38 38 - fdo_iface.call_get_active.assert_awaited_once_with() 26 26 + class TestIsScreenLocked: 27 27 + """Test screen lock fallback order.""" 39 28 29 29 + @pytest.mark.asyncio 30 30 + async def test_fdo_backend_returns_true_without_gnome_fallback(self): 31 31 + bus = MagicMock() 32 32 + bus.introspect = AsyncMock(return_value=object()) 33 33 + iface = MagicMock() 34 34 + iface.call_get_active = AsyncMock(return_value=True) 35 35 + bus.get_proxy_object.return_value = _make_proxy_with_interface(iface) 40 36 41 41 - @pytest.mark.asyncio 42 42 - async def test_is_screen_locked_prefers_fdo_false(): 43 43 - bus = MagicMock() 44 44 - fdo_iface = AsyncMock() 45 45 - fdo_iface.call_get_active.return_value = False 46 46 - bus.introspect = AsyncMock(return_value="fdo-intro") 47 47 - bus.get_proxy_object.return_value = _make_proxy( 48 48 - activity.FDO_SCREENSAVER_IFACE, fdo_iface 49 49 - ) 37 37 + result = await activity.is_screen_locked(bus) 50 38 51 51 - result = await activity.is_screen_locked(bus) 39 39 + assert result is True 40 40 + assert bus.introspect.await_count == 1 41 41 + bus.introspect.assert_awaited_once_with( 42 42 + activity.FDO_SCREENSAVER_BUS, activity.FDO_SCREENSAVER_PATH 43 43 + ) 52 44 53 53 - assert result is False 54 54 - bus.introspect.assert_awaited_once_with( 55 55 - activity.FDO_SCREENSAVER_BUS, activity.FDO_SCREENSAVER_PATH 56 56 - ) 57 57 - fdo_iface.call_get_active.assert_awaited_once_with() 45 45 + @pytest.mark.asyncio 46 46 + async def test_fdo_backend_returns_false_without_gnome_fallback(self): 47 47 + bus = MagicMock() 48 48 + bus.introspect = AsyncMock(return_value=object()) 49 49 + iface = MagicMock() 50 50 + iface.call_get_active = AsyncMock(return_value=False) 51 51 + bus.get_proxy_object.return_value = _make_proxy_with_interface(iface) 58 52 53 53 + result = await activity.is_screen_locked(bus) 59 54 60 60 - @pytest.mark.asyncio 61 61 - async def test_is_screen_locked_falls_back_to_gnome(): 62 62 - bus = MagicMock() 63 63 - gnome_iface = AsyncMock() 64 64 - gnome_iface.call_get_active.return_value = True 65 65 - bus.introspect = AsyncMock( 66 66 - side_effect=[Exception("fdo unavailable"), "gnome-intro"] 67 67 - ) 68 68 - 69 69 - def get_proxy_object(bus_name: str, path: str, intro: str) -> MagicMock: 70 70 - assert (bus_name, path, intro) == ( 71 71 - activity.GNOME_SCREENSAVER_BUS, 72 72 - activity.GNOME_SCREENSAVER_PATH, 73 73 - "gnome-intro", 55 55 + assert result is False 56 56 + assert bus.introspect.await_count == 1 57 57 + bus.introspect.assert_awaited_once_with( 58 58 + activity.FDO_SCREENSAVER_BUS, activity.FDO_SCREENSAVER_PATH 74 59 ) 75 75 - return _make_proxy(activity.GNOME_SCREENSAVER_IFACE, gnome_iface) 76 60 77 77 - bus.get_proxy_object.side_effect = get_proxy_object 61 61 + @pytest.mark.asyncio 62 62 + async def test_fdo_failure_gnome_returns_true(self): 63 63 + bus = MagicMock() 64 64 + bus.introspect = AsyncMock(side_effect=[Exception("fdo unavailable"), object()]) 65 65 + gnome_iface = MagicMock() 66 66 + gnome_iface.call_get_active = AsyncMock(return_value=True) 67 67 + bus.get_proxy_object.return_value = _make_proxy_with_interface(gnome_iface) 78 68 79 79 - result = await activity.is_screen_locked(bus) 69 69 + result = await activity.is_screen_locked(bus) 80 70 81 81 - assert result is True 82 82 - assert bus.introspect.await_args_list == [ 83 83 - ((activity.FDO_SCREENSAVER_BUS, activity.FDO_SCREENSAVER_PATH),), 84 84 - ((activity.GNOME_SCREENSAVER_BUS, activity.GNOME_SCREENSAVER_PATH),), 85 85 - ] 86 86 - gnome_iface.call_get_active.assert_awaited_once_with() 71 71 + assert result is True 72 72 + assert bus.introspect.await_args_list == [ 73 73 + call(activity.FDO_SCREENSAVER_BUS, activity.FDO_SCREENSAVER_PATH), 74 74 + call(activity.GNOME_SCREENSAVER_BUS, activity.GNOME_SCREENSAVER_PATH), 75 75 + ] 87 76 77 77 + @pytest.mark.asyncio 78 78 + async def test_fdo_failure_gnome_returns_false(self): 79 79 + bus = MagicMock() 80 80 + bus.introspect = AsyncMock(side_effect=[Exception("fdo unavailable"), object()]) 81 81 + gnome_iface = MagicMock() 82 82 + gnome_iface.call_get_active = AsyncMock(return_value=False) 83 83 + bus.get_proxy_object.return_value = _make_proxy_with_interface(gnome_iface) 88 84 89 89 - @pytest.mark.asyncio 90 90 - async def test_is_screen_locked_returns_false_when_all_backends_fail(): 91 91 - bus = MagicMock() 92 92 - bus.introspect = AsyncMock(side_effect=Exception("unavailable")) 85 85 + result = await activity.is_screen_locked(bus) 93 86 94 94 - result = await activity.is_screen_locked(bus) 95 95 - 96 96 - assert result is False 97 97 - assert bus.introspect.await_count == 2 87 87 + assert result is False 88 88 + assert bus.introspect.await_args_list == [ 89 89 + call(activity.FDO_SCREENSAVER_BUS, activity.FDO_SCREENSAVER_PATH), 90 90 + call(activity.GNOME_SCREENSAVER_BUS, activity.GNOME_SCREENSAVER_PATH), 91 91 + ] 98 92 93 93 + @pytest.mark.asyncio 94 94 + async def test_both_backends_fail_returns_false(self): 95 95 + bus = MagicMock() 96 96 + bus.introspect = AsyncMock( 97 97 + side_effect=[Exception("fdo unavailable"), Exception("gnome unavailable")] 98 98 + ) 99 99 100 100 - @pytest.mark.asyncio 101 101 - async def test_is_power_save_active_uses_gnome_display_config_when_nonzero(): 102 102 - bus = MagicMock() 103 103 - props_iface = AsyncMock() 104 104 - props_iface.call_get.return_value = SimpleNamespace(value=2) 105 105 - bus.introspect = AsyncMock(return_value="display-intro") 106 106 - bus.get_proxy_object.return_value = _make_proxy( 107 107 - "org.freedesktop.DBus.Properties", props_iface 108 108 - ) 100 100 + result = await activity.is_screen_locked(bus) 109 101 110 110 - result = await activity.is_power_save_active(bus) 102 102 + assert result is False 103 103 + assert bus.introspect.await_args_list == [ 104 104 + call(activity.FDO_SCREENSAVER_BUS, activity.FDO_SCREENSAVER_PATH), 105 105 + call(activity.GNOME_SCREENSAVER_BUS, activity.GNOME_SCREENSAVER_PATH), 106 106 + ] 111 107 112 112 - assert result is True 113 113 - bus.introspect.assert_awaited_once_with( 114 114 - activity.DISPLAY_CONFIG_BUS, activity.DISPLAY_CONFIG_PATH 115 115 - ) 116 116 - props_iface.call_get.assert_awaited_once_with( 117 117 - activity.DISPLAY_CONFIG_IFACE, "PowerSaveMode" 118 118 - ) 119 108 109 109 + class TestIsPowerSaveActive: 110 110 + """Test power save fallback order.""" 120 111 121 121 - @pytest.mark.asyncio 122 122 - async def test_is_power_save_active_uses_gnome_display_config_when_zero(): 123 123 - bus = MagicMock() 124 124 - props_iface = AsyncMock() 125 125 - props_iface.call_get.return_value = SimpleNamespace(value=0) 126 126 - bus.introspect = AsyncMock(return_value="display-intro") 127 127 - bus.get_proxy_object.return_value = _make_proxy( 128 128 - "org.freedesktop.DBus.Properties", props_iface 129 129 - ) 112 112 + @pytest.mark.asyncio 113 113 + async def test_gnome_backend_nonzero_mode_returns_true(self): 114 114 + bus = MagicMock() 115 115 + bus.introspect = AsyncMock(return_value=object()) 116 116 + iface = MagicMock() 117 117 + iface.call_get = AsyncMock(return_value=_make_variant(2)) 118 118 + bus.get_proxy_object.return_value = _make_proxy_with_interface(iface) 130 119 131 131 - result = await activity.is_power_save_active(bus) 120 120 + result = await activity.is_power_save_active(bus) 132 121 133 133 - assert result is False 134 134 - bus.introspect.assert_awaited_once_with( 135 135 - activity.DISPLAY_CONFIG_BUS, activity.DISPLAY_CONFIG_PATH 136 136 - ) 137 137 - props_iface.call_get.assert_awaited_once_with( 138 138 - activity.DISPLAY_CONFIG_IFACE, "PowerSaveMode" 139 139 - ) 122 122 + assert result is True 123 123 + bus.introspect.assert_awaited_once_with( 124 124 + activity.DISPLAY_CONFIG_BUS, activity.DISPLAY_CONFIG_PATH 125 125 + ) 140 126 127 127 + @pytest.mark.asyncio 128 128 + async def test_gnome_backend_zero_mode_returns_false(self): 129 129 + bus = MagicMock() 130 130 + bus.introspect = AsyncMock(return_value=object()) 131 131 + iface = MagicMock() 132 132 + iface.call_get = AsyncMock(return_value=_make_variant(0)) 133 133 + bus.get_proxy_object.return_value = _make_proxy_with_interface(iface) 141 134 142 142 - @pytest.mark.asyncio 143 143 - async def test_is_power_save_active_falls_back_to_kde(): 144 144 - bus = MagicMock() 145 145 - kde_iface = AsyncMock() 146 146 - kde_iface.call_is_lid_closed.return_value = True 147 147 - bus.introspect = AsyncMock( 148 148 - side_effect=[Exception("gnome unavailable"), "kde-intro"] 149 149 - ) 135 135 + result = await activity.is_power_save_active(bus) 150 136 151 151 - def get_proxy_object(bus_name: str, path: str, intro: str) -> MagicMock: 152 152 - assert (bus_name, path, intro) == ( 153 153 - activity.KDE_POWER_BUS, 154 154 - activity.KDE_POWER_PATH, 155 155 - "kde-intro", 137 137 + assert result is False 138 138 + bus.introspect.assert_awaited_once_with( 139 139 + activity.DISPLAY_CONFIG_BUS, activity.DISPLAY_CONFIG_PATH 156 140 ) 157 157 - return _make_proxy(activity.KDE_POWER_IFACE, kde_iface) 158 141 159 159 - bus.get_proxy_object.side_effect = get_proxy_object 142 142 + @pytest.mark.asyncio 143 143 + async def test_gnome_failure_kde_lid_closed_returns_true(self): 144 144 + bus = MagicMock() 145 145 + bus.introspect = AsyncMock( 146 146 + side_effect=[Exception("gnome unavailable"), object()] 147 147 + ) 148 148 + kde_iface = MagicMock() 149 149 + kde_iface.call_is_lid_closed = AsyncMock(return_value=True) 150 150 + bus.get_proxy_object.return_value = _make_proxy_with_interface(kde_iface) 160 151 161 161 - result = await activity.is_power_save_active(bus) 152 152 + result = await activity.is_power_save_active(bus) 162 153 163 163 - assert result is True 164 164 - assert bus.introspect.await_args_list == [ 165 165 - ((activity.DISPLAY_CONFIG_BUS, activity.DISPLAY_CONFIG_PATH),), 166 166 - ((activity.KDE_POWER_BUS, activity.KDE_POWER_PATH),), 167 167 - ] 168 168 - kde_iface.call_is_lid_closed.assert_awaited_once_with() 154 154 + assert result is True 155 155 + assert bus.introspect.await_args_list == [ 156 156 + call(activity.DISPLAY_CONFIG_BUS, activity.DISPLAY_CONFIG_PATH), 157 157 + call(activity.KDE_POWER_BUS, activity.KDE_POWER_PATH), 158 158 + ] 169 159 160 160 + @pytest.mark.asyncio 161 161 + async def test_gnome_failure_kde_lid_open_returns_false(self): 162 162 + bus = MagicMock() 163 163 + bus.introspect = AsyncMock( 164 164 + side_effect=[Exception("gnome unavailable"), object()] 165 165 + ) 166 166 + kde_iface = MagicMock() 167 167 + kde_iface.call_is_lid_closed = AsyncMock(return_value=False) 168 168 + bus.get_proxy_object.return_value = _make_proxy_with_interface(kde_iface) 170 169 171 171 - @pytest.mark.asyncio 172 172 - async def test_is_power_save_active_returns_false_when_all_backends_fail(): 173 173 - bus = MagicMock() 174 174 - bus.introspect = AsyncMock(side_effect=Exception("unavailable")) 170 170 + result = await activity.is_power_save_active(bus) 175 171 176 176 - result = await activity.is_power_save_active(bus) 172 172 + assert result is False 173 173 + assert bus.introspect.await_args_list == [ 174 174 + call(activity.DISPLAY_CONFIG_BUS, activity.DISPLAY_CONFIG_PATH), 175 175 + call(activity.KDE_POWER_BUS, activity.KDE_POWER_PATH), 176 176 + ] 177 177 178 178 - assert result is False 179 179 - assert bus.introspect.await_count == 2 178 178 + @pytest.mark.asyncio 179 179 + async def test_both_backends_fail_returns_false(self): 180 180 + bus = MagicMock() 181 181 + bus.introspect = AsyncMock( 182 182 + side_effect=[Exception("gnome unavailable"), Exception("kde unavailable")] 183 183 + ) 180 184 185 185 + result = await activity.is_power_save_active(bus) 181 186 182 182 - @pytest.mark.asyncio 183 183 - async def test_probe_activity_services_all_available(): 184 184 - bus = MagicMock() 185 185 - bus.introspect = AsyncMock(return_value="intro") 187 187 + assert result is False 188 188 + assert bus.introspect.await_args_list == [ 189 189 + call(activity.DISPLAY_CONFIG_BUS, activity.DISPLAY_CONFIG_PATH), 190 190 + call(activity.KDE_POWER_BUS, activity.KDE_POWER_PATH), 191 191 + ] 186 192 187 187 - results = await activity.probe_activity_services(bus) 188 193 189 189 - assert results["fdo_screensaver"] is True 190 190 - assert results["gnome_screensaver"] is True 191 191 - assert results["gnome_display_config"] is True 192 192 - assert results["kde_power"] is True 193 193 - assert results["gtk4"] is activity._HAS_GTK 194 194 + class TestProbeActivityServices: 195 195 + """Test activity backend probing and logging.""" 194 196 197 197 + @pytest.mark.asyncio 198 198 + async def test_all_services_available_returns_true_results(self): 199 199 + bus = MagicMock() 200 200 + bus.introspect = AsyncMock(return_value=object()) 195 201 196 196 - @pytest.mark.asyncio 197 197 - async def test_probe_activity_services_all_unavailable(): 198 198 - bus = MagicMock() 199 199 - bus.introspect = AsyncMock(side_effect=Exception("unavailable")) 202 202 + results = await activity.probe_activity_services(bus) 200 203 201 201 - results = await activity.probe_activity_services(bus) 202 202 - 203 203 - assert results["fdo_screensaver"] is False 204 204 - assert results["gnome_screensaver"] is False 205 205 - assert results["gnome_display_config"] is False 206 206 - assert results["kde_power"] is False 207 207 - assert results["gtk4"] is activity._HAS_GTK 204 204 + assert results["fdo_screensaver"] is True 205 205 + assert results["gnome_screensaver"] is True 206 206 + assert results["gnome_display_config"] is True 207 207 + assert results["kde_power"] is True 208 208 + assert results["gtk4"] is activity._HAS_GTK 208 209 210 210 + @pytest.mark.asyncio 211 211 + async def test_no_services_available_logs_warning(self, caplog): 212 212 + bus = MagicMock() 213 213 + bus.introspect = AsyncMock(side_effect=Exception("missing")) 209 214 210 210 - @pytest.mark.asyncio 211 211 - async def test_probe_activity_services_mixed_availability(): 212 212 - bus = MagicMock() 215 215 + with caplog.at_level(logging.WARNING): 216 216 + results = await activity.probe_activity_services(bus) 213 217 214 214 - async def introspect(bus_name: str, path: str) -> str: 215 215 - if bus_name in {activity.FDO_SCREENSAVER_BUS, activity.KDE_POWER_BUS}: 216 216 - return "intro" 217 217 - raise Exception(f"{bus_name} unavailable") 218 218 + assert results["fdo_screensaver"] is False 219 219 + assert results["gnome_screensaver"] is False 220 220 + assert results["gnome_display_config"] is False 221 221 + assert results["kde_power"] is False 222 222 + assert "No activity backends available" in caplog.text 218 223 219 219 - bus.introspect = AsyncMock(side_effect=introspect) 224 224 + @pytest.mark.asyncio 225 225 + async def test_mixed_service_availability_returns_correct_results(self): 226 226 + bus = MagicMock() 227 227 + bus.introspect = AsyncMock( 228 228 + side_effect=[ 229 229 + object(), 230 230 + Exception("missing"), 231 231 + object(), 232 232 + Exception("missing"), 233 233 + ] 234 234 + ) 220 235 221 221 - results = await activity.probe_activity_services(bus) 236 236 + results = await activity.probe_activity_services(bus) 222 237 223 223 - assert results["fdo_screensaver"] is True 224 224 - assert results["gnome_screensaver"] is False 225 225 - assert results["gnome_display_config"] is False 226 226 - assert results["kde_power"] is True 227 227 - assert results["gtk4"] is activity._HAS_GTK 238 238 + assert results["fdo_screensaver"] is True 239 239 + assert results["gnome_screensaver"] is False 240 240 + assert results["gnome_display_config"] is True 241 241 + assert results["kde_power"] is False