81 lines
2.4 KiB
Python
81 lines
2.4 KiB
Python
import asyncio
|
|
|
|
import pytest
|
|
|
|
from app.core.errors import EngineError
|
|
from app.core.interface import EngineBackend
|
|
from app.core.manager import EngineManager
|
|
from app.core.registry import BackendRegistry
|
|
from app.models import (
|
|
CapabilitySet,
|
|
EngineKind,
|
|
EngineState,
|
|
RuntimeRequest,
|
|
RuntimeSnapshot,
|
|
RuntimeStats,
|
|
)
|
|
|
|
|
|
class FakeBackend(EngineBackend):
|
|
name = "fake"
|
|
capabilities = CapabilitySet(supports_proxy=True, supports_desktop=True)
|
|
|
|
def __init__(self) -> None:
|
|
self.state = RuntimeSnapshot(engine=EngineKind.SIGNBOX)
|
|
|
|
async def start(self, request: RuntimeRequest) -> RuntimeSnapshot:
|
|
self.state = RuntimeSnapshot(
|
|
state=EngineState.RUNNING, engine=request.engine, mode=request.mode
|
|
)
|
|
return self.state
|
|
|
|
async def stop(self) -> RuntimeSnapshot:
|
|
self.state = self.state.model_copy(update={"state": EngineState.STOPPED})
|
|
return self.state
|
|
|
|
async def restart(self, request: RuntimeRequest) -> RuntimeSnapshot:
|
|
return await self.start(request)
|
|
|
|
async def get_state(self) -> RuntimeSnapshot:
|
|
return self.state
|
|
|
|
async def get_stats(self) -> RuntimeStats:
|
|
return RuntimeStats()
|
|
|
|
|
|
def manager() -> EngineManager:
|
|
registry = BackendRegistry()
|
|
registry.register(EngineKind.SIGNBOX, FakeBackend)
|
|
registry.register(EngineKind.FREETURN, FakeBackend)
|
|
return EngineManager(registry)
|
|
|
|
|
|
def request(engine: EngineKind) -> RuntimeRequest:
|
|
return RuntimeRequest(engine=engine, mode="proxy", profile_id="profile")
|
|
|
|
|
|
def test_manager_serializes_and_prevents_conflict() -> None:
|
|
async def run() -> None:
|
|
value = manager()
|
|
assert (await value.start(request(EngineKind.SIGNBOX))).state is EngineState.RUNNING
|
|
with pytest.raises(EngineError):
|
|
await value.start(request(EngineKind.FREETURN))
|
|
assert (await value.stop()).state is EngineState.STOPPED
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_registry_has_no_core_if_branch_requirement() -> None:
|
|
value = manager()
|
|
assert set(value.registry.kinds()) == {EngineKind.SIGNBOX, EngineKind.FREETURN}
|
|
|
|
|
|
def test_manager_can_switch_engine_after_stop() -> None:
|
|
async def run() -> None:
|
|
value = manager()
|
|
await value.start(request(EngineKind.SIGNBOX))
|
|
await value.stop()
|
|
snapshot = await value.start(request(EngineKind.FREETURN))
|
|
assert snapshot.engine is EngineKind.FREETURN
|
|
|
|
asyncio.run(run())
|