This commit is contained in:
+11
-189
@@ -6,26 +6,16 @@ import dbus
|
|||||||
|
|
||||||
from dbusmock import DBusTestCase, MOCK_IFACE
|
from dbusmock import DBusTestCase, MOCK_IFACE
|
||||||
|
|
||||||
import os
|
|
||||||
import time
|
|
||||||
import subprocess
|
|
||||||
import pytest
|
|
||||||
import dbus
|
|
||||||
|
|
||||||
import time
|
|
||||||
import pytest
|
|
||||||
import dbus
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(scope="session")
|
@pytest.fixture(scope="session")
|
||||||
def notifier():
|
def notifier():
|
||||||
"""Importe DBusNotify avec un SessionBus actif (dunst -print)."""
|
# Import DBusNotify with an active SessionBus (dunst -print).
|
||||||
import slic.utils.dbusnotify as dbn
|
import slic.utils.dbusnotify as dbn
|
||||||
return dbn.DBusNotify()
|
return dbn.DBusNotify()
|
||||||
|
|
||||||
|
|
||||||
|
# Test creation of a simple notification
|
||||||
def test_notify_create(notifier):
|
def test_notify_create(notifier):
|
||||||
"""Test la création d'une notification simple."""
|
|
||||||
from slic.utils.dbusnotify import Notification
|
from slic.utils.dbusnotify import Notification
|
||||||
notification = notifier.notify("Test Notification", "This is a test notification.")
|
notification = notifier.notify("Test Notification", "This is a test notification.")
|
||||||
|
|
||||||
@@ -35,8 +25,8 @@ def test_notify_create(notifier):
|
|||||||
assert notification.body == "This is a test notification."
|
assert notification.body == "This is a test notification."
|
||||||
|
|
||||||
|
|
||||||
|
# Test updating an existing notification
|
||||||
def test_notify_update(notifier):
|
def test_notify_update(notifier):
|
||||||
"""Test la mise à jour d'une notification."""
|
|
||||||
notification = notifier.notify("Test Notification", "This is a test notification.")
|
notification = notifier.notify("Test Notification", "This is a test notification.")
|
||||||
notification.update(summary="Updated Test", body="This notification has been updated.")
|
notification.update(summary="Updated Test", body="This notification has been updated.")
|
||||||
|
|
||||||
@@ -44,8 +34,8 @@ def test_notify_update(notifier):
|
|||||||
assert notification.body == "This notification has been updated."
|
assert notification.body == "This notification has been updated."
|
||||||
|
|
||||||
|
|
||||||
|
# Test retrieving DBus server information
|
||||||
def test_get_server_info(notifier):
|
def test_get_server_info(notifier):
|
||||||
"""Test l'obtention des informations du serveur DBus."""
|
|
||||||
server_info = notifier.get_server_info()
|
server_info = notifier.get_server_info()
|
||||||
|
|
||||||
assert isinstance(server_info, dict)
|
assert isinstance(server_info, dict)
|
||||||
@@ -55,14 +45,14 @@ def test_get_server_info(notifier):
|
|||||||
assert "spec" in server_info and server_info["spec"], "Server spec is empty"
|
assert "spec" in server_info and server_info["spec"], "Server spec is empty"
|
||||||
|
|
||||||
|
|
||||||
|
# Test retrieving server capabilities
|
||||||
def test_get_capabilities(notifier):
|
def test_get_capabilities(notifier):
|
||||||
"""Test l'obtention des capacités du serveur DBus."""
|
|
||||||
capabilities = [str(x) for x in list(notifier.get_capabilities())]
|
capabilities = [str(x) for x in list(notifier.get_capabilities())]
|
||||||
|
|
||||||
assert isinstance(capabilities, list), "Capabilities should be a list."
|
assert isinstance(capabilities, list), "Capabilities should be a list."
|
||||||
assert len(capabilities) > 0, "Capabilities list is empty."
|
assert len(capabilities) > 0, "Capabilities list is empty."
|
||||||
|
|
||||||
# on sait que dunst expose typiquement "body", "actions", "icon", "sound"
|
# dunst typically exposes "body", "actions", "icon", "sound"
|
||||||
expected = {"actions", "body", "icon", "sound"}
|
expected = {"actions", "body", "icon", "sound"}
|
||||||
assert expected.intersection(capabilities), "Capabilities don't contain expected values"
|
assert expected.intersection(capabilities), "Capabilities don't contain expected values"
|
||||||
|
|
||||||
@@ -70,24 +60,24 @@ def test_get_capabilities(notifier):
|
|||||||
assert isinstance(cap, str), f"Capability should be str, got {type(cap)}"
|
assert isinstance(cap, str), f"Capability should be str, got {type(cap)}"
|
||||||
|
|
||||||
|
|
||||||
|
# Test sending and closing a notification
|
||||||
def test_notify_and_close(notifier):
|
def test_notify_and_close(notifier):
|
||||||
"""Test envoi + fermeture d'une notification."""
|
|
||||||
notification = notifier.notify("Test Close", "This notification will be closed.")
|
notification = notifier.notify("Test Close", "This notification will be closed.")
|
||||||
time.sleep(0.2) # petite marge
|
time.sleep(0.2) # small margin
|
||||||
try:
|
try:
|
||||||
notification.close()
|
notification.close()
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
pytest.fail(f"close() raised an exception: {e}")
|
pytest.fail(f"close() raised an exception: {e}")
|
||||||
|
|
||||||
|
|
||||||
|
# Test that an invalid body type raises an exception
|
||||||
def test_notify_invalid_value(notifier):
|
def test_notify_invalid_value(notifier):
|
||||||
"""On attend une exception si le body n'est pas une string."""
|
|
||||||
with pytest.raises(Exception):
|
with pytest.raises(Exception):
|
||||||
notifier.notify("Invalid Test", 1234)
|
notifier.notify("Invalid Test", 1234)
|
||||||
|
|
||||||
|
|
||||||
|
# Test conversion of dbus.String
|
||||||
def test_convert_dbus_strings():
|
def test_convert_dbus_strings():
|
||||||
"""Vérifie la conversion dbus.String -> str dans une séquence mixte."""
|
|
||||||
from slic.utils.dbusnotify import convert_dbus_strings
|
from slic.utils.dbusnotify import convert_dbus_strings
|
||||||
test_sequence = [dbus.String('Test String 1'), 42, dbus.String('Test String 2'), 3.14]
|
test_sequence = [dbus.String('Test String 1'), 42, dbus.String('Test String 2'), 3.14]
|
||||||
converted = convert_dbus_strings(test_sequence)
|
converted = convert_dbus_strings(test_sequence)
|
||||||
@@ -97,172 +87,4 @@ def test_convert_dbus_strings():
|
|||||||
assert isinstance(converted[1], int), f"Expected int, got {type(converted[1])}"
|
assert isinstance(converted[1], int), f"Expected int, got {type(converted[1])}"
|
||||||
assert isinstance(converted[3], float), f"Expected float, got {type(converted[3])}"
|
assert isinstance(converted[3], float), f"Expected float, got {type(converted[3])}"
|
||||||
assert converted[0] == 'Test String 1'
|
assert converted[0] == 'Test String 1'
|
||||||
assert converted[2] == 'Test String 2'
|
assert converted[2] == 'Test String 2'
|
||||||
|
|
||||||
'''
|
|
||||||
# Process du serveur mock Notifications (exposé aux tests)
|
|
||||||
_DBUS_NOTIFY_PROC = None
|
|
||||||
|
|
||||||
|
|
||||||
class _DBusEnv(DBusTestCase):
|
|
||||||
"""Accès aux helpers de DBusTestCase sans classe de tests."""
|
|
||||||
pass
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(scope="session", autouse=True)
|
|
||||||
def _dbus_session_notifications():
|
|
||||||
"""Lance un bus de session et un service Notifications mocké pour toute la session."""
|
|
||||||
_DBusEnv.start_session_bus()
|
|
||||||
|
|
||||||
# Démarrer le mock org.freedesktop.Notifications et CAPTURER stdout
|
|
||||||
p = _DBusEnv.spawn_server(
|
|
||||||
'org.freedesktop.Notifications',
|
|
||||||
'/org/freedesktop/Notifications',
|
|
||||||
'org.freedesktop.Notifications',
|
|
||||||
False, # system_bus=False
|
|
||||||
subprocess.PIPE # stdout du mock lisible côté test
|
|
||||||
)
|
|
||||||
|
|
||||||
# Récupérer l'objet mock et injecter les méthodes nécessaires
|
|
||||||
bus = _DBusEnv.get_dbus(False) # False => session bus
|
|
||||||
obj = bus.get_object('org.freedesktop.Notifications',
|
|
||||||
'/org/freedesktop/Notifications')
|
|
||||||
mock = dbus.Interface(obj, MOCK_IFACE)
|
|
||||||
|
|
||||||
# NB: 5e argument = code Python (string) exécuté côté mock
|
|
||||||
mock.AddMethod('org.freedesktop.Notifications', 'GetServerInformation',
|
|
||||||
'', 'ssss',
|
|
||||||
'ret = ("pytest-notify","pytest-vendor","1.0","1.2")')
|
|
||||||
mock.AddMethod('org.freedesktop.Notifications', 'GetCapabilities',
|
|
||||||
'', 'as',
|
|
||||||
'ret = ["actions","body","icon","sound"]')
|
|
||||||
mock.AddMethod('org.freedesktop.Notifications', 'Notify',
|
|
||||||
'susssasa{sv}i', 'u',
|
|
||||||
'ret = 1')
|
|
||||||
mock.AddMethod('org.freedesktop.Notifications', 'CloseNotification',
|
|
||||||
'u', '',
|
|
||||||
'import sys; print("Closed %d" % args[0]); sys.stdout.flush()')
|
|
||||||
|
|
||||||
# Attendre que le nom soit effectivement enregistré sur le bus (évite la course)
|
|
||||||
dbus_daemon = dbus.Interface(
|
|
||||||
bus.get_object('org.freedesktop.DBus', '/org/freedesktop/DBus'),
|
|
||||||
'org.freedesktop.DBus'
|
|
||||||
)
|
|
||||||
for _ in range(50):
|
|
||||||
if dbus_daemon.NameHasOwner('org.freedesktop.Notifications'):
|
|
||||||
break
|
|
||||||
time.sleep(0.1)
|
|
||||||
else:
|
|
||||||
pytest.fail("org.freedesktop.Notifications n'a pas pris le nom sur le bus")
|
|
||||||
|
|
||||||
# Exposer le process du mock aux tests
|
|
||||||
global _DBUS_NOTIFY_PROC
|
|
||||||
_DBUS_NOTIFY_PROC = p
|
|
||||||
|
|
||||||
yield
|
|
||||||
# Pas de teardown spécifique requis
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
|
||||||
def notify_proc():
|
|
||||||
"""Retourne le process du serveur mock pour lire son stdout."""
|
|
||||||
return _DBUS_NOTIFY_PROC
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
|
||||||
def notifier(monkeypatch):
|
|
||||||
"""Instancie DBusNotify en forçant l'usage du même SessionBus que la fixture."""
|
|
||||||
bus = _DBusEnv.get_dbus(False) # session bus lancé ci-dessus
|
|
||||||
import slic.utils.dbusnotify as dbn
|
|
||||||
# S'assure que DBusNotify (et code associé) utilise EXACTEMENT ce bus
|
|
||||||
monkeypatch.setattr(dbn.dbus, "SessionBus", lambda: bus)
|
|
||||||
monkeypatch.setattr(dbn.dbus, "SystemBus", lambda: bus) # garde-fou
|
|
||||||
return dbn.DBusNotify()
|
|
||||||
|
|
||||||
|
|
||||||
def test_notify_create(notifier):
|
|
||||||
"""Test la création d'une notification."""
|
|
||||||
from slic.utils.dbusnotify import Notification
|
|
||||||
notification = notifier.notify("Test Notification", "This is a test notification.")
|
|
||||||
|
|
||||||
assert isinstance(notification, Notification)
|
|
||||||
assert notification.summary == "Test Notification"
|
|
||||||
assert notification.body == "This is a test notification."
|
|
||||||
|
|
||||||
|
|
||||||
def test_notify_update(notifier):
|
|
||||||
"""Test la mise à jour d'une notification."""
|
|
||||||
notification = notifier.notify("Test Notification", "This is a test notification.")
|
|
||||||
notification.update(summary="Updated Test", body="This notification has been updated.")
|
|
||||||
|
|
||||||
assert notification.summary == "Updated Test"
|
|
||||||
assert notification.body == "This notification has been updated."
|
|
||||||
|
|
||||||
|
|
||||||
def test_get_server_info(notifier):
|
|
||||||
"""Test l'obtention des informations du serveur."""
|
|
||||||
server_info = notifier.get_server_info()
|
|
||||||
|
|
||||||
assert isinstance(server_info, dict)
|
|
||||||
assert "name" in server_info and server_info["name"], "Server name is empty"
|
|
||||||
assert "vendor" in server_info and server_info["vendor"], "Server vendor is empty"
|
|
||||||
assert "version" in server_info and server_info["version"], "Server version is empty"
|
|
||||||
assert "spec" in server_info and server_info["spec"], "Server spec is empty"
|
|
||||||
|
|
||||||
|
|
||||||
def test_get_capabilities(notifier):
|
|
||||||
"""Test l'obtention des capacités du serveur."""
|
|
||||||
# Normaliser: list[str] (gère tuple/dbus.Array/dbus.String)
|
|
||||||
capabilities = [str(x) for x in list(notifier.get_capabilities())]
|
|
||||||
|
|
||||||
assert isinstance(capabilities, list), "Capabilities should be a list."
|
|
||||||
assert len(capabilities) > 0, "Capabilities list is empty."
|
|
||||||
|
|
||||||
expected_capabilities = ['actions', 'body', 'icon', 'sound']
|
|
||||||
for cap in expected_capabilities:
|
|
||||||
assert cap in capabilities, f"Expected capability '{cap}' not found."
|
|
||||||
|
|
||||||
for cap in capabilities:
|
|
||||||
assert isinstance(cap, str), f"Each capability should be a string, but found {type(cap)}."
|
|
||||||
|
|
||||||
|
|
||||||
def test_notify_and_close(notifier, notify_proc):
|
|
||||||
"""Test l'envoi d'une notification et sa fermeture."""
|
|
||||||
notification = notifier.notify("Test Close", "This notification will be closed.")
|
|
||||||
time.sleep(0.2) # petite marge
|
|
||||||
notification.close()
|
|
||||||
|
|
||||||
# Lire la ligne imprimée par le mock ("Closed <id>")
|
|
||||||
deadline = time.time() + 2.0
|
|
||||||
seen = False
|
|
||||||
while time.time() < deadline:
|
|
||||||
line = notify_proc.stdout.readline().decode(errors="ignore").strip()
|
|
||||||
if not line:
|
|
||||||
time.sleep(0.05)
|
|
||||||
continue
|
|
||||||
if "Closed" in line:
|
|
||||||
seen = True
|
|
||||||
break
|
|
||||||
|
|
||||||
assert seen, "Notification was not closed properly."
|
|
||||||
|
|
||||||
|
|
||||||
def test_notify_invalid_value(notifier):
|
|
||||||
"""On attend une exception si la valeur est incorrecte (body non string)."""
|
|
||||||
with pytest.raises(Exception):
|
|
||||||
notifier.notify("Invalid Test", 1234)
|
|
||||||
|
|
||||||
|
|
||||||
def test_convert_dbus_strings():
|
|
||||||
"""Vérifie la conversion de dbus.String en str dans une séquence mixte."""
|
|
||||||
from slic.utils.dbusnotify import convert_dbus_strings
|
|
||||||
test_sequence = [dbus.String('Test String 1'), 42, dbus.String('Test String 2'), 3.14]
|
|
||||||
converted = convert_dbus_strings(test_sequence)
|
|
||||||
|
|
||||||
assert isinstance(converted[0], str), f"Expected str, got {type(converted[0])}"
|
|
||||||
assert isinstance(converted[2], str), f"Expected str, got {type(converted[2])}"
|
|
||||||
assert isinstance(converted[1], int), f"Expected int, got {type(converted[1])}"
|
|
||||||
assert isinstance(converted[3], float), f"Expected float, got {type(converted[3])}"
|
|
||||||
assert converted[0] == 'Test String 1'
|
|
||||||
assert converted[2] == 'Test String 2'
|
|
||||||
'''
|
|
||||||
Reference in New Issue
Block a user