diff --git a/tests/unit/services/test_nextcloud_service_edges.py b/tests/unit/services/test_nextcloud_service_edges.py new file mode 100644 index 0000000..b061934 --- /dev/null +++ b/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")