test(ariadne): cover nextcloud service edges

tests/unit/services/test_nextcloud_service_edges.py

@@ -0,0 +1,188 @@
+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")