diff --git a/ariadne/services/nextcloud.py b/ariadne/services/nextcloud.py index 375c34f..6e9c9e7 100644 --- a/ariadne/services/nextcloud.py +++ b/ariadne/services/nextcloud.py @@ -34,13 +34,7 @@ class NextcloudService: settings.nextcloud_container, ) - def _exec_with_fallback( - self, - primary: list[str], - fallback: list[str], - env: dict[str, str] | None = None, - check: bool = True, - ) -> ExecResult: + def _exec_with_fallback(self, primary: list[str], fallback: list[str], env: dict[str, str] | None = None, check: bool = True) -> ExecResult: try: result = self._executor.exec( primary, @@ -67,12 +61,7 @@ class NextcloudService: ) return result - def _occ_exec( - self, - args: list[str], - env: dict[str, str] | None = None, - check: bool = True, - ) -> ExecResult: + def _occ_exec(self, args: list[str], env: dict[str, str] | None = None, check: bool = True) -> ExecResult: command = ["runuser", "-u", "www-data", "--", "php", "/var/www/html/occ", *args] fallback = ["php", "/var/www/html/occ", *args] return self._exec_with_fallback(command, fallback, env=env, check=check) @@ -81,12 +70,7 @@ class NextcloudService: result = self._occ_exec(args, check=True) return result.stdout - def _ensure_nextcloud_user( - self, - username: str, - mailu_email: str, - display_name: str, - ) -> None: + def _ensure_nextcloud_user(self, username: str, mailu_email: str, display_name: str) -> None: result = self._occ_exec(["user:info", username], check=False) if result.ok: return @@ -191,11 +175,7 @@ class NextcloudService: full_user = user return username_val, user_id, full_user - def _list_mail_accounts_safe( - self, - username: str, - counters: MailSyncCounters, - ) -> list[tuple[str, str]] | None: + def _list_mail_accounts_safe(self, username: str, counters: MailSyncCounters) -> list[tuple[str, str]] | None: try: return self._list_mail_accounts(username) except Exception as exc: @@ -207,11 +187,7 @@ class NextcloudService: ) return None - def _select_primary_account( - self, - mailu_accounts: list[tuple[str, str]], - mailu_email: str, - ) -> tuple[str, str]: + def _select_primary_account(self, mailu_accounts: list[tuple[str, str]], mailu_email: str) -> tuple[str, str]: primary_id = "" primary_email = "" for account_id, account_email in mailu_accounts: @@ -224,13 +200,7 @@ class NextcloudService: break return primary_id, primary_email - def _update_mail_account( - self, - username: str, - primary_id: str, - mailu_email: str, - app_pw: str, - ) -> str | None: + def _update_mail_account(self, username: str, primary_id: str, mailu_email: str, app_pw: str) -> str | None: try: self._occ( [ @@ -295,12 +265,7 @@ class NextcloudService: except Exception as exc: return str(exc) - def _delete_extra_accounts( - self, - mailu_accounts: list[tuple[str, str]], - primary_id: str, - counters: MailSyncCounters, - ) -> int: + def _delete_extra_accounts(self, mailu_accounts: list[tuple[str, str]], primary_id: str, counters: MailSyncCounters) -> int: deleted = 0 for account_id, _account_email in mailu_accounts: if account_id == primary_id: @@ -319,11 +284,7 @@ class NextcloudService: if email.lower().endswith(f"@{settings.mailu_domain.lower()}") ] - def _summarize_mail_accounts( - self, - accounts: list[tuple[str, str]], - mailu_email: str, - ) -> tuple[int, str, list[str]]: + def _summarize_mail_accounts(self, accounts: list[tuple[str, str]], mailu_email: str) -> tuple[int, str, list[str]]: mailu_accounts = self._mailu_accounts(accounts) account_count = len(mailu_accounts) primary_email = "" @@ -337,11 +298,7 @@ class NextcloudService: primary_email = account_email return account_count, primary_email, editor_mode_ids - def _mail_sync_context( - self, - user: dict[str, Any], - counters: MailSyncCounters, - ) -> tuple[str, str, str, str, dict[str, Any]] | None: + def _mail_sync_context(self, user: dict[str, Any], counters: MailSyncCounters) -> tuple[str, str, str, str, dict[str, Any]] | None: normalized = self._normalize_user(user) if not normalized: counters.skipped += 1 @@ -360,14 +317,7 @@ class NextcloudService: pass return username, user_id, mailu_email, app_pw, full_user - def _sync_mail_accounts( - self, - username: str, - mailu_email: str, - app_pw: str, - accounts: list[tuple[str, str]], - counters: MailSyncCounters, - ) -> bool: + def _sync_mail_accounts(self, username: str, mailu_email: str, app_pw: str, accounts: list[tuple[str, str]], counters: MailSyncCounters) -> bool: mailu_accounts = self._mailu_accounts(accounts) if mailu_accounts: primary_id, _primary_email = self._select_primary_account(mailu_accounts, mailu_email) @@ -385,12 +335,7 @@ class NextcloudService: counters.created += 1 return True - def _apply_mail_metadata( - self, - user_id: str, - mailu_email: str, - accounts: list[tuple[str, str]], - ) -> None: + def _apply_mail_metadata(self, user_id: str, mailu_email: str, accounts: list[tuple[str, str]]) -> None: account_count, primary_email, editor_mode_ids = self._summarize_mail_accounts(accounts, mailu_email) self._set_editor_mode_richtext(editor_mode_ids) if user_id: diff --git a/tests/unit/services/test_nextcloud_service_edges.py b/tests/unit/services/test_nextcloud_service_edges.py index b061934..15adef6 100644 --- a/tests/unit/services/test_nextcloud_service_edges.py +++ b/tests/unit/services/test_nextcloud_service_edges.py @@ -186,3 +186,118 @@ def test_nextcloud_sync_user_mail_and_external_api(monkeypatch) -> None: monkeypatch.setattr(nextcloud_module, "settings", _settings(nextcloud_url="")) with pytest.raises(RuntimeError): svc._external_api("GET", "/apps") + + +def test_nextcloud_cron_and_occ_account_listing_edges(monkeypatch) -> None: + monkeypatch.setattr(nextcloud_module, "settings", _settings(nextcloud_namespace="")) + svc = NextcloudService() + with pytest.raises(RuntimeError): + svc.run_cron() + + monkeypatch.setattr(nextcloud_module, "settings", _settings()) + svc = NextcloudService() + monkeypatch.setattr(svc, "_exec_with_fallback", lambda *_args, **_kwargs: (_ for _ in ()).throw(ExecError("boom"))) + assert svc.run_cron() == {"status": "error", "detail": "boom"} + + monkeypatch.setattr(svc, "_exec_with_fallback", lambda *_args, **_kwargs: types.SimpleNamespace(stdout="", stderr="", ok=True)) + assert svc.run_cron() == {"status": "ok"} + + monkeypatch.setattr(svc, "_occ", lambda args: "raw export" if args == ["mail:account:export", "alice"] else "") + monkeypatch.setattr(nextcloud_module, "_parse_mail_export", lambda output: [("1", output)]) + assert svc._list_mail_accounts("alice") == [("1", "raw export")] + + +def test_nextcloud_editor_mode_skip_and_failure_edges(monkeypatch) -> None: + monkeypatch.setattr(nextcloud_module, "settings", _settings()) + svc = NextcloudService() + svc._set_editor_mode_richtext(["not-a-number"]) + svc._set_editor_mode_richtext(["12"]) + + monkeypatch.setattr(nextcloud_module, "settings", _settings(nextcloud_db_host="db", nextcloud_db_password="pw")) + monkeypatch.setattr(nextcloud_module.psycopg, "connect", lambda **_kwargs: (_ for _ in ()).throw(RuntimeError("db down"))) + svc._set_editor_mode_richtext(["12"]) + + +def test_nextcloud_keycloak_metadata_and_normalization_failures(monkeypatch) -> None: + monkeypatch.setattr(nextcloud_module, "settings", _settings()) + svc = NextcloudService() + monkeypatch.setattr(nextcloud_module.keycloak_admin, "update_user_safe", lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("kc down"))) + svc._set_user_mail_meta("uid", "alice@bstein.dev", 1) + + assert svc._normalize_user({"username": "alice", "enabled": False}) is None + user = {"id": "uid", "username": "alice", "enabled": True} + monkeypatch.setattr(nextcloud_module.keycloak_admin, "get_user", lambda _user_id: (_ for _ in ()).throw(RuntimeError("kc down"))) + assert svc._normalize_user(user) == ("alice", "uid", user) + + +def test_nextcloud_mail_sync_context_skip_and_attribute_edges(monkeypatch) -> None: + monkeypatch.setattr(nextcloud_module, "settings", _settings()) + svc = NextcloudService() + counters = MailSyncCounters() + + assert svc._mail_sync_context({"username": ""}, counters) is None + assert svc._mail_sync_context({"username": "alice", "attributes": {}}, counters) is None + assert counters.skipped == 2 + + user = {"username": "alice", "attributes": {"mailu_app_password": ["pw"]}} + monkeypatch.setattr(nextcloud_module, "_resolve_mailu_email", lambda username, _user: f"{username}@bstein.dev") + monkeypatch.setattr(nextcloud_module.keycloak_admin, "set_user_attribute", lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("kc down"))) + assert svc._mail_sync_context(user, counters) == ("alice", "", "alice@bstein.dev", "pw", user) + + +def test_nextcloud_sync_user_mail_short_circuits(monkeypatch) -> None: + monkeypatch.setattr(nextcloud_module, "settings", _settings()) + svc = NextcloudService() + counters = MailSyncCounters() + svc._sync_user_mail({"username": ""}, counters) + assert counters.skipped == 1 + + context = ("alice", "uid", "alice@bstein.dev", "pw", {"username": "alice"}) + monkeypatch.setattr(svc, "_mail_sync_context", lambda *_args: context) + monkeypatch.setattr(svc, "_ensure_nextcloud_user", lambda *_args: (_ for _ in ()).throw(RuntimeError("ensure failed"))) + svc._sync_user_mail({"username": "alice"}, counters) + assert counters.last_error == "nextcloud user ensure failed: ensure failed" + + monkeypatch.setattr(svc, "_ensure_nextcloud_user", lambda *_args: None) + monkeypatch.setattr(svc, "_list_mail_accounts_safe", lambda *_args: None) + before = counters.processed + svc._sync_user_mail({"username": "alice"}, counters) + assert counters.processed == before + + monkeypatch.setattr(svc, "_list_mail_accounts_safe", lambda *_args: [("1", "alice@bstein.dev")]) + monkeypatch.setattr(svc, "_sync_mail_accounts", lambda *_args: False) + svc._sync_user_mail({"username": "alice"}, counters) + assert counters.processed == before + 1 + + account_calls = iter([[("1", "alice@bstein.dev")], None]) + monkeypatch.setattr(svc, "_list_mail_accounts_safe", lambda *_args: next(account_calls)) + 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)) + svc._sync_user_mail({"username": "alice"}, counters) + assert not applied + + +def test_nextcloud_external_api_credentials_and_non_json_edges(monkeypatch) -> None: + monkeypatch.setattr(nextcloud_module, "settings", _settings(nextcloud_admin_password="")) + svc = NextcloudService() + with pytest.raises(RuntimeError): + svc._external_api("GET", "/apps") + + 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() + + monkeypatch.setattr(nextcloud_module, "settings", _settings()) + monkeypatch.setattr(nextcloud_module.httpx, "Client", FakeClient) + assert svc._external_api("GET", "/apps") == {}