particle-os-tools/test_dbus_integrated.py

#!/usr/bin/env python3
"""
Integrated D-Bus testing - starts daemon and tests it in the same process
"""

import asyncio
import json
import logging
import sys
from dbus_next import BusType, DBusError
from dbus_next.aio import MessageBus
from dbus_next.message import Message
from dbus_next.service import ServiceInterface, method
from dbus_next import Variant

# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('test')

class TestSysrootInterface(ServiceInterface):
    """Test implementation of Sysroot interface"""
    
    def __init__(self):
        super().__init__("org.debian.aptostree1.Sysroot")
        self.test_mode = True
    
    @method()
    def GetStatus(self) -> 's':
        """Get system status as JSON string"""
        status = {
            'daemon_running': True,
            'test_mode': True,
            'active_transactions': 0,
            'sysroot_path': '/var/lib/apt-ostree'
        }
        return json.dumps(status)
    
    @method()
    def GetOS(self) -> 'ao':
        """Get list of OS instances"""
        return ['/org/debian/aptostree1/OS/default']
    
    @method()
    def Reload(self):
        """Reload sysroot state"""
        logger.info("Reload called")
    
    @method()
    def ReloadConfig(self):
        """Reload configuration"""
        logger.info("ReloadConfig called")
    
    @method()
    def RegisterClient(self, options: 'a{sv}'):
        """Register a client"""
        logger.info(f"RegisterClient called with options: {options}")
    
    @method()
    def UnregisterClient(self, options: 'a{sv}'):
        """Unregister a client"""
        logger.info(f"UnregisterClient called with options: {options}")
    
    @method()
    def InstallPackages(self, packages: 'as', live_install: 'b') -> 'a{sv}':
        """Install packages"""
        logger.info(f"InstallPackages called with packages: {packages}, live_install: {live_install}")
        return {
            'success': True,
            'transaction_id': 'test-123',
            'packages': list(packages),
            'live_install': live_install,
            'message': 'Test installation successful'
        }
    
    @method()
    def RemovePackages(self, packages: 'as', live_remove: 'b') -> 'a{sv}':
        """Remove packages"""
        logger.info(f"RemovePackages called with packages: {packages}, live_remove: {live_remove}")
        return {
            'success': True,
            'transaction_id': 'test-456',
            'packages': list(packages),
            'live_remove': live_remove,
            'message': 'Test removal successful'
        }
    
    @method()
    def Deploy(self, layer_name: 's', options: 'a{sv}') -> 'a{sv}':
        """Deploy a layer"""
        logger.info(f"Deploy called with layer_name: {layer_name}, options: {options}")
        return {
            'success': True,
            'transaction_id': 'test-789',
            'layer_name': layer_name,
            'message': 'Test deployment successful'
        }
    
    @method()
    def Upgrade(self, options: 'a{sv}') -> 'a{sv}':
        """Upgrade system"""
        logger.info(f"Upgrade called with options: {options}")
        return {
            'success': True,
            'transaction_id': 'test-upgrade',
            'message': 'Test upgrade successful'
        }
    
    @method()
    def Rollback(self, options: 'a{sv}') -> 'a{sv}':
        """Rollback system"""
        logger.info(f"Rollback called with options: {options}")
        return {
            'success': True,
            'transaction_id': 'test-rollback',
            'message': 'Test rollback successful'
        }
    
    @method()
    def CreateComposeFSLayer(self, source_dir: 's', layer_path: 's', digest_store: 's') -> 'a{sv}':
        """Create ComposeFS layer"""
        logger.info(f"CreateComposeFSLayer called with source_dir: {source_dir}, layer_path: {layer_path}, digest_store: {digest_store}")
        return {
            'success': True,
            'transaction_id': 'test-composefs',
            'source_dir': source_dir,
            'layer_path': layer_path,
            'digest_store': digest_store,
            'message': 'Test ComposeFS layer creation successful'
        }

class TestOSInterface(ServiceInterface):
    """Test implementation of OS interface"""
    
    def __init__(self):
        super().__init__("org.debian.aptostree1.OS")
        self.test_mode = True
    
    @method()
    def GetBootedDeployment(self) -> 's':
        """Get currently booted deployment"""
        deployment = {
            'booted_deployment': {
                'id': 'test-booted',
                'osname': 'default',
                'deployment_id': 'test-deployment-1'
            }
        }
        return json.dumps(deployment)
    
    @method()
    def GetDefaultDeployment(self) -> 's':
        """Get default deployment"""
        deployment = {
            'default_deployment': {
                'id': 'test-default',
                'osname': 'default',
                'deployment_id': 'test-deployment-1'
            }
        }
        return json.dumps(deployment)
    
    @method()
    def ListDeployments(self) -> 's':
        """List all deployments"""
        deployments = {
            'deployments': [
                {
                    'id': 'test-deployment-1',
                    'osname': 'default',
                    'booted': True
                },
                {
                    'id': 'test-deployment-2',
                    'osname': 'default',
                    'booted': False
                }
            ]
        }
        return json.dumps(deployments)

async def start_test_daemon():
    """Start the test daemon"""
    logger.info("Starting test daemon...")
    
    # Create D-Bus connection
    bus = await MessageBus(bus_type=BusType.SYSTEM).connect()
    
    # Create and export interfaces
    sysroot_interface = TestSysrootInterface()
    os_interface = TestOSInterface()
    
    # Export interfaces
    bus.export('/org/debian/aptostree1/Sysroot', sysroot_interface)
    bus.export('/org/debian/aptostree1/OS/default', os_interface)
    
    # Request name
    await bus.request_name('org.debian.aptostree1')
    
    logger.info("Test daemon started successfully")
    return bus

async def test_dbus_methods(bus):
    """Test all D-Bus methods"""
    
    print("=== Testing D-Bus Methods ===")
    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: InstallPackages
    print("3. Testing InstallPackages...")
    try:
        packages = ['curl', 'wget']
        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 4: RemovePackages
    print("4. 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 5: Deploy
    print("5. 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 6: Upgrade
    print("6. 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 7: Rollback
    print("7. 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 8: CreateComposeFSLayer
    print("8. 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 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 ===")

async def main():
    """Main test function"""
    try:
        # Start test daemon
        bus = await start_test_daemon()
        
        # Wait a moment for daemon to be ready
        await asyncio.sleep(1)
        
        # Test methods
        await test_dbus_methods(bus)
        
        # Keep daemon running for a bit to see results
        await asyncio.sleep(2)
        
    except Exception as e:
        logger.error(f"Test failed: {e}")
        return 1
    
    return 0

if __name__ == "__main__":
    sys.exit(asyncio.run(main()))