94 lines
2.9 KiB
Python
94 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from infra_controller.config import ControllerConfig
|
|
from infra_controller.controller import InfraController
|
|
from infra_controller.discovery import AppRegistration, InfraMetadata
|
|
|
|
|
|
@dataclass
|
|
class FakeServiceResult:
|
|
returncode: int = 0
|
|
stdout: str = ""
|
|
stderr: str = ""
|
|
|
|
|
|
class FakeServiceManager:
|
|
def __init__(self):
|
|
self.applied: list[str] = []
|
|
self.stopped: list[str] = []
|
|
|
|
def apply_service(self, service_name: str) -> FakeServiceResult:
|
|
self.applied.append(service_name)
|
|
return FakeServiceResult()
|
|
|
|
def stop_service(self, service_name: str) -> FakeServiceResult:
|
|
self.stopped.append(service_name)
|
|
return FakeServiceResult()
|
|
|
|
|
|
class FakeDiscoveryManager:
|
|
def __init__(self, apps: dict[str, AppRegistration]):
|
|
self._apps = apps
|
|
|
|
def set_apps(self, apps: dict[str, AppRegistration]) -> None:
|
|
self._apps = apps
|
|
|
|
def discover_all(self) -> dict[str, AppRegistration]:
|
|
return dict(self._apps)
|
|
|
|
|
|
def _app(name: str, services: list[str]) -> AppRegistration:
|
|
md = InfraMetadata(project=name, requires={"services": services})
|
|
return AppRegistration(name=name, metadata=md, last_seen=datetime.now(), discovery_method="test")
|
|
|
|
|
|
def test_controller_stops_unused_services_after_grace_period(tmp_path: Path, monkeypatch):
|
|
cfg = ControllerConfig()
|
|
cfg.services.grace_period_minutes = 0
|
|
cfg.services.state_file = tmp_path / "state.json"
|
|
|
|
discovery = FakeDiscoveryManager({"a": _app("a", ["svc1"])})
|
|
services = FakeServiceManager()
|
|
|
|
c = InfraController(cfg, discovery=discovery, services=services)
|
|
|
|
monkeypatch.setattr("infra_controller.controller.time.time", lambda: 0.0)
|
|
c.run_once()
|
|
assert services.applied == ["svc1"]
|
|
assert services.stopped == []
|
|
|
|
discovery.set_apps({})
|
|
monkeypatch.setattr("infra_controller.controller.time.time", lambda: 10.0)
|
|
c.run_once()
|
|
assert services.stopped == []
|
|
|
|
monkeypatch.setattr("infra_controller.controller.time.time", lambda: 20.0)
|
|
c.run_once()
|
|
assert services.stopped == ["svc1"]
|
|
|
|
|
|
def test_controller_does_not_stop_service_within_grace_period(tmp_path: Path, monkeypatch):
|
|
cfg = ControllerConfig()
|
|
cfg.services.grace_period_minutes = 1
|
|
cfg.services.state_file = tmp_path / "state.json"
|
|
|
|
discovery = FakeDiscoveryManager({"a": _app("a", ["svc1"])})
|
|
services = FakeServiceManager()
|
|
c = InfraController(cfg, discovery=discovery, services=services)
|
|
|
|
monkeypatch.setattr("infra_controller.controller.time.time", lambda: 0.0)
|
|
c.run_once()
|
|
|
|
discovery.set_apps({})
|
|
monkeypatch.setattr("infra_controller.controller.time.time", lambda: 10.0)
|
|
c.run_once()
|
|
|
|
monkeypatch.setattr("infra_controller.controller.time.time", lambda: 20.0)
|
|
c.run_once()
|
|
assert services.stopped == []
|