ariadne/tests/unit/services/test_firefly_service.py

from tests.unit.services.service_helpers import *


def test_firefly_check_rotation_marks_rotated(monkeypatch) -> None:
    dummy = types.SimpleNamespace(
        firefly_namespace="finance",
        firefly_pod_label="app=firefly",
        firefly_container="firefly",
        firefly_user_sync_wait_timeout_sec=60.0,
        mailu_domain="bstein.dev",
    )
    monkeypatch.setattr("ariadne.services.firefly.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"],
                    "firefly_password": ["pw"],
                },
            }

        def set_user_attribute(self, username: str, key: str, value: str) -> None:
            calls.append((username, key, value))

    monkeypatch.setattr("ariadne.services.firefly.keycloak_admin", DummyAdmin())

    svc = FireflyService()
    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 == "firefly_password_rotated_at" for _user, key, _value in calls)

def test_firefly_sync_user_exec(monkeypatch) -> None:
    dummy = types.SimpleNamespace(
        firefly_namespace="finance",
        firefly_user_sync_cronjob="firefly-user-sync",
        firefly_user_sync_wait_timeout_sec=60.0,
        firefly_pod_label="app=firefly",
        firefly_container="firefly",
        firefly_cron_base_url="http://firefly/cron",
        firefly_cron_token="token",
        firefly_cron_timeout_sec=10.0,
    )
    monkeypatch.setattr("ariadne.services.firefly.settings", dummy)

    class DummyExecutor:
        def exec(self, _cmd, env=None, timeout_sec=None, check=True):
            return types.SimpleNamespace(stdout="ok", stderr="", exit_code=0, ok=True)

    monkeypatch.setattr("ariadne.services.firefly.PodExecutor", lambda *_args, **_kwargs: DummyExecutor())

    svc = FireflyService()
    result = svc.sync_user("alice@bstein.dev", "pw", wait=True)
    assert result["status"] == "ok"

def test_firefly_sync_missing_inputs(monkeypatch) -> None:
    dummy = types.SimpleNamespace(
        firefly_namespace="finance",
        firefly_user_sync_cronjob="firefly-user-sync",
        firefly_user_sync_wait_timeout_sec=60.0,
        firefly_pod_label="app=firefly",
        firefly_container="firefly",
        firefly_cron_base_url="http://firefly/cron",
        firefly_cron_token="token",
        firefly_cron_timeout_sec=10.0,
    )
    monkeypatch.setattr("ariadne.services.firefly.settings", dummy)
    monkeypatch.setattr("ariadne.services.firefly.PodExecutor", lambda *_args, **_kwargs: None)

    svc = FireflyService()
    with pytest.raises(RuntimeError):
        svc.sync_user("", "pw", wait=True)
    with pytest.raises(RuntimeError):
        svc.sync_user("alice@bstein.dev", "", wait=True)

def test_firefly_sync_missing_config(monkeypatch) -> None:
    dummy = types.SimpleNamespace(
        firefly_namespace="",
        firefly_user_sync_cronjob="",
        firefly_user_sync_wait_timeout_sec=60.0,
        firefly_pod_label="app=firefly",
        firefly_container="firefly",
        firefly_cron_base_url="http://firefly/cron",
        firefly_cron_token="token",
        firefly_cron_timeout_sec=10.0,
    )
    monkeypatch.setattr("ariadne.services.firefly.settings", dummy)
    svc = FireflyService()
    with pytest.raises(RuntimeError):
        svc.sync_user("alice@bstein.dev", "pw", wait=True)

def test_firefly_run_cron(monkeypatch) -> None:
    dummy = types.SimpleNamespace(
        firefly_namespace="finance",
        firefly_user_sync_cronjob="firefly-user-sync",
        firefly_user_sync_wait_timeout_sec=60.0,
        firefly_pod_label="app=firefly",
        firefly_container="firefly",
        firefly_cron_base_url="http://firefly/cron",
        firefly_cron_token="token",
        firefly_cron_timeout_sec=10.0,
    )
    monkeypatch.setattr("ariadne.services.firefly.settings", dummy)
    monkeypatch.setattr("ariadne.services.firefly.PodExecutor", lambda *_args, **_kwargs: DummyExecutor())

    class DummyHTTP:
        def __init__(self):
            self.calls = []

        def __enter__(self):
            return self

        def __exit__(self, exc_type, exc, tb):
            return False

        def get(self, url):
            self.calls.append(url)
            return types.SimpleNamespace(status_code=200)

    monkeypatch.setattr("ariadne.services.firefly.httpx.Client", lambda *args, **kwargs: DummyHTTP())

    svc = FireflyService()
    result = svc.run_cron()
    assert result["status"] == "ok"

def test_firefly_sync_users(monkeypatch) -> None:
    dummy = types.SimpleNamespace(
        firefly_namespace="finance",
        firefly_user_sync_wait_timeout_sec=60.0,
        firefly_pod_label="app=firefly",
        firefly_container="firefly",
        firefly_cron_base_url="http://firefly/cron",
        firefly_cron_token="token",
        firefly_cron_timeout_sec=10.0,
        mailu_domain="bstein.dev",
    )
    monkeypatch.setattr("ariadne.services.firefly.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.firefly.keycloak_admin", DummyAdmin())
    monkeypatch.setattr(
        "ariadne.services.firefly.mailu.resolve_mailu_email",
        lambda *_args, **_kwargs: "alice@bstein.dev",
    )
    monkeypatch.setattr("ariadne.services.firefly.random_password", lambda *_args: "pw")

    def fake_sync_user(self, *_args, **_kwargs):
        return {"status": "ok", "detail": "ok"}

    monkeypatch.setattr(FireflyService, "sync_user", fake_sync_user)

    svc = FireflyService()
    result = svc.sync_users()
    assert result["status"] == "ok"
    assert any(key == "firefly_password" for _user, key, _value in calls)
    assert any(key == "firefly_password_updated_at" for _user, key, _value in calls)

def test_firefly_sync_marks_rotated(monkeypatch) -> None:
    dummy = types.SimpleNamespace(
        firefly_namespace="finance",
        firefly_user_sync_wait_timeout_sec=60.0,
        firefly_pod_label="app=firefly",
        firefly_container="firefly",
        firefly_cron_base_url="http://firefly/cron",
        firefly_cron_token="token",
        firefly_cron_timeout_sec=10.0,
        mailu_domain="bstein.dev",
    )
    monkeypatch.setattr("ariadne.services.firefly.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"],
                        "firefly_password": ["pw"],
                        "firefly_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"],
                    "firefly_password": ["pw"],
                    "firefly_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.firefly.keycloak_admin", DummyAdmin())
    monkeypatch.setattr(
        "ariadne.services.firefly.mailu.resolve_mailu_email",
        lambda *_args, **_kwargs: "alice@bstein.dev",
    )

    def fake_check(self, *_args, **_kwargs):
        return {"status": "mismatch", "detail": "mismatch"}

    monkeypatch.setattr(FireflyService, "check_password", fake_check)
    monkeypatch.setattr(FireflyService, "sync_user", lambda *_args, **_kwargs: {"status": "ok"})

    svc = FireflyService()
    result = svc.sync_users()
    assert result["status"] == "ok"
    assert any(key == "firefly_password_rotated_at" for _user, key, _value in calls)