AskaEth
e6875f0b4b
- 13-service async plugin framework - Textual TUI with CLI fallback - Plugin hot-reload + permission system - Web management panel (aiohttp) - Bridge-based inter-module communication - 10 regression tests Fixes applied: - PBKDF2-SHA256 auth (was plain SHA256) - Auth bypass removed (was allow-all on fail) - Bare excepts replaced with logged errors - CatFramework/DreamSu -> SenSu naming unified - ServiceManager: health checks + startup_order - Env var credentials (SENSU_ADMIN_PASSWORD etc)
50 lines
1.5 KiB
Python
50 lines
1.5 KiB
Python
import pytest, asyncio
|
|
from service_manager import ServiceManager
|
|
|
|
class MockService:
|
|
def __init__(self, name="mock"):
|
|
self.name = name
|
|
self.shutdown_called = False
|
|
def shutdown(self):
|
|
self.shutdown_called = True
|
|
return True
|
|
|
|
class AsyncMockService:
|
|
def __init__(self, name="async_mock"):
|
|
self.name = name
|
|
self.shutdown_called = False
|
|
async def shutdown(self):
|
|
self.shutdown_called = True
|
|
|
|
class TestServiceManager:
|
|
def test_register_get(self):
|
|
sm = ServiceManager(); s = MockService()
|
|
sm.register_service("t", s)
|
|
assert sm.get_service("t") is s
|
|
|
|
def test_missing_raises(self):
|
|
with pytest.raises(ValueError):
|
|
ServiceManager().get_service("x")
|
|
|
|
def test_has_service(self):
|
|
sm = ServiceManager()
|
|
sm.register_service("a", MockService())
|
|
assert sm.has_service("a") and not sm.has_service("b")
|
|
|
|
def test_shutdown_sync(self):
|
|
sm = ServiceManager(); s = MockService()
|
|
sm.register_service("s", s)
|
|
sm.shutdown_all()
|
|
assert s.shutdown_called
|
|
|
|
def test_health(self):
|
|
sm = ServiceManager()
|
|
sm.register_service("h", MockService(), health_check=lambda: True)
|
|
assert asyncio.run(sm.check_health())["h"]
|
|
|
|
def test_startup_order(self):
|
|
sm = ServiceManager()
|
|
sm.register_service("1", MockService())
|
|
sm.register_service("2", MockService())
|
|
assert sm.startup_order == ["1", "2"]
|