Ryan Malloy
fa41ffbf80
- Add asyncio.wait_for timeout (30s) to all D-Bus calls
- Add asyncio.Lock to BusManager.get_bus to prevent race conditions
- Convert recursive tree walk to bounded BFS with visited set, max depth (20), max nodes (500)
- Add input validation for D-Bus names and object paths at tool boundary
- Standardize error handling (RuntimeError, TimeoutError) across all tools
- Catch JSONDecodeError in deserialize_args and set_property
- Check arg count matches signature in deserialize_args
- Add explicit variant wrapping via {"signature": "ai", "value": [1,2,3]}
- Narrow resource exception handling from Exception to (RuntimeError, OSError)
- Guard disconnect_all per-bus to avoid partial cleanup on error
- Audit log system bus calls and all set_property calls to stderr
- Consolidate _get_mgr into get_mgr in _bus.py
- Sort MPRIS players for deterministic auto-discovery
- Fix progress reporting (pass None as total when unknown)
- Add 7 new tests for validation, arg count, and JSON error paths (32 total)
114 lines
3.6 KiB
Python
114 lines
3.6 KiB
Python
"""Tests for BusManager and serialization utilities."""
|
|
|
|
import pytest
|
|
|
|
from mcdbus._bus import (
|
|
BusManager,
|
|
deserialize_args,
|
|
serialize_variant,
|
|
validate_bus_name,
|
|
validate_object_path,
|
|
)
|
|
|
|
|
|
class TestBusManager:
|
|
async def test_connect_session(self, bus_manager: BusManager):
|
|
bus = await bus_manager.get_bus("session")
|
|
assert bus.connected
|
|
|
|
async def test_connect_system(self, bus_manager: BusManager):
|
|
bus = await bus_manager.get_bus("system")
|
|
assert bus.connected
|
|
|
|
async def test_cached_connection(self, bus_manager: BusManager):
|
|
bus1 = await bus_manager.get_bus("session")
|
|
bus2 = await bus_manager.get_bus("session")
|
|
assert bus1 is bus2
|
|
|
|
async def test_invalid_bus_type(self, bus_manager: BusManager):
|
|
with pytest.raises(ValueError, match="must be 'session' or 'system'"):
|
|
await bus_manager.get_bus("invalid")
|
|
|
|
async def test_disconnect_all(self, bus_manager: BusManager):
|
|
bus = await bus_manager.get_bus("session")
|
|
assert bus.connected
|
|
await bus_manager.disconnect_all()
|
|
assert not bus.connected
|
|
|
|
|
|
class TestSerializeVariant:
|
|
def test_plain_values(self):
|
|
assert serialize_variant("hello") == "hello"
|
|
assert serialize_variant(42) == 42
|
|
assert serialize_variant(True) is True
|
|
|
|
def test_list(self):
|
|
assert serialize_variant([1, 2, 3]) == [1, 2, 3]
|
|
|
|
def test_dict(self):
|
|
assert serialize_variant({"a": 1}) == {"a": 1}
|
|
|
|
def test_nested(self):
|
|
result = serialize_variant({"a": [1, {"b": "c"}]})
|
|
assert result == {"a": [1, {"b": "c"}]}
|
|
|
|
def test_bytes_utf8(self):
|
|
assert serialize_variant(b"hello") == "hello"
|
|
|
|
def test_bytes_binary(self):
|
|
result = serialize_variant(b"\xff\xfe")
|
|
assert result == [255, 254]
|
|
|
|
|
|
class TestDeserializeArgs:
|
|
def test_empty(self):
|
|
assert deserialize_args("", "") == []
|
|
assert deserialize_args("[]", "") == []
|
|
assert deserialize_args("null", "") == []
|
|
|
|
def test_simple_args(self):
|
|
result = deserialize_args('["hello", 42]', "su")
|
|
assert result == ["hello", 42]
|
|
|
|
def test_no_signature(self):
|
|
result = deserialize_args('["hello"]', "")
|
|
assert result == ["hello"]
|
|
|
|
def test_single_value_wraps_in_list(self):
|
|
result = deserialize_args('"hello"', "s")
|
|
assert result == ["hello"]
|
|
|
|
def test_arg_count_mismatch_raises(self):
|
|
with pytest.raises(ValueError, match="expects 1 argument"):
|
|
deserialize_args('["hello", "extra"]', "s")
|
|
|
|
def test_too_few_args_raises(self):
|
|
with pytest.raises(ValueError, match="expects 2 argument"):
|
|
deserialize_args('["hello"]', "su")
|
|
|
|
def test_invalid_json_raises(self):
|
|
with pytest.raises(ValueError, match="Invalid JSON"):
|
|
deserialize_args("{broken", "s")
|
|
|
|
|
|
class TestValidation:
|
|
def test_valid_bus_name(self):
|
|
validate_bus_name("org.freedesktop.DBus")
|
|
validate_bus_name("org.kde.KWin")
|
|
|
|
def test_invalid_bus_name(self):
|
|
with pytest.raises(ValueError, match="Invalid D-Bus"):
|
|
validate_bus_name("not-a-bus-name")
|
|
with pytest.raises(ValueError, match="Invalid D-Bus"):
|
|
validate_bus_name("")
|
|
|
|
def test_valid_object_path(self):
|
|
validate_object_path("/")
|
|
validate_object_path("/org/freedesktop/DBus")
|
|
|
|
def test_invalid_object_path(self):
|
|
with pytest.raises(ValueError, match="Invalid D-Bus object path"):
|
|
validate_object_path("not/a/path")
|
|
with pytest.raises(ValueError, match="Invalid D-Bus object path"):
|
|
validate_object_path("")
|