239 lines
8.0 KiB
Python
239 lines
8.0 KiB
Python
from __future__ import annotations
|
|
|
|
import types
|
|
|
|
import pytest
|
|
|
|
from ariadne.services.comms import CommsService
|
|
from ariadne.services.firefly import FireflyService
|
|
from ariadne.services.mailu import MailuService
|
|
from ariadne.services.nextcloud import NextcloudService
|
|
from ariadne.services.vault import VaultService
|
|
from ariadne.services.wger import WgerService
|
|
from ariadne.services.vaultwarden import VaultwardenService
|
|
|
|
|
|
class DummySpawner:
|
|
def __init__(self, namespace, cronjob):
|
|
self.namespace = namespace
|
|
self.cronjob = cronjob
|
|
self.calls = []
|
|
|
|
def trigger_and_wait(self, label_suffix, env_overrides, timeout_sec, job_ttl_seconds=None):
|
|
self.calls.append((label_suffix, env_overrides, timeout_sec, job_ttl_seconds))
|
|
return {"job": "test", "status": "ok"}
|
|
|
|
def trigger(self, label_suffix, env_overrides, job_ttl_seconds=None):
|
|
self.calls.append((label_suffix, env_overrides, job_ttl_seconds))
|
|
return {"job": "test", "status": "queued"}
|
|
|
|
|
|
class DummyClient:
|
|
def __init__(self):
|
|
self.url = ""
|
|
self.payload = None
|
|
self.status_code = 200
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
def post(self, url, json=None):
|
|
self.url = url
|
|
self.payload = json
|
|
return types.SimpleNamespace(status_code=self.status_code)
|
|
|
|
|
|
class DummyResponse:
|
|
def __init__(self, status_code=200, text=""):
|
|
self.status_code = status_code
|
|
self.text = text
|
|
|
|
def raise_for_status(self):
|
|
return None
|
|
|
|
|
|
class DummyVaultwardenClient:
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
def post(self, path, json=None, data=None):
|
|
self.calls.append((path, json, data))
|
|
return DummyResponse(200, "")
|
|
|
|
def close(self):
|
|
return None
|
|
|
|
|
|
def test_nextcloud_sync_mail_builds_env(monkeypatch) -> None:
|
|
dummy = types.SimpleNamespace(
|
|
nextcloud_namespace="nextcloud",
|
|
nextcloud_mail_sync_cronjob="nextcloud-mail-sync",
|
|
nextcloud_mail_sync_wait_timeout_sec=90.0,
|
|
nextcloud_mail_sync_job_ttl_sec=3600,
|
|
)
|
|
monkeypatch.setattr("ariadne.services.nextcloud.settings", dummy)
|
|
monkeypatch.setattr("ariadne.services.nextcloud.JobSpawner", lambda ns, cj: DummySpawner(ns, cj))
|
|
|
|
svc = NextcloudService()
|
|
result = svc.sync_mail("alice", wait=True)
|
|
|
|
assert result["status"] == "ok"
|
|
spawner = svc._spawner
|
|
assert spawner.calls
|
|
label, env, timeout, ttl = spawner.calls[0]
|
|
assert label == "alice"
|
|
assert {item["name"]: item["value"] for item in env}["ONLY_USERNAME"] == "alice"
|
|
assert ttl == 3600
|
|
|
|
|
|
def test_wger_sync_user_env(monkeypatch) -> None:
|
|
dummy = types.SimpleNamespace(
|
|
wger_namespace="health",
|
|
wger_user_sync_cronjob="wger-user-sync",
|
|
wger_admin_cronjob="wger-admin-ensure",
|
|
wger_user_sync_wait_timeout_sec=60.0,
|
|
)
|
|
monkeypatch.setattr("ariadne.services.wger.settings", dummy)
|
|
monkeypatch.setattr("ariadne.services.wger.JobSpawner", lambda ns, cj: DummySpawner(ns, cj))
|
|
|
|
svc = WgerService()
|
|
result = svc.sync_user("alice", "alice@bstein.dev", "pw", wait=True)
|
|
|
|
assert result["status"] == "ok"
|
|
user_spawner = svc._user_spawner
|
|
label, env, _, _ = user_spawner.calls[0]
|
|
assert label == "alice"
|
|
env_map = {item["name"]: item["value"] for item in env}
|
|
assert env_map["WGER_USERNAME"] == "alice"
|
|
assert env_map["WGER_EMAIL"] == "alice@bstein.dev"
|
|
|
|
|
|
def test_firefly_sync_user_env(monkeypatch) -> None:
|
|
dummy = types.SimpleNamespace(
|
|
firefly_namespace="finance",
|
|
firefly_user_sync_cronjob="firefly-user-sync",
|
|
firefly_user_sync_wait_timeout_sec=60.0,
|
|
)
|
|
monkeypatch.setattr("ariadne.services.firefly.settings", dummy)
|
|
monkeypatch.setattr("ariadne.services.firefly.JobSpawner", lambda ns, cj: DummySpawner(ns, cj))
|
|
|
|
svc = FireflyService()
|
|
result = svc.sync_user("alice@bstein.dev", "pw", wait=True)
|
|
|
|
assert result["status"] == "ok"
|
|
spawner = svc._spawner
|
|
label, env, _, _ = spawner.calls[0]
|
|
assert label == "alice"
|
|
env_map = {item["name"]: item["value"] for item in env}
|
|
assert env_map["FIREFLY_USER_EMAIL"] == "alice@bstein.dev"
|
|
|
|
|
|
def test_vault_sync_jobs(monkeypatch) -> None:
|
|
dummy = types.SimpleNamespace(
|
|
vault_namespace="vault",
|
|
vault_k8s_auth_cronjob="vault-k8s-auth-config",
|
|
vault_oidc_cronjob="vault-oidc-config",
|
|
vault_job_wait_timeout_sec=120.0,
|
|
)
|
|
monkeypatch.setattr("ariadne.services.vault.settings", dummy)
|
|
monkeypatch.setattr("ariadne.services.vault.JobSpawner", lambda ns, cj: DummySpawner(ns, cj))
|
|
|
|
svc = VaultService()
|
|
result = svc.sync_k8s_auth(wait=True)
|
|
|
|
assert result["status"] == "ok"
|
|
spawner = svc._k8s_auth_spawner
|
|
label, _, timeout, _ = spawner.calls[0]
|
|
assert label == "k8s-auth"
|
|
assert timeout == 120.0
|
|
|
|
|
|
def test_comms_jobs(monkeypatch) -> None:
|
|
dummy = types.SimpleNamespace(
|
|
comms_namespace="comms",
|
|
comms_guest_name_cronjob="guest-name-randomizer",
|
|
comms_pin_invite_cronjob="pin-othrys-invite",
|
|
comms_reset_room_cronjob="othrys-room-reset",
|
|
comms_seed_room_cronjob="seed-othrys-room",
|
|
comms_job_wait_timeout_sec=60.0,
|
|
)
|
|
monkeypatch.setattr("ariadne.services.comms.settings", dummy)
|
|
monkeypatch.setattr("ariadne.services.comms.JobSpawner", lambda ns, cj: DummySpawner(ns, cj))
|
|
|
|
svc = CommsService()
|
|
result = svc.run_guest_name_randomizer(wait=True)
|
|
|
|
assert result["status"] == "ok"
|
|
spawner = svc._guest_name_spawner
|
|
label, _, timeout, _ = spawner.calls[0]
|
|
assert label == "guest-name"
|
|
assert timeout == 60.0
|
|
|
|
|
|
def test_mailu_sync_includes_force(monkeypatch) -> None:
|
|
dummy_settings = types.SimpleNamespace(
|
|
mailu_sync_url="http://mailu",
|
|
mailu_sync_wait_timeout_sec=10.0,
|
|
mailu_db_host="localhost",
|
|
mailu_db_port=5432,
|
|
mailu_db_name="mailu",
|
|
mailu_db_user="mailu",
|
|
mailu_db_password="secret",
|
|
)
|
|
client = DummyClient()
|
|
|
|
monkeypatch.setattr("ariadne.services.mailu.settings", dummy_settings)
|
|
monkeypatch.setattr("ariadne.services.mailu.httpx.Client", lambda *args, **kwargs: client)
|
|
|
|
svc = MailuService()
|
|
svc.sync("provision", force=True)
|
|
|
|
assert client.url == "http://mailu"
|
|
assert client.payload["wait"] is True
|
|
assert client.payload["force"] is True
|
|
|
|
|
|
def test_vaultwarden_invite_uses_admin_session(monkeypatch) -> None:
|
|
dummy_settings = types.SimpleNamespace(
|
|
vaultwarden_namespace="vaultwarden",
|
|
vaultwarden_admin_secret_name="vaultwarden-admin",
|
|
vaultwarden_admin_secret_key="ADMIN_TOKEN",
|
|
vaultwarden_admin_rate_limit_backoff_sec=600,
|
|
vaultwarden_admin_session_ttl_sec=900,
|
|
vaultwarden_service_host="vaultwarden-service.vaultwarden.svc.cluster.local",
|
|
vaultwarden_pod_label="app=vaultwarden",
|
|
vaultwarden_pod_port=80,
|
|
)
|
|
client = DummyVaultwardenClient()
|
|
|
|
monkeypatch.setattr("ariadne.services.vaultwarden.settings", dummy_settings)
|
|
monkeypatch.setattr("ariadne.services.vaultwarden.get_secret_value", lambda *args, **kwargs: "token")
|
|
monkeypatch.setattr("ariadne.services.vaultwarden.httpx.Client", lambda *args, **kwargs: client)
|
|
monkeypatch.setattr(
|
|
"ariadne.services.vaultwarden.VaultwardenService._find_pod_ip",
|
|
staticmethod(lambda *args, **kwargs: "127.0.0.1"),
|
|
)
|
|
|
|
svc = VaultwardenService()
|
|
result = svc.invite_user("alice@bstein.dev")
|
|
|
|
assert result.ok is True
|
|
assert any(call[0] == "/admin" for call in client.calls)
|
|
assert any(call[0] == "/admin/invite" for call in client.calls)
|
|
|
|
|
|
def test_nextcloud_missing_config(monkeypatch) -> None:
|
|
dummy = types.SimpleNamespace(
|
|
nextcloud_namespace="",
|
|
nextcloud_mail_sync_cronjob="",
|
|
nextcloud_mail_sync_wait_timeout_sec=90.0,
|
|
nextcloud_mail_sync_job_ttl_sec=3600,
|
|
)
|
|
monkeypatch.setattr("ariadne.services.nextcloud.settings", dummy)
|
|
svc = NextcloudService()
|
|
with pytest.raises(RuntimeError):
|
|
svc.sync_mail("alice")
|