robojerk/particle-os-tools
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
particle-os-tools/test_dbus_next.py
Joe Particle 8c470a56b5 feat: Implement production-ready systemd service best practices 
- Update systemd service file with Type=simple and comprehensive locking
- Enhance D-Bus service and policy files for proper activation
- Remove OSTree dependency for test mode compatibility
- Implement automated service file installation and cleanup
- Add comprehensive systemd usage documentation
- Update changelog and TODO to reflect completed systemd improvements
- Service now successfully running under systemd management

This completes the systemd service integration with production-ready
configuration and best practices for daemon lifecycle management.
2025-07-16 16:13:35 +00:00

332 lines
No EOL
11 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
D-Bus method testing using dbus-next
Based on https://python-dbus-next.readthedocs.io/en/latest/
"""
import asyncio
import json
import sys
from dbus_next import BusType, DBusError
from dbus_next.aio import MessageBus
from dbus_next.message import Message
async def test_dbus_methods():
"""Test all D-Bus methods using dbus-next"""
try:
# Connect to system bus
bus = await MessageBus(bus_type=BusType.SYSTEM).connect()
print("=== Testing D-Bus Methods with dbus-next ===")
print()
# Test 1: GetStatus
print("1. Testing GetStatus...")
try:
reply = await bus.call(
Message(
destination='org.debian.aptostree1',
path='/org/debian/aptostree1/Sysroot',
interface='org.debian.aptostree1.Sysroot',
member='GetStatus'
)
)
result = json.loads(reply.body[0])
print(f" ✓ SUCCESS: {result}")
except Exception as e:
print(f" ✗ FAILED: {e}")
print()
# Test 2: GetOS
print("2. Testing GetOS...")
try:
reply = await bus.call(
Message(
destination='org.debian.aptostree1',
path='/org/debian/aptostree1/Sysroot',
interface='org.debian.aptostree1.Sysroot',
member='GetOS'
)
)
print(f" ✓ SUCCESS: {reply.body[0]}")
except Exception as e:
print(f" ✗ FAILED: {e}")
print()
# Test 3: Reload
print("3. Testing Reload...")
try:
reply = await bus.call(
Message(
destination='org.debian.aptostree1',
path='/org/debian/aptostree1/Sysroot',
interface='org.debian.aptostree1.Sysroot',
member='Reload'
)
)
print(" ✓ SUCCESS")
except Exception as e:
print(f" ✗ FAILED: {e}")
print()
# Test 4: ReloadConfig
print("4. Testing ReloadConfig...")
try:
reply = await bus.call(
Message(
destination='org.debian.aptostree1',
path='/org/debian/aptostree1/Sysroot',
interface='org.debian.aptostree1.Sysroot',
member='ReloadConfig'
)
)
print(" ✓ SUCCESS")
except Exception as e:
print(f" ✗ FAILED: {e}")
print()
# Test 5: RegisterClient
print("5. Testing RegisterClient...")
try:
options = {'id': 'test-client'}
reply = await bus.call(
Message(
destination='org.debian.aptostree1',
path='/org/debian/aptostree1/Sysroot',
interface='org.debian.aptostree1.Sysroot',
member='RegisterClient',
body=[options]
)
)
print(" ✓ SUCCESS")
except Exception as e:
print(f" ✗ FAILED: {e}")
print()
# Test 6: InstallPackages
print("6. Testing InstallPackages...")
try:
packages = ['curl']
reply = await bus.call(
Message(
destination='org.debian.aptostree1',
path='/org/debian/aptostree1/Sysroot',
interface='org.debian.aptostree1.Sysroot',
member='InstallPackages',
body=[packages, False]
)
)
result = reply.body[0]
print(f" ✓ SUCCESS: {result}")
except Exception as e:
print(f" ✗ FAILED: {e}")
print()
# Test 7: RemovePackages
print("7. Testing RemovePackages...")
try:
packages = ['curl']
reply = await bus.call(
Message(
destination='org.debian.aptostree1',
path='/org/debian/aptostree1/Sysroot',
interface='org.debian.aptostree1.Sysroot',
member='RemovePackages',
body=[packages, False]
)
)
result = reply.body[0]
print(f" ✓ SUCCESS: {result}")
except Exception as e:
print(f" ✗ FAILED: {e}")
print()
# Test 8: Deploy
print("8. Testing Deploy...")
try:
options = {'test': 'value'}
reply = await bus.call(
Message(
destination='org.debian.aptostree1',
path='/org/debian/aptostree1/Sysroot',
interface='org.debian.aptostree1.Sysroot',
member='Deploy',
body=['test-layer', options]
)
)
result = reply.body[0]
print(f" ✓ SUCCESS: {result}")
except Exception as e:
print(f" ✗ FAILED: {e}")
print()
# Test 9: Upgrade
print("9. Testing Upgrade...")
try:
options = {'test': 'value'}
reply = await bus.call(
Message(
destination='org.debian.aptostree1',
path='/org/debian/aptostree1/Sysroot',
interface='org.debian.aptostree1.Sysroot',
member='Upgrade',
body=[options]
)
)
result = reply.body[0]
print(f" ✓ SUCCESS: {result}")
except Exception as e:
print(f" ✗ FAILED: {e}")
print()
# Test 10: Rollback
print("10. Testing Rollback...")
try:
options = {'test': 'value'}
reply = await bus.call(
Message(
destination='org.debian.aptostree1',
path='/org/debian/aptostree1/Sysroot',
interface='org.debian.aptostree1.Sysroot',
member='Rollback',
body=[options]
)
)
result = reply.body[0]
print(f" ✓ SUCCESS: {result}")
except Exception as e:
print(f" ✗ FAILED: {e}")
print()
# Test 11: CreateComposeFSLayer
print("11. Testing CreateComposeFSLayer...")
try:
reply = await bus.call(
Message(
destination='org.debian.aptostree1',
path='/org/debian/aptostree1/Sysroot',
interface='org.debian.aptostree1.Sysroot',
member='CreateComposeFSLayer',
body=['/tmp/test-source', '/tmp/test-layer', '/tmp/test-digest']
)
)
result = reply.body[0]
print(f" ✓ SUCCESS: {result}")
except Exception as e:
print(f" ✗ FAILED: {e}")
print()
# Test 12: UnregisterClient
print("12. Testing UnregisterClient...")
try:
options = {'id': 'test-client'}
reply = await bus.call(
Message(
destination='org.debian.aptostree1',
path='/org/debian/aptostree1/Sysroot',
interface='org.debian.aptostree1.Sysroot',
member='UnregisterClient',
body=[options]
)
)
print(" ✓ SUCCESS")
except Exception as e:
print(f" ✗ FAILED: {e}")
print()
# Test Properties
print("=== Testing D-Bus Properties ===")
print()
# Test Sysroot properties
properties_to_test = [
'Booted', 'Path', 'ActiveTransaction', 'ActiveTransactionPath',
'Deployments', 'AutomaticUpdatePolicy'
]
for prop in properties_to_test:
print(f"Testing property: {prop}")
try:
reply = await bus.call(
Message(
destination='org.debian.aptostree1',
path='/org/debian/aptostree1/Sysroot',
interface='org.freedesktop.DBus.Properties',
member='Get',
body=['org.debian.aptostree1.Sysroot', prop]
)
)
print(f" ✓ SUCCESS: {reply.body[0]}")
except Exception as e:
print(f" ✗ FAILED: {e}")
print()
# Test OS interface methods
print("=== Testing OS Interface Methods ===")
print()
# Test GetBootedDeployment
print("Testing GetBootedDeployment...")
try:
reply = await bus.call(
Message(
destination='org.debian.aptostree1',
path='/org/debian/aptostree1/OS/default',
interface='org.debian.aptostree1.OS',
member='GetBootedDeployment'
)
)
result = json.loads(reply.body[0])
print(f" ✓ SUCCESS: {result}")
except Exception as e:
print(f" ✗ FAILED: {e}")
print()
# Test GetDefaultDeployment
print("Testing GetDefaultDeployment...")
try:
reply = await bus.call(
Message(
destination='org.debian.aptostree1',
path='/org/debian/aptostree1/OS/default',
interface='org.debian.aptostree1.OS',
member='GetDefaultDeployment'
)
)
result = json.loads(reply.body[0])
print(f" ✓ SUCCESS: {result}")
except Exception as e:
print(f" ✗ FAILED: {e}")
print()
# Test ListDeployments
print("Testing ListDeployments...")
try:
reply = await bus.call(
Message(
destination='org.debian.aptostree1',
path='/org/debian/aptostree1/OS/default',
interface='org.debian.aptostree1.OS',
member='ListDeployments'
)
)
result = json.loads(reply.body[0])
print(f" ✓ SUCCESS: {result}")
except Exception as e:
print(f" ✗ FAILED: {e}")
print()
print("=== Testing Complete ===")
except DBusError as e:
print(f"D-Bus error: {e}")
return 1
except Exception as e:
print(f"Error: {e}")
return 1
return 0
if __name__ == "__main__":
sys.exit(asyncio.run(test_dbus_methods()))
Reference in a new issue View git blame Copy permalink