from __future__ import annotations import types import pytest from ariadne.k8s.exec import ExecError from ariadne.services import nextcloud as nextcloud_module from ariadne.services.nextcloud import NextcloudService from ariadne.services.nextcloud_mail_models import MailSyncCounters from tests.unit.services.service_helpers import DummyExecutor, DummyResponse def _settings(**overrides): values = { "nextcloud_namespace": "nextcloud", "nextcloud_pod_label": "app=nextcloud", "nextcloud_container": "nextcloud", "nextcloud_exec_timeout_sec": 30.0, "nextcloud_db_host": "", "nextcloud_db_port": 5432, "nextcloud_db_name": "nextcloud", "nextcloud_db_user": "nextcloud", "nextcloud_db_password": "", "nextcloud_url": "https://nextcloud.example", "nextcloud_admin_user": "admin", "nextcloud_admin_password": "secret", "mailu_domain": "bstein.dev", "mailu_host": "mail.bstein.dev", } values.update(overrides) return types.SimpleNamespace(**values) def test_nextcloud_exec_fallback_paths(monkeypatch) -> None: monkeypatch.setattr(nextcloud_module, "settings", _settings()) svc = NextcloudService() calls: list[list[str]] = [] def fake_exec(command, **_kwargs): calls.append(command) if len(calls) == 1: raise ExecError("runuser: may not be used by non-root users") return types.SimpleNamespace(stdout="fallback", stderr="", ok=True) monkeypatch.setattr(svc._executor, "exec", fake_exec) assert svc._exec_with_fallback(["runuser"], ["php"]).stdout == "fallback" calls.clear() monkeypatch.setattr( svc._executor, "exec", lambda command, **_kwargs: types.SimpleNamespace(stdout="", stderr="runuser: may not be used by non-root users", ok=False), ) assert svc._exec_with_fallback(["runuser"], ["php"]).ok is False def test_nextcloud_user_creation_and_lookup_errors(monkeypatch) -> None: monkeypatch.setattr(nextcloud_module, "settings", _settings()) svc = NextcloudService() calls: list[tuple[list[str], dict[str, str] | None, bool]] = [] def fake_occ_exec(args, env=None, check=True): calls.append((args, env, check)) if args[:1] == ["user:info"]: return types.SimpleNamespace(ok=False, stdout="not found", stderr="") return types.SimpleNamespace(ok=True, stdout="", stderr="") monkeypatch.setattr(svc, "_occ_exec", fake_occ_exec) monkeypatch.setattr(nextcloud_module, "random_password", lambda _length: "generated") svc._ensure_nextcloud_user("alice", "alice@bstein.dev", "Alice A.") assert calls[-1][0][-1] == "alice" assert calls[-1][1] == {"OC_PASS": "generated"} monkeypatch.setattr(svc, "_occ_exec", lambda *_args, **_kwargs: types.SimpleNamespace(ok=False, stdout="permission denied", stderr="")) with pytest.raises(RuntimeError): svc._ensure_nextcloud_user("alice", "alice@bstein.dev", "") def test_nextcloud_editor_metadata_and_context_helpers(monkeypatch) -> None: monkeypatch.setattr(nextcloud_module, "settings", _settings(nextcloud_db_host="db", nextcloud_db_password="pw")) svc = NextcloudService() executed: list[str] = [] class FakeCursor: def __enter__(self): return self def __exit__(self, exc_type, exc, tb) -> None: return None def execute(self, query: str) -> None: executed.append(query) class FakeConn: def __enter__(self): return self def __exit__(self, exc_type, exc, tb) -> None: return None def cursor(self): return FakeCursor() monkeypatch.setattr(nextcloud_module.psycopg, "connect", lambda **_kwargs: FakeConn()) svc._set_editor_mode_richtext(["1", "bad", "2"]) assert "1,2" in executed[0] updated: list[tuple[str, dict]] = [] monkeypatch.setattr(nextcloud_module.keycloak_admin, "update_user_safe", lambda user_id, payload: updated.append((user_id, payload))) svc._set_user_mail_meta("uid", "alice@bstein.dev", 2) assert updated[0][1]["attributes"]["nextcloud_mail_account_count"] == ["2"] user = {"id": "uid", "username": " alice ", "enabled": True, "attributes": {"mailu_app_password": ["pw"]}} monkeypatch.setattr(nextcloud_module.keycloak_admin, "get_user", lambda _user_id: {**user, "email": "alice@bstein.dev"}) assert svc._normalize_user(user)[0] == "alice" assert svc._normalize_user({"username": ""}) is None assert svc._normalize_user({"username": "svc", "serviceAccountClientId": "client"}) is None counters = MailSyncCounters() monkeypatch.setattr(svc, "_list_mail_accounts", lambda _username: (_ for _ in ()).throw(RuntimeError("bad export"))) assert svc._list_mail_accounts_safe("alice", counters) is None assert counters.failures == 1 def test_nextcloud_mail_account_sync_edges(monkeypatch) -> None: monkeypatch.setattr(nextcloud_module, "settings", _settings()) svc = NextcloudService() counters = MailSyncCounters() assert svc._select_primary_account([("1", "old@bstein.dev"), ("2", "alice@bstein.dev")], "alice@bstein.dev") == ( "2", "alice@bstein.dev", ) assert svc._mailu_accounts([("1", "a@bstein.dev"), ("2", "a@example.com")]) == [("1", "a@bstein.dev")] assert svc._summarize_mail_accounts([("1", "a@bstein.dev")], "missing@bstein.dev") == (1, "a@bstein.dev", ["1"]) monkeypatch.setattr(svc, "_occ", lambda _args: (_ for _ in ()).throw(RuntimeError("occ failed"))) assert svc._update_mail_account("alice", "1", "alice@bstein.dev", "pw") == "occ failed" assert svc._create_mail_account("alice", "alice@bstein.dev", "pw") == "occ failed" assert svc._delete_extra_accounts([("1", "a@bstein.dev"), ("2", "b@bstein.dev")], "1", counters) == 0 assert counters.failures == 1 counters = MailSyncCounters() monkeypatch.setattr(svc, "_update_mail_account", lambda *_args: "nope") assert svc._sync_mail_accounts("alice", "alice@bstein.dev", "pw", [("1", "alice@bstein.dev")], counters) is False monkeypatch.setattr(svc, "_create_mail_account", lambda *_args: "nope") assert svc._sync_mail_accounts("alice", "alice@bstein.dev", "pw", [], counters) is False def test_nextcloud_sync_user_mail_and_external_api(monkeypatch) -> None: monkeypatch.setattr(nextcloud_module, "settings", _settings()) svc = NextcloudService() user = { "id": "uid", "username": "alice", "enabled": True, "attributes": {"mailu_app_password": ["pw"], "mailu_email": ["alice@bstein.dev"]}, } monkeypatch.setattr(nextcloud_module.keycloak_admin, "get_user", lambda _user_id: user) monkeypatch.setattr(svc, "_ensure_nextcloud_user", lambda *_args: None) monkeypatch.setattr(svc, "_list_mail_accounts_safe", lambda *_args: [("1", "alice@bstein.dev")]) monkeypatch.setattr(svc, "_sync_mail_accounts", lambda *_args: True) applied: list[tuple[str, str, list[tuple[str, str]]]] = [] monkeypatch.setattr(svc, "_apply_mail_metadata", lambda *args: applied.append(args)) counters = MailSyncCounters() svc._sync_user_mail(user, counters) assert counters.processed == 1 assert applied class FakeClient: def __init__(self, *args, **kwargs): self.args = args self.kwargs = kwargs def __enter__(self): return self def __exit__(self, exc_type, exc, tb) -> None: return None def request(self, *_args, **_kwargs): return DummyResponse(json_data={"ocs": {"meta": {"status": "ok"}}}) monkeypatch.setattr(nextcloud_module.httpx, "Client", FakeClient) assert svc._external_api("POST", "/apps", {"x": "y"})["ocs"]["meta"]["status"] == "ok" monkeypatch.setattr(nextcloud_module, "settings", _settings(nextcloud_url="")) with pytest.raises(RuntimeError): svc._external_api("GET", "/apps")