qzl
139 lines
4.1 KiB
Python
139 lines
4.1 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from services.base.service_interface import (
|
|
BaseServiceProvider,
|
|
ServiceRegistry,
|
|
close_registered_services,
|
|
initialize_registered_services,
|
|
register_service,
|
|
register_service_instance,
|
|
)
|
|
|
|
|
|
class _DummyService(BaseServiceProvider):
|
|
def __init__(self, name: str = "dummy") -> None:
|
|
super().__init__(name)
|
|
|
|
async def initialize(self, **_: object) -> bool:
|
|
self._set_initialized(True)
|
|
return True
|
|
|
|
async def close(self) -> bool:
|
|
self._set_initialized(False)
|
|
return True
|
|
|
|
async def health_check(self) -> dict[str, object]:
|
|
return {"status": "healthy", "details": {}}
|
|
|
|
|
|
def test_register_service_and_create_service() -> None:
|
|
@register_service("dummy-service")
|
|
class _RegisteredService(_DummyService):
|
|
pass
|
|
|
|
created = ServiceRegistry.create_service("dummy-service")
|
|
|
|
assert created is not None
|
|
assert created.get_service_info()["name"] == "dummy"
|
|
|
|
|
|
def test_register_service_and_get_service() -> None:
|
|
@register_service("dummy-service-get")
|
|
class _RegisteredService(_DummyService):
|
|
pass
|
|
|
|
resolved = ServiceRegistry.get_service("dummy-service-get")
|
|
|
|
assert resolved is not None
|
|
assert resolved.get_service_info()["name"] == "dummy"
|
|
|
|
|
|
def test_register_service_instance_returns_same_instance() -> None:
|
|
instance = _DummyService("singleton")
|
|
|
|
returned = register_service_instance("dummy-singleton", instance)
|
|
created = ServiceRegistry.create_service("dummy-singleton")
|
|
|
|
assert returned is instance
|
|
assert created is instance
|
|
|
|
|
|
def test_create_service_returns_none_for_missing() -> None:
|
|
assert ServiceRegistry.create_service("missing-service") is None
|
|
|
|
|
|
def test_get_service_returns_none_for_missing() -> None:
|
|
assert ServiceRegistry.get_service("missing-service") is None
|
|
|
|
|
|
class _LifecycleService(BaseServiceProvider):
|
|
def __init__(self, name: str, recorder: list[str], fail_on_init: bool = False) -> None:
|
|
super().__init__(name)
|
|
self._recorder = recorder
|
|
self._fail_on_init = fail_on_init
|
|
|
|
async def initialize(self, **_: object) -> bool:
|
|
self._recorder.append(f"init:{self.service_name}")
|
|
if self._fail_on_init:
|
|
return False
|
|
self._set_initialized(True)
|
|
return True
|
|
|
|
async def close(self) -> bool:
|
|
self._recorder.append(f"close:{self.service_name}")
|
|
self._set_initialized(False)
|
|
return True
|
|
|
|
async def health_check(self) -> dict[str, object]:
|
|
return {"status": "healthy", "details": {}}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_initialize_registered_services_success() -> None:
|
|
recorder: list[str] = []
|
|
first = register_service_instance(
|
|
"lifecycle-success-first", _LifecycleService("first", recorder)
|
|
)
|
|
second = register_service_instance(
|
|
"lifecycle-success-second", _LifecycleService("second", recorder)
|
|
)
|
|
|
|
initialized, services = await initialize_registered_services(
|
|
["lifecycle-success-first", "lifecycle-success-second"]
|
|
)
|
|
|
|
assert initialized is True
|
|
assert services == [first, second]
|
|
assert recorder == ["init:first", "init:second"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_initialize_registered_services_failure_rolls_back() -> None:
|
|
recorder: list[str] = []
|
|
register_service_instance("lifecycle-fail-first", _LifecycleService("first", recorder))
|
|
register_service_instance(
|
|
"lifecycle-fail-second", _LifecycleService("second", recorder, fail_on_init=True)
|
|
)
|
|
|
|
initialized, services = await initialize_registered_services(
|
|
["lifecycle-fail-first", "lifecycle-fail-second"]
|
|
)
|
|
|
|
assert initialized is False
|
|
assert services == []
|
|
assert recorder == ["init:first", "init:second", "close:first"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_close_registered_services_closes_in_reverse_order() -> None:
|
|
recorder: list[str] = []
|
|
first = _LifecycleService("first", recorder)
|
|
second = _LifecycleService("second", recorder)
|
|
|
|
closed = await close_registered_services([first, second])
|
|
|
|
assert closed is True
|
|
assert recorder == ["close:second", "close:first"]
|