"""Tests for core.engine — Module lifecycle manager.""" import os import time from unittest.mock import patch, MagicMock import pytest from core.engine import Engine, HARDWARE_TIERS, _topo_sort, ModuleEntry from core.bus import EventBus from core.state import StateManager from modules.base import BaseModule # --------------------------------------------------------------------------- # Mock module for testing # --------------------------------------------------------------------------- class MockModule(BaseModule): """Minimal BaseModule subclass for testing.""" name = "mock_module" module_type = "passive" priority = 100 dependencies = [] requires_root = False def start(self): self._running = True import os, time as _t self._pid = os.getpid() self._start_time = _t.time() # Block to keep the process alive import signal signal.pause() def stop(self): self._running = False def status(self): return {"running": self._running, "pid": self._pid} def configure(self, config): self.config.update(config) class MockActiveModule(BaseModule): """Active module for phase-gating tests.""" name = "mock_active" module_type = "active" priority = 200 dependencies = [] requires_root = False def start(self): self._running = True def stop(self): self._running = False def status(self): return {"running": self._running} def configure(self, config): pass class MockDepModule(BaseModule): """Module that depends on mock_module.""" name = "mock_dep" module_type = "passive" priority = 150 dependencies = ["mock_module"] requires_root = False def start(self): self._running = True def stop(self): self._running = False def status(self): return {"running": self._running} def configure(self, config): pass class TestEngine: def test_load_module(self, mock_bus, mock_state, mock_config): """Engine.register() adds a module to the registry.""" engine = Engine(bus=mock_bus, state=mock_state, config=mock_config) engine.register(MockModule, config={"test": True}) assert "mock_module" in engine._modules assert engine._modules["mock_module"].module_class is MockModule assert engine._modules["mock_module"].config == {"test": True} def test_start_stop_module(self, mock_bus, mock_state, mock_config): """Engine can start and stop a module process.""" engine = Engine(bus=mock_bus, state=mock_state, config=mock_config) engine.register(MockModule, config=mock_config) success = engine.start("mock_module") assert success is True entry = engine._modules["mock_module"] assert entry.process is not None assert entry.process.is_alive() # Stop engine.stop("mock_module") time.sleep(0.5) assert entry.process is None def test_dependency_resolution(self, mock_bus, mock_state, mock_config): """_topo_sort orders modules by dependency (dependencies first).""" engine = Engine(bus=mock_bus, state=mock_state, config=mock_config) engine.register(MockDepModule) engine.register(MockModule) order = _topo_sort(engine._modules) assert order.index("mock_module") < order.index("mock_dep") def test_hardware_tier_enforcement(self, mock_bus, mock_state, mock_config, hardware_tier_override): """Pi Zero tier blocks more than max_passive modules.""" # Set platform to pi_zero directly in config mock_config["device"]["platform"] = "pi_zero" with hardware_tier_override("pi_zero"): engine = Engine(bus=mock_bus, state=mock_state, config=mock_config) # pi_zero allows max 4 passive modules assert engine._tier == "pi_zero" limits = HARDWARE_TIERS["pi_zero"] assert limits["max_passive"] == 4 # Register 5 passive modules (create unique subclasses) for i in range(5): cls = type(f"MockPassive{i}", (MockModule,), {"name": f"passive_{i}"}) engine.register(cls, config=mock_config) # Start modules — first 4 should succeed, 5th should be blocked results = [] for i in range(5): # Mock the process as alive for previously started ones for name, entry in engine._modules.items(): if entry.process is not None: entry.process = MagicMock() entry.process.is_alive.return_value = True results.append(engine.start(f"passive_{i}")) # At least some should succeed and some should be blocked started = sum(1 for r in results if r is True) # The first 4 should start, 5th blocked by tier limit assert started <= limits["max_passive"] def test_engagement_phase_gating(self, mock_bus, mock_state, mock_config): """passive_only phase blocks active modules.""" mock_config["engagement_phase"]["mode"] = "passive_only" engine = Engine(bus=mock_bus, state=mock_state, config=mock_config) engine.register(MockActiveModule, config=mock_config) success = engine.start("mock_active") assert success is False # Switch to active_allowed engine.set_phase("active_allowed") # Register again since it was already registered success = engine.start("mock_active") assert success is True engine.stop("mock_active") def test_capture_bus_instantiated_in_start_all(self, mock_bus, mock_state, mock_config): """Engine.start_all() creates and injects CaptureBus into passive modules.""" # Create a passive module that requires CaptureBus class MockPassiveWithCaptureBus(BaseModule): name = "mock_passive_capture" module_type = "passive" priority = 100 dependencies = [] requires_root = False requires_capture_bus = True def start(self): # This module should receive capture_bus in config self._capture_bus = self.config.get("capture_bus") if not self._capture_bus: raise RuntimeError("capture_bus not provided in config") self._running = True self._pid = os.getpid() self._start_time = time.time() def stop(self): self._running = False def status(self): return {"running": self._running, "pid": self._pid} def configure(self, config): pass engine = Engine(bus=mock_bus, state=mock_state, config=mock_config) engine.register(MockPassiveWithCaptureBus, config=mock_config) # Mock CaptureBus.start() to avoid permission errors during test with patch("core.capture_bus.CaptureBus.start"): # start_all() should create CaptureBus and inject it results = engine.start_all() # Verify CaptureBus was created assert engine.capture_bus is not None assert results.get("mock_passive_capture") is True # Verify the module config has capture_bus injected entry = engine._modules["mock_passive_capture"] assert "capture_bus" in entry.config assert entry.config["capture_bus"] is not None # Cleanup engine.stop_all()