308 lines
11 KiB
Python
308 lines
11 KiB
Python
|
from tests.unit.services.service_helpers import *
|
||
|
|
|
||
|
|
|
||
|
|
def test_wger_sync_user_exec(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,
|
||
|
|
wger_pod_label="app=wger",
|
||
|
|
wger_container="wger",
|
||
|
|
wger_admin_username="admin",
|
||
|
|
wger_admin_password="pw",
|
||
|
|
wger_admin_email="admin@bstein.dev",
|
||
|
|
)
|
||
|
|
monkeypatch.setattr("ariadne.services.wger.settings", dummy)
|
||
|
|
|
||
|
|
calls: list[dict[str, str]] = []
|
||
|
|
|
||
|
|
class DummyExecutor:
|
||
|
|
def exec(self, _cmd, env=None, timeout_sec=None, check=True):
|
||
|
|
calls.append(env or {})
|
||
|
|
return types.SimpleNamespace(stdout="ok", stderr="", exit_code=0, ok=True)
|
||
|
|
|
||
|
|
monkeypatch.setattr("ariadne.services.wger.PodExecutor", lambda *_args, **_kwargs: DummyExecutor())
|
||
|
|
|
||
|
|
svc = WgerService()
|
||
|
|
result = svc.sync_user("alice", "alice@bstein.dev", "pw", wait=True)
|
||
|
|
assert result["status"] == "ok"
|
||
|
|
assert calls[0]["WGER_USERNAME"] == "alice"
|
||
|
|
|
||
|
|
def test_wger_ensure_admin_exec(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,
|
||
|
|
wger_pod_label="app=wger",
|
||
|
|
wger_container="wger",
|
||
|
|
wger_admin_username="admin",
|
||
|
|
wger_admin_password="pw",
|
||
|
|
wger_admin_email="admin@bstein.dev",
|
||
|
|
)
|
||
|
|
monkeypatch.setattr("ariadne.services.wger.settings", dummy)
|
||
|
|
|
||
|
|
calls: list[dict[str, str]] = []
|
||
|
|
|
||
|
|
class DummyExecutor:
|
||
|
|
def exec(self, _cmd, env=None, timeout_sec=None, check=True):
|
||
|
|
calls.append(env or {})
|
||
|
|
return types.SimpleNamespace(stdout="ok", stderr="", exit_code=0, ok=True)
|
||
|
|
|
||
|
|
monkeypatch.setattr("ariadne.services.wger.PodExecutor", lambda *_args, **_kwargs: DummyExecutor())
|
||
|
|
|
||
|
|
svc = WgerService()
|
||
|
|
result = svc.ensure_admin(wait=False)
|
||
|
|
assert result["status"] == "ok"
|
||
|
|
assert calls[0]["WGER_ADMIN_USERNAME"] == "admin"
|
||
|
|
|
||
|
|
def test_wger_sync_users(monkeypatch) -> None:
|
||
|
|
dummy = types.SimpleNamespace(
|
||
|
|
wger_namespace="health",
|
||
|
|
wger_user_sync_wait_timeout_sec=60.0,
|
||
|
|
wger_pod_label="app=wger",
|
||
|
|
wger_container="wger",
|
||
|
|
wger_admin_username="admin",
|
||
|
|
wger_admin_password="pw",
|
||
|
|
wger_admin_email="admin@bstein.dev",
|
||
|
|
mailu_domain="bstein.dev",
|
||
|
|
)
|
||
|
|
monkeypatch.setattr("ariadne.services.wger.settings", dummy)
|
||
|
|
|
||
|
|
calls: list[tuple[str, str, str]] = []
|
||
|
|
|
||
|
|
class DummyAdmin:
|
||
|
|
def ready(self) -> bool:
|
||
|
|
return True
|
||
|
|
|
||
|
|
def iter_users(self, page_size=200, brief=False):
|
||
|
|
return [{"id": "1", "username": "alice", "attributes": {}}]
|
||
|
|
|
||
|
|
def get_user(self, user_id: str):
|
||
|
|
return {"id": user_id, "username": "alice", "attributes": {}}
|
||
|
|
|
||
|
|
def set_user_attribute(self, username: str, key: str, value: str) -> None:
|
||
|
|
calls.append((username, key, value))
|
||
|
|
|
||
|
|
monkeypatch.setattr("ariadne.services.wger.keycloak_admin", DummyAdmin())
|
||
|
|
monkeypatch.setattr(
|
||
|
|
"ariadne.services.wger.mailu.resolve_mailu_email",
|
||
|
|
lambda *_args, **_kwargs: "alice@bstein.dev",
|
||
|
|
)
|
||
|
|
monkeypatch.setattr("ariadne.services.wger.random_password", lambda *_args: "pw")
|
||
|
|
|
||
|
|
def fake_sync_user(self, *_args, **_kwargs):
|
||
|
|
return {"status": "ok", "detail": "ok"}
|
||
|
|
|
||
|
|
monkeypatch.setattr(WgerService, "sync_user", fake_sync_user)
|
||
|
|
|
||
|
|
svc = WgerService()
|
||
|
|
result = svc.sync_users()
|
||
|
|
assert result["status"] == "ok"
|
||
|
|
assert any(key == "wger_password" for _user, key, _value in calls)
|
||
|
|
assert any(key == "wger_password_updated_at" for _user, key, _value in calls)
|
||
|
|
|
||
|
|
def test_wger_check_rotation_marks_rotated(monkeypatch) -> None:
|
||
|
|
dummy = types.SimpleNamespace(
|
||
|
|
wger_namespace="health",
|
||
|
|
wger_user_sync_wait_timeout_sec=60.0,
|
||
|
|
wger_pod_label="app=wger",
|
||
|
|
wger_container="wger",
|
||
|
|
wger_admin_username="admin",
|
||
|
|
wger_admin_password="pw",
|
||
|
|
wger_admin_email="admin@bstein.dev",
|
||
|
|
mailu_domain="bstein.dev",
|
||
|
|
)
|
||
|
|
monkeypatch.setattr("ariadne.services.wger.settings", dummy)
|
||
|
|
|
||
|
|
calls: list[tuple[str, str, str]] = []
|
||
|
|
|
||
|
|
class DummyAdmin:
|
||
|
|
def ready(self) -> bool:
|
||
|
|
return True
|
||
|
|
|
||
|
|
def find_user(self, username: str):
|
||
|
|
return {"id": "1", "username": username, "attributes": {}}
|
||
|
|
|
||
|
|
def get_user(self, user_id: str):
|
||
|
|
return {
|
||
|
|
"id": user_id,
|
||
|
|
"username": "alice",
|
||
|
|
"attributes": {
|
||
|
|
"mailu_email": ["alice@bstein.dev"],
|
||
|
|
"wger_password": ["pw"],
|
||
|
|
},
|
||
|
|
}
|
||
|
|
|
||
|
|
def set_user_attribute(self, username: str, key: str, value: str) -> None:
|
||
|
|
calls.append((username, key, value))
|
||
|
|
|
||
|
|
monkeypatch.setattr("ariadne.services.wger.keycloak_admin", DummyAdmin())
|
||
|
|
|
||
|
|
svc = WgerService()
|
||
|
|
monkeypatch.setattr(svc, "check_password", lambda *_args, **_kwargs: {"status": "mismatch"})
|
||
|
|
|
||
|
|
result = svc.check_rotation_for_user("alice")
|
||
|
|
assert result["status"] == "ok"
|
||
|
|
assert result["rotated"] is True
|
||
|
|
assert any(key == "wger_password_rotated_at" for _user, key, _value in calls)
|
||
|
|
|
||
|
|
def test_wger_sync_marks_rotated(monkeypatch) -> None:
|
||
|
|
dummy = types.SimpleNamespace(
|
||
|
|
wger_namespace="health",
|
||
|
|
wger_user_sync_wait_timeout_sec=60.0,
|
||
|
|
wger_pod_label="app=wger",
|
||
|
|
wger_container="wger",
|
||
|
|
wger_admin_username="admin",
|
||
|
|
wger_admin_password="pw",
|
||
|
|
wger_admin_email="admin@bstein.dev",
|
||
|
|
mailu_domain="bstein.dev",
|
||
|
|
)
|
||
|
|
monkeypatch.setattr("ariadne.services.wger.settings", dummy)
|
||
|
|
|
||
|
|
calls: list[tuple[str, str, str]] = []
|
||
|
|
|
||
|
|
class DummyAdmin:
|
||
|
|
def ready(self) -> bool:
|
||
|
|
return True
|
||
|
|
|
||
|
|
def iter_users(self, page_size=200, brief=False):
|
||
|
|
return [
|
||
|
|
{
|
||
|
|
"id": "1",
|
||
|
|
"username": "alice",
|
||
|
|
"attributes": {
|
||
|
|
"mailu_email": ["alice@bstein.dev"],
|
||
|
|
"wger_password": ["pw"],
|
||
|
|
"wger_password_updated_at": ["2025-01-01T00:00:00Z"],
|
||
|
|
},
|
||
|
|
}
|
||
|
|
]
|
||
|
|
|
||
|
|
def get_user(self, user_id: str):
|
||
|
|
return {
|
||
|
|
"id": user_id,
|
||
|
|
"username": "alice",
|
||
|
|
"attributes": {
|
||
|
|
"mailu_email": ["alice@bstein.dev"],
|
||
|
|
"wger_password": ["pw"],
|
||
|
|
"wger_password_updated_at": ["2025-01-01T00:00:00Z"],
|
||
|
|
},
|
||
|
|
}
|
||
|
|
|
||
|
|
def set_user_attribute(self, username: str, key: str, value: str) -> None:
|
||
|
|
calls.append((username, key, value))
|
||
|
|
|
||
|
|
monkeypatch.setattr("ariadne.services.wger.keycloak_admin", DummyAdmin())
|
||
|
|
monkeypatch.setattr(
|
||
|
|
"ariadne.services.wger.mailu.resolve_mailu_email",
|
||
|
|
lambda *_args, **_kwargs: "alice@bstein.dev",
|
||
|
|
)
|
||
|
|
|
||
|
|
def fake_check(self, *_args, **_kwargs):
|
||
|
|
return {"status": "mismatch", "detail": "mismatch"}
|
||
|
|
|
||
|
|
monkeypatch.setattr(WgerService, "check_password", fake_check)
|
||
|
|
monkeypatch.setattr(WgerService, "sync_user", lambda *_args, **_kwargs: {"status": "ok"})
|
||
|
|
|
||
|
|
svc = WgerService()
|
||
|
|
result = svc.sync_users()
|
||
|
|
assert result["status"] == "ok"
|
||
|
|
assert any(key == "wger_password_rotated_at" for _user, key, _value in calls)
|
||
|
|
|
||
|
|
def test_wger_sync_missing_inputs(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,
|
||
|
|
wger_pod_label="app=wger",
|
||
|
|
wger_container="wger",
|
||
|
|
wger_admin_username="admin",
|
||
|
|
wger_admin_password="pw",
|
||
|
|
wger_admin_email="admin@bstein.dev",
|
||
|
|
)
|
||
|
|
monkeypatch.setattr("ariadne.services.wger.settings", dummy)
|
||
|
|
monkeypatch.setattr("ariadne.services.wger.PodExecutor", lambda *_args, **_kwargs: DummyExecutor())
|
||
|
|
|
||
|
|
svc = WgerService()
|
||
|
|
with pytest.raises(RuntimeError):
|
||
|
|
svc.sync_user("", "email", "pw", wait=True)
|
||
|
|
with pytest.raises(RuntimeError):
|
||
|
|
svc.sync_user("alice", "email", "", wait=True)
|
||
|
|
|
||
|
|
def test_wger_sync_missing_config(monkeypatch) -> None:
|
||
|
|
dummy = types.SimpleNamespace(
|
||
|
|
wger_namespace="",
|
||
|
|
wger_user_sync_cronjob="",
|
||
|
|
wger_admin_cronjob="wger-admin-ensure",
|
||
|
|
wger_user_sync_wait_timeout_sec=60.0,
|
||
|
|
wger_pod_label="app=wger",
|
||
|
|
wger_container="wger",
|
||
|
|
wger_admin_username="admin",
|
||
|
|
wger_admin_password="pw",
|
||
|
|
wger_admin_email="admin@bstein.dev",
|
||
|
|
)
|
||
|
|
monkeypatch.setattr("ariadne.services.wger.settings", dummy)
|
||
|
|
monkeypatch.setattr("ariadne.services.wger.PodExecutor", lambda *_args, **_kwargs: DummyExecutor())
|
||
|
|
svc = WgerService()
|
||
|
|
with pytest.raises(RuntimeError):
|
||
|
|
svc.sync_user("alice", "email", "pw", wait=True)
|
||
|
|
|
||
|
|
def test_wger_ensure_admin(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,
|
||
|
|
wger_pod_label="app=wger",
|
||
|
|
wger_container="wger",
|
||
|
|
wger_admin_username="admin",
|
||
|
|
wger_admin_password="pw",
|
||
|
|
wger_admin_email="admin@bstein.dev",
|
||
|
|
)
|
||
|
|
monkeypatch.setattr("ariadne.services.wger.settings", dummy)
|
||
|
|
monkeypatch.setattr("ariadne.services.wger.PodExecutor", lambda *_args, **_kwargs: DummyExecutor())
|
||
|
|
|
||
|
|
svc = WgerService()
|
||
|
|
result = svc.ensure_admin(wait=True)
|
||
|
|
assert result["status"] == "ok"
|
||
|
|
|
||
|
|
def test_wger_ensure_admin_missing_creds(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,
|
||
|
|
wger_pod_label="app=wger",
|
||
|
|
wger_container="wger",
|
||
|
|
wger_admin_username="",
|
||
|
|
wger_admin_password="",
|
||
|
|
wger_admin_email="",
|
||
|
|
)
|
||
|
|
monkeypatch.setattr("ariadne.services.wger.settings", dummy)
|
||
|
|
monkeypatch.setattr("ariadne.services.wger.PodExecutor", lambda *_args, **_kwargs: DummyExecutor())
|
||
|
|
|
||
|
|
svc = WgerService()
|
||
|
|
result = svc.ensure_admin(wait=True)
|
||
|
|
assert result["status"] == "error"
|
||
|
|
|
||
|
|
def test_wger_ensure_admin_missing_config(monkeypatch) -> None:
|
||
|
|
dummy = types.SimpleNamespace(
|
||
|
|
wger_namespace="",
|
||
|
|
wger_user_sync_cronjob="wger-user-sync",
|
||
|
|
wger_admin_cronjob="",
|
||
|
|
wger_user_sync_wait_timeout_sec=60.0,
|
||
|
|
wger_pod_label="app=wger",
|
||
|
|
wger_container="wger",
|
||
|
|
wger_admin_username="admin",
|
||
|
|
wger_admin_password="pw",
|
||
|
|
wger_admin_email="admin@bstein.dev",
|
||
|
|
)
|
||
|
|
monkeypatch.setattr("ariadne.services.wger.settings", dummy)
|
||
|
|
monkeypatch.setattr("ariadne.services.wger.PodExecutor", lambda *_args, **_kwargs: DummyExecutor())
|
||
|
|
svc = WgerService()
|
||
|
|
with pytest.raises(RuntimeError):
|
||
|
|
svc.ensure_admin(wait=True)
|