83 lines
3.2 KiB
Python
83 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from ariadne.services import metis_token_sync as module
|
|
|
|
|
|
def _settings() -> SimpleNamespace:
|
|
return SimpleNamespace(
|
|
metis_token_sync_namespace="maintenance",
|
|
metis_token_sync_service_account="metis-token-sync",
|
|
metis_token_sync_node_name="titan-0a",
|
|
metis_token_sync_image="hashicorp/vault:1.17.6",
|
|
metis_token_sync_job_ttl_sec=1800,
|
|
metis_token_sync_wait_timeout_sec=10.0,
|
|
metis_token_sync_vault_addr="http://vault.vault.svc.cluster.local:8200",
|
|
metis_token_sync_vault_k8s_role="maintenance-metis-token-sync",
|
|
)
|
|
|
|
|
|
def test_payload_matches_expected_contract(monkeypatch) -> None:
|
|
monkeypatch.setattr(module, "settings", _settings())
|
|
payload = module.MetisTokenSyncService()._job_payload("sync-job")
|
|
|
|
assert payload["metadata"]["namespace"] == "maintenance"
|
|
assert payload["metadata"]["labels"]["atlas.bstein.dev/trigger"] == "ariadne"
|
|
spec = payload["spec"]["template"]["spec"]
|
|
assert spec["serviceAccountName"] == "metis-token-sync"
|
|
assert spec["nodeName"] == "titan-0a"
|
|
assert spec["containers"][0]["image"] == "hashicorp/vault:1.17.6"
|
|
assert spec["containers"][0]["volumeMounts"][0]["mountPath"] == "/host/var/lib/rancher/k3s/server"
|
|
|
|
|
|
def test_run_wait_success(monkeypatch) -> None:
|
|
monkeypatch.setattr(module, "settings", _settings())
|
|
monkeypatch.setattr(module.time, "time", lambda: 1710000000)
|
|
|
|
posted: dict[str, object] = {}
|
|
|
|
def fake_post(path: str, payload: dict[str, object]) -> dict[str, object]:
|
|
posted["path"] = path
|
|
posted["payload"] = payload
|
|
return {"metadata": {"name": "sync-job-1"}}
|
|
|
|
calls = iter(
|
|
[
|
|
{"status": {"active": 1}},
|
|
{"status": {"succeeded": 1}},
|
|
]
|
|
)
|
|
|
|
monkeypatch.setattr(module, "post_json", fake_post)
|
|
monkeypatch.setattr(module, "get_json", lambda _path: next(calls))
|
|
monkeypatch.setattr(module.time, "sleep", lambda _seconds: None)
|
|
|
|
result = module.MetisTokenSyncService().run(wait=True)
|
|
|
|
assert posted["path"] == "/apis/batch/v1/namespaces/maintenance/jobs"
|
|
assert result == {"job": "sync-job-1", "status": "ok"}
|
|
|
|
|
|
def test_run_wait_failure_raises(monkeypatch) -> None:
|
|
monkeypatch.setattr(module, "settings", _settings())
|
|
monkeypatch.setattr(module.time, "time", lambda: 1710000000)
|
|
monkeypatch.setattr(module, "post_json", lambda _path, _payload: {"metadata": {"name": "sync-job-2"}})
|
|
monkeypatch.setattr(module, "get_json", lambda _path: {"status": {"failed": 1}})
|
|
monkeypatch.setattr(module.time, "sleep", lambda _seconds: None)
|
|
|
|
with pytest.raises(RuntimeError, match="metis token sync job sync-job-2 error"):
|
|
module.MetisTokenSyncService().run(wait=True)
|
|
|
|
|
|
def test_run_without_wait_queues(monkeypatch) -> None:
|
|
monkeypatch.setattr(module, "settings", _settings())
|
|
monkeypatch.setattr(module.time, "time", lambda: 1710000000)
|
|
monkeypatch.setattr(module, "post_json", lambda _path, _payload: {"metadata": {"name": "sync-job-3"}})
|
|
|
|
result = module.MetisTokenSyncService().run(wait=False)
|
|
|
|
assert result == {"job": "sync-job-3", "status": "queued"}
|