Skip to content

PR-03: DBus IPC (Service, Client, Signals) #2

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Coldaine merged 1 commit into from
Nov 7, 2025
Merged

PR-03: DBus IPC (Service, Client, Signals) #2

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
96 changes: 96 additions & 0 deletions sage/dbus_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,96 @@
"""DBus client for testing and interacting with Shortcut Sage daemon."""

import json
import logging
from collections.abc import Callable
from typing import Any

logger = logging.getLogger(__name__)

# Try to import DBus, but allow fallback if not available
try:
import dbus

DBUS_AVAILABLE = True
except ImportError:
DBUS_AVAILABLE = False
logger.warning("DBus not available, client will not work")


class DBusClient:
"""Client for interacting with Shortcut Sage DBus daemon."""

BUS_NAME = "org.shortcutsage.Daemon"
OBJECT_PATH = "/org/shortcutsage/Daemon"
INTERFACE = "org.shortcutsage.Daemon"

def __init__(self) -> None:
"""Initialize the DBus client."""
if not DBUS_AVAILABLE:
raise ImportError("DBus not available")

self.bus = dbus.SessionBus()
self.proxy = self.bus.get_object(self.BUS_NAME, self.OBJECT_PATH)
self.interface = dbus.Interface(self.proxy, dbus_interface=self.INTERFACE)

def send_event(self, event_json: str | dict[str, Any]) -> None:
"""Send an event to the daemon.

Args:
event_json: Event as JSON string or dict. If dict, will be serialized.

Raises:
dbus.DBusException: If the daemon is not running or the call fails.
Comment on lines +42 to +43
Copy link

Copilot AI Nov 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The docstring states the method raises dbus.DBusException, but based on the error handling in other parts of the codebase and common dbus-python behavior, it could also raise dbus.exceptions.DBusException or other DBus-related exceptions. Consider being more general in the documentation, e.g., "Raises: DBus-related exceptions if the daemon is not running or the call fails."

Copilot uses AI. Check for mistakes.
"""
if isinstance(event_json, dict):
event_json = json.dumps(event_json)

self.interface.SendEvent(event_json)
logger.debug(f"Sent event: {event_json}")
Copy link

Copilot AI Nov 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The logger configuration uses an f-string for the log message which will be evaluated even when the debug level is not enabled. For performance, consider using lazy logging with arguments: logger.debug("Sent event: %s", event_json) instead of logger.debug(f"Sent event: {event_json}"). This avoids string formatting when debug logging is disabled.

Suggested change
logger.debug(f"Sent event: {event_json}")
logger.debug("Sent event: %s", event_json)

Copilot uses AI. Check for mistakes.

def ping(self) -> str:
"""Ping the daemon to check if it's alive.

Returns:
"pong" if the daemon is alive.

Raises:
dbus.DBusException: If the daemon is not running.
"""
result = self.interface.Ping()
logger.debug(f"Ping result: {result}")
Copy link

Copilot AI Nov 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same issue as above - the f-string in the debug log will be evaluated even when debug logging is disabled. Use lazy logging: logger.debug("Ping result: %s", result).

Suggested change
logger.debug(f"Ping result: {result}")
logger.debug("Ping result: %s", result)

Copilot uses AI. Check for mistakes.
return result

def subscribe_suggestions(self, callback: Callable[[str], None]) -> None:
"""Subscribe to the Suggestions signal.

Args:
callback: Function to call when suggestions are received.
Takes a JSON string of suggestions.
"""

def signal_handler(suggestions_json: str) -> None:
logger.debug(f"Received suggestions: {suggestions_json}")
Copy link

Copilot AI Nov 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same issue - use lazy logging: logger.debug("Received suggestions: %s", suggestions_json).

Suggested change
logger.debug(f"Received suggestions: {suggestions_json}")
logger.debug("Received suggestions: %s", suggestions_json)

Copilot uses AI. Check for mistakes.
callback(suggestions_json)

self.bus.add_signal_receiver(
signal_handler,
dbus_interface=self.INTERFACE,
signal_name="Suggestions",
)

@staticmethod
def is_daemon_running() -> bool:
"""Check if the daemon is running.

Returns:
True if the daemon is running, False otherwise.
"""
if not DBUS_AVAILABLE:
return False

try:
bus = dbus.SessionBus()
return bus.name_has_owner(DBusClient.BUS_NAME)
except dbus.DBusException:
return False
252 changes: 252 additions & 0 deletions tests/integration/test_dbus.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,252 @@
"""Integration tests for DBus IPC."""

import json
import time
from datetime import datetime
from pathlib import Path
from typing import Any

import pytest

from sage.dbus_client import DBUS_AVAILABLE

# Skip all tests if DBus is not available
pytestmark = pytest.mark.skipif(not DBUS_AVAILABLE, reason="DBus not available")


if DBUS_AVAILABLE:
import dbus
from dbus.mainloop.glib import DBusGMainLoop
from gi.repository import GLib # type: ignore[import-not-found]

from sage.dbus_client import DBusClient
from sage.dbus_daemon import Daemon


@pytest.fixture
def temp_config_dir(tmp_path: Path) -> Path:
"""Create a temporary config directory with test configuration."""
config_dir = tmp_path / "config"
config_dir.mkdir()

# Create minimal shortcuts.yaml
shortcuts_yaml = config_dir / "shortcuts.yaml"
shortcuts_yaml.write_text(
"""version: "1.0"
shortcuts:
- key: "Meta+Tab"
action: "overview"
description: "Show overview"
category: "desktop"
- key: "Meta+Left"
action: "tile_left"
description: "Tile window left"
category: "window"
"""
)

# Create minimal rules.yaml
rules_yaml = config_dir / "rules.yaml"
rules_yaml.write_text(
"""version: "1.0"
rules:
- name: "after_show_desktop"
context:
type: "event_sequence"
pattern: ["show_desktop"]
window: 3
suggest:
- action: "overview"
priority: 80
cooldown: 300
"""
)

return config_dir


@pytest.fixture
def daemon_process(temp_config_dir: Path) -> Daemon:
"""Start a daemon process for testing."""
# Initialize DBus main loop
DBusGMainLoop(set_as_default=True)

# Create daemon with test configuration
log_dir = temp_config_dir.parent / "logs"
daemon = Daemon(str(temp_config_dir), enable_dbus=True, log_dir=log_dir)
Copy link

Copilot AI Nov 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The Daemon class is initialized with enable_dbus=True, but this only sets up the DBus connection infrastructure - it doesn't actually register the DBus service methods (SendEvent, Ping) or signal (Suggestions). Looking at the dbus_daemon.py implementation, the DBus service methods are only registered when running via main() which creates a separate DBusService wrapper class. Without this wrapper, the daemon won't respond to DBus method calls. The test will fail because the DBus interface won't be available. Consider either: 1) Creating the DBusService wrapper in the fixture, or 2) Refactoring the Daemon class to properly expose DBus methods when enable_dbus=True.

Copilot uses AI. Check for mistakes.

# Start the daemon in a background thread
daemon.start()
Comment on lines +68 to +79
Copy link

@chatgpt-codex-connector chatgpt-codex-connector bot Nov 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P0 Badge Start DBus service before exercising client

daemon_process creates Daemon(..., enable_dbus=True) and calls daemon.start(), but it never instantiates the DBusService or runs a GLib.MainLoop. The DBus methods (SendEvent, Ping, Suggestions) are only registered inside main() in dbus_daemon.py; simply constructing Daemon leaves no object on the bus to answer requests. On systems where DBus is available, every call via DBusClient in these tests will time out or return org.freedesktop.DBus.Error.UnknownMethod, so the suite cannot pass. The fixture should run the daemon through the DBus service (e.g. in a subprocess or with a background GLib loop) before invoking the client.

Useful? React with 👍 / 👎.

Copy link

Copilot AI Nov 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The daemon is started in the main thread, but the Daemon class doesn't actually run in a background thread as the comment suggests. The Daemon.start() method only starts the config watcher and logs a message, but doesn't run a DBus main loop or background processing. This could lead to race conditions or the daemon not being properly initialized for receiving DBus calls. Consider using threading to run the daemon's main loop in the background, or update the comment to accurately reflect the behavior.

Copilot uses AI. Check for mistakes.

# Give daemon time to initialize
time.sleep(0.5)
Comment on lines +81 to +82
Copy link

Copilot AI Nov 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The time.sleep(0.5) is a hard-coded wait time that may be insufficient or excessive depending on the system load. This can lead to flaky tests - too short could cause failures on slower systems, too long wastes time on faster systems. Consider using a more robust approach like polling with a timeout, or implementing a readiness check in the daemon that can be queried.

Suggested change
# Give daemon time to initialize
time.sleep(0.5)
# Wait for daemon to be ready by polling DBus
timeout = 5.0 # seconds
interval = 0.05 # seconds
start_time = time.time()
while True:
try:
client = DBusClient()
if client.ping() == "pong":
break
except Exception:
pass
if time.time() - start_time > timeout:
raise RuntimeError("Timed out waiting for daemon to be ready")
time.sleep(interval)

Copilot uses AI. Check for mistakes.

yield daemon

# Cleanup
daemon.stop()


@pytest.fixture
def dbus_client() -> DBusClient:
"""Create a DBus client for testing."""
return DBusClient()


def test_ping(daemon_process: Daemon, dbus_client: DBusClient) -> None:
"""Test the Ping method."""
result = dbus_client.ping()
assert result == "pong"


def test_send_event_valid_json(daemon_process: Daemon, dbus_client: DBusClient) -> None:
"""Test SendEvent with valid JSON."""
event_data = {
"timestamp": datetime.now().isoformat(),
"type": "window_focus",
"action": "show_desktop",
"metadata": {},
}

# Should not raise an exception
dbus_client.send_event(event_data)

# Give daemon time to process
time.sleep(0.1)
Copy link

Copilot AI Nov 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Similar to the previous test, the hard-coded time.sleep(0.1) wait times make the tests potentially flaky. Consider using polling with assertions or implementing a proper synchronization mechanism to verify that events have been processed.

Copilot uses AI. Check for mistakes.

# Verify event was added to buffer
assert len(daemon_process.buffer.events) == 1
assert daemon_process.buffer.events[0].action == "show_desktop"


def test_send_event_valid_json_string(daemon_process: Daemon, dbus_client: DBusClient) -> None:
"""Test SendEvent with valid JSON string."""
event_json = json.dumps(
{
"timestamp": datetime.now().isoformat(),
"type": "window_focus",
"action": "tile_left",
"metadata": {},
}
)

# Should not raise an exception
dbus_client.send_event(event_json)

# Give daemon time to process
time.sleep(0.1)

# Verify event was added to buffer
assert len(daemon_process.buffer.events) == 1
assert daemon_process.buffer.events[0].action == "tile_left"


def test_send_event_malformed_json(daemon_process: Daemon, dbus_client: DBusClient) -> None:
"""Test SendEvent with malformed JSON."""
# Send malformed JSON - should not crash but should log error
dbus_client.send_event("{invalid json}")

# Give daemon time to process
time.sleep(0.1)

# Buffer should be empty (event was rejected)
assert len(daemon_process.buffer.events) == 0


def test_send_event_missing_fields(daemon_process: Daemon, dbus_client: DBusClient) -> None:
"""Test SendEvent with missing required fields."""
# Missing 'action' field
event_data = {
"timestamp": datetime.now().isoformat(),
"type": "window_focus",
"metadata": {},
}

# Should not crash but should handle error gracefully
dbus_client.send_event(event_data)

# Give daemon time to process
time.sleep(0.1)

# Buffer should be empty (event was rejected)
assert len(daemon_process.buffer.events) == 0


def test_suggestions_signal(daemon_process: Daemon, dbus_client: DBusClient) -> None:
"""Test Suggestions signal emission."""
received_suggestions: list[Any] = []

def callback(suggestions_json: str) -> None:
suggestions = json.loads(suggestions_json)
received_suggestions.extend(suggestions)

# Subscribe to suggestions signal
dbus_client.subscribe_suggestions(callback)

# Send an event that should trigger suggestions
event_data = {
"timestamp": datetime.now().isoformat(),
"type": "window_focus",
"action": "show_desktop",
"metadata": {},
}

dbus_client.send_event(event_data)

# Process pending DBus messages
context = GLib.MainContext.default()
for _ in range(10): # Try up to 10 iterations
context.iteration(False)
time.sleep(0.05)

# Should have received suggestion for "overview" after "show_desktop"
assert len(received_suggestions) > 0
assert any(s["action"] == "overview" for s in received_suggestions)
Comment on lines +175 to +204
Copy link

@chatgpt-codex-connector chatgpt-codex-connector bot Nov 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Suggestions signal asserted although daemon never emits it

test_suggestions_signal subscribes to Suggestions and asserts a non‑empty received_suggestions, but Daemon.emit_suggestions() currently only returns the JSON string and does not emit any DBus signal when DBus is enabled. Even if a bus service is started, the callback will never fire and this assertion will always fail; external clients also cannot receive suggestions. The daemon needs to emit the Suggestions signal (e.g. by calling the DBus signal defined in dbus_daemon.py) before this test or client code can work.

Useful? React with 👍 / 👎.

Comment on lines +175 to +204
Copy link

Copilot AI Nov 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The test expects the daemon to emit suggestions via DBus signal, but there's no mechanism to actually trigger the signal emission. Looking at the Daemon code, emit_suggestions() returns the JSON string but doesn't actually emit a DBus signal when DBus is enabled. The Daemon class defines emit_suggestions() but doesn't call the DBus signal method. This test will likely fail because no signal will be received.

Copilot uses AI. Check for mistakes.


def test_multiple_events_sequence(daemon_process: Daemon, dbus_client: DBusClient) -> None:
"""Test sending multiple events in sequence."""
events = [
{
"timestamp": datetime.now().isoformat(),
"type": "window_focus",
"action": "show_desktop",
"metadata": {},
},
{
"timestamp": datetime.now().isoformat(),
"type": "window_focus",
"action": "tile_left",
"metadata": {},
},
{
"timestamp": datetime.now().isoformat(),
"type": "window_focus",
"action": "tile_right",
"metadata": {},
},
]

for event in events:
dbus_client.send_event(event)
time.sleep(0.05)

# Verify all events were processed
assert len(daemon_process.buffer.events) == 3


def test_daemon_is_running() -> None:
"""Test checking if daemon is running."""
# This test doesn't need a daemon fixture
# Just test the utility method
# Note: May be False if no daemon is actually running
result = DBusClient.is_daemon_running()
assert isinstance(result, bool)


def test_dbus_error_handling(temp_config_dir: Path) -> None:
"""Test error handling when daemon is not running."""
# Don't start daemon
with pytest.raises((dbus.DBusException, dbus.exceptions.DBusException)): # type: ignore[attr-defined]
Copy link

Copilot AI Nov 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The exception type check uses dbus.exceptions.DBusException which may not exist depending on the dbus-python version. The dbus library typically uses dbus.DBusException directly. The dbus.exceptions.DBusException reference appears to be redundant and may cause issues if that submodule doesn't exist. Consider using only dbus.DBusException in the exception tuple.

Suggested change
with pytest.raises((dbus.DBusException, dbus.exceptions.DBusException)): # type: ignore[attr-defined]
with pytest.raises(dbus.DBusException):

Copilot uses AI. Check for mistakes.
client = DBusClient()
client.ping()
Copy link

Copilot AI Nov 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The DBusClient constructor attempts to get a DBus object that may not exist yet, which will raise an exception. However, the test test_dbus_error_handling expects the exception to be raised when calling client.ping(), not during DBusClient() construction. This means the test logic is incorrect - the exception will be raised on line 251 (during client creation) not line 252 (during ping). Consider moving the DBusClient instantiation inside the with pytest.raises() block.

Suggested change
client.ping()

Copilot uses AI. Check for mistakes.
Loading