136 lines
4.5 KiB
Python
136 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
|
|
import httpx
|
|
|
|
from ariadne.services import soteria as soteria_module
|
|
|
|
|
|
class DummyResponse:
|
|
def __init__(self, status_code: int = 200, payload: object | None = None) -> None:
|
|
self.status_code = status_code
|
|
self._payload = payload or {}
|
|
|
|
def raise_for_status(self) -> None:
|
|
if self.status_code >= 400:
|
|
request = httpx.Request("POST", "http://example.test")
|
|
raise httpx.HTTPStatusError("boom", request=request, response=self)
|
|
|
|
def json(self):
|
|
return self._payload
|
|
|
|
|
|
class DummyClient:
|
|
def __init__(self, responses: list[DummyResponse]) -> None:
|
|
self._responses = list(responses)
|
|
self.calls: list[tuple[str, dict[str, object] | None]] = []
|
|
self.kwargs = None
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
def post(self, url: str, json: dict[str, object] | None = None):
|
|
self.calls.append((url, json))
|
|
if not self._responses:
|
|
return DummyResponse(payload={})
|
|
return self._responses.pop(0)
|
|
|
|
|
|
def test_scheduled_backup_posts_for_targets(monkeypatch) -> None:
|
|
dummy = SimpleNamespace(
|
|
soteria_base_url="http://soteria.maintenance.svc.cluster.local",
|
|
soteria_backup_url="",
|
|
soteria_restore_test_url="",
|
|
soteria_timeout_sec=7.5,
|
|
soteria_backup_targets=["jenkins/jenkins", "postgres/postgres-data-postgres-0"],
|
|
soteria_restore_test_targets=[],
|
|
)
|
|
monkeypatch.setattr("ariadne.services.soteria.settings", dummy)
|
|
client = DummyClient(
|
|
[
|
|
DummyResponse(payload={"status": "ok", "backup": "first"}),
|
|
DummyResponse(payload={"status": "ok", "backup": "second"}),
|
|
]
|
|
)
|
|
captured: dict[str, object] = {}
|
|
|
|
def factory(**kwargs):
|
|
captured.update(kwargs)
|
|
return client
|
|
|
|
monkeypatch.setattr(soteria_module.httpx, "Client", factory)
|
|
|
|
summary = soteria_module.SoteriaService().run_scheduled_backups()
|
|
|
|
assert summary.status == "ok"
|
|
assert summary.attempted == 2
|
|
assert summary.succeeded == 2
|
|
assert summary.failed == 0
|
|
assert captured["timeout"] == 7.5
|
|
assert client.calls[0][0].endswith("/v1/backup")
|
|
|
|
|
|
def test_scheduled_backup_skips_when_unconfigured(monkeypatch) -> None:
|
|
dummy = SimpleNamespace(
|
|
soteria_base_url="",
|
|
soteria_backup_url="",
|
|
soteria_restore_test_url="",
|
|
soteria_timeout_sec=7.5,
|
|
soteria_backup_targets=[],
|
|
soteria_restore_test_targets=[],
|
|
)
|
|
monkeypatch.setattr("ariadne.services.soteria.settings", dummy)
|
|
|
|
summary = soteria_module.SoteriaService().run_scheduled_backups()
|
|
|
|
assert summary.status == "skipped"
|
|
assert summary.attempted == 0
|
|
assert summary.detail == "soteria backup url not configured"
|
|
|
|
|
|
def test_scheduled_backup_handles_mixed_results(monkeypatch) -> None:
|
|
dummy = SimpleNamespace(
|
|
soteria_base_url="http://soteria.maintenance.svc.cluster.local",
|
|
soteria_backup_url="",
|
|
soteria_restore_test_url="",
|
|
soteria_timeout_sec=7.5,
|
|
soteria_backup_targets=["bad-target", "gitea/gitea-data"],
|
|
soteria_restore_test_targets=[],
|
|
)
|
|
monkeypatch.setattr("ariadne.services.soteria.settings", dummy)
|
|
client = DummyClient([DummyResponse(status_code=502, payload={"detail": "upstream fail"})])
|
|
monkeypatch.setattr(soteria_module.httpx, "Client", lambda **kwargs: client)
|
|
|
|
summary = soteria_module.SoteriaService().run_scheduled_backups()
|
|
|
|
assert summary.status == "error"
|
|
assert summary.attempted == 1
|
|
assert summary.failed == 1
|
|
assert summary.skipped == 1
|
|
|
|
|
|
def test_scheduled_restore_posts_for_targets(monkeypatch) -> None:
|
|
dummy = SimpleNamespace(
|
|
soteria_base_url="http://soteria.maintenance.svc.cluster.local",
|
|
soteria_backup_url="",
|
|
soteria_restore_test_url="",
|
|
soteria_timeout_sec=7.5,
|
|
soteria_backup_targets=[],
|
|
soteria_restore_test_targets=["jenkins/jenkins=soteria-restore-smoke"],
|
|
)
|
|
monkeypatch.setattr("ariadne.services.soteria.settings", dummy)
|
|
client = DummyClient([DummyResponse(payload={"status": "ok", "volume": "restore-vol"})])
|
|
monkeypatch.setattr(soteria_module.httpx, "Client", lambda **kwargs: client)
|
|
|
|
summary = soteria_module.SoteriaService().run_scheduled_restore_tests()
|
|
|
|
assert summary.status == "ok"
|
|
assert summary.attempted == 1
|
|
assert summary.succeeded == 1
|
|
assert client.calls[0][0].endswith("/v1/restore-test")
|
|
|