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, _parse_mail_export def test_parse_mail_export() -> None: output = "\n".join( [ "Account 12:", " - Name: Alice", " - E-mail: alice@bstein.dev", "Account 13:", " - E-mail: extra@bstein.dev", ] ) accounts = _parse_mail_export(output) assert accounts == [("12", "alice@bstein.dev"), ("13", "extra@bstein.dev")] def test_nextcloud_sync_mail_create(monkeypatch) -> None: dummy_settings = types.SimpleNamespace( nextcloud_namespace="nextcloud", nextcloud_mail_sync_cronjob="nextcloud-mail-sync", nextcloud_mail_sync_wait_timeout_sec=90.0, nextcloud_mail_sync_job_ttl_sec=3600, 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="", mailu_domain="bstein.dev", mailu_host="mail.bstein.dev", ) monkeypatch.setattr(nextcloud_module, "settings", dummy_settings) user = { "id": "uid-1", "username": "alice", "enabled": True, "attributes": { "mailu_app_password": ["pw"], "mailu_email": ["alice@bstein.dev"], }, } monkeypatch.setattr(nextcloud_module.keycloak_admin, "ready", lambda: True) monkeypatch.setattr(nextcloud_module.keycloak_admin, "iter_users", lambda **_kwargs: [user]) monkeypatch.setattr(nextcloud_module.keycloak_admin, "get_user", lambda *_args, **_kwargs: user) occ_calls: list[list[str]] = [] list_calls = [[], [("42", "alice@bstein.dev")]] def fake_occ(args): occ_calls.append(args) return "" svc = NextcloudService() monkeypatch.setattr(svc, "_occ", fake_occ) monkeypatch.setattr(svc, "_ensure_nextcloud_user", lambda *_args, **_kwargs: None) monkeypatch.setattr(svc, "_list_mail_accounts", lambda *_args, **_kwargs: list_calls.pop(0)) monkeypatch.setattr(svc, "_set_editor_mode_richtext", lambda *_args, **_kwargs: None) monkeypatch.setattr(svc, "_set_user_mail_meta", lambda *_args, **_kwargs: None) result = svc.sync_mail() assert result["status"] == "ok" summary = result["summary"] assert summary.created == 1 assert any("mail:account:create" in call[0] for call in occ_calls) def test_nextcloud_sync_mail_keycloak_not_ready(monkeypatch) -> None: dummy_settings = types.SimpleNamespace( nextcloud_namespace="nextcloud", nextcloud_mail_sync_cronjob="nextcloud-mail-sync", nextcloud_mail_sync_wait_timeout_sec=90.0, nextcloud_mail_sync_job_ttl_sec=3600, 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="", mailu_domain="bstein.dev", mailu_host="mail.bstein.dev", ) monkeypatch.setattr(nextcloud_module, "settings", dummy_settings) monkeypatch.setattr(nextcloud_module.keycloak_admin, "ready", lambda: False) svc = NextcloudService() result = svc.sync_mail() assert result["status"] == "error" def test_nextcloud_sync_mail_update_and_delete(monkeypatch) -> None: dummy_settings = types.SimpleNamespace( nextcloud_namespace="nextcloud", nextcloud_mail_sync_cronjob="nextcloud-mail-sync", nextcloud_mail_sync_wait_timeout_sec=90.0, nextcloud_mail_sync_job_ttl_sec=3600, 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="", mailu_domain="bstein.dev", mailu_host="mail.bstein.dev", ) monkeypatch.setattr(nextcloud_module, "settings", dummy_settings) user = { "id": "uid-1", "username": "alice", "enabled": True, "attributes": { "mailu_app_password": ["pw"], "mailu_email": ["alice@bstein.dev"], }, } monkeypatch.setattr(nextcloud_module.keycloak_admin, "ready", lambda: True) monkeypatch.setattr(nextcloud_module.keycloak_admin, "iter_users", lambda **_kwargs: [user]) monkeypatch.setattr(nextcloud_module.keycloak_admin, "get_user", lambda *_args, **_kwargs: user) occ_calls: list[list[str]] = [] list_calls = [ [("1", "alice@bstein.dev"), ("2", "extra@bstein.dev")], [("1", "alice@bstein.dev")], ] def fake_occ(args): occ_calls.append(args) return "" svc = NextcloudService() monkeypatch.setattr(svc, "_occ", fake_occ) monkeypatch.setattr(svc, "_ensure_nextcloud_user", lambda *_args, **_kwargs: None) monkeypatch.setattr(svc, "_list_mail_accounts", lambda *_args, **_kwargs: list_calls.pop(0)) monkeypatch.setattr(svc, "_set_editor_mode_richtext", lambda *_args, **_kwargs: None) monkeypatch.setattr(svc, "_set_user_mail_meta", lambda *_args, **_kwargs: None) result = svc.sync_mail() assert result["status"] == "ok" summary = result["summary"] assert summary.updated == 1 assert summary.deleted == 1 assert any("mail:account:update" in call[0] for call in occ_calls) def test_nextcloud_run_cron(monkeypatch) -> None: dummy_settings = types.SimpleNamespace( nextcloud_namespace="nextcloud", nextcloud_mail_sync_cronjob="nextcloud-mail-sync", nextcloud_mail_sync_wait_timeout_sec=90.0, nextcloud_mail_sync_job_ttl_sec=3600, 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="", mailu_domain="bstein.dev", mailu_host="mail.bstein.dev", ) monkeypatch.setattr(nextcloud_module, "settings", dummy_settings) svc = NextcloudService() def fake_exec(*_args, **_kwargs): return types.SimpleNamespace(stdout="ok", stderr="", exit_code=0, ok=True) monkeypatch.setattr(svc._executor, "exec", fake_exec) result = svc.run_cron() assert result["status"] == "ok" def test_nextcloud_occ_fallback(monkeypatch) -> None: dummy_settings = types.SimpleNamespace( nextcloud_namespace="nextcloud", nextcloud_mail_sync_cronjob="nextcloud-mail-sync", nextcloud_mail_sync_wait_timeout_sec=90.0, nextcloud_mail_sync_job_ttl_sec=3600, 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="", mailu_domain="bstein.dev", mailu_host="mail.bstein.dev", ) monkeypatch.setattr(nextcloud_module, "settings", dummy_settings) svc = NextcloudService() class DummyExec: def __init__(self) -> None: self.calls: list[list[str]] = [] self.fail = True def exec(self, command, **_kwargs): self.calls.append(command) if self.fail: self.fail = False raise ExecError( "pod exec failed exit_code=1 stderr=runuser: may not be used by non-root users" ) return types.SimpleNamespace(stdout="ok", stderr="", exit_code=0, ok=True) executor = DummyExec() svc._executor = executor output = svc._occ(["status"]) assert output == "ok" assert executor.calls[0][0] == "runuser" assert executor.calls[1][0] == "php" def test_nextcloud_ensure_user_created(monkeypatch) -> None: dummy_settings = types.SimpleNamespace( nextcloud_namespace="nextcloud", nextcloud_mail_sync_cronjob="nextcloud-mail-sync", nextcloud_mail_sync_wait_timeout_sec=90.0, nextcloud_mail_sync_job_ttl_sec=3600, 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="", mailu_domain="bstein.dev", mailu_host="mail.bstein.dev", ) monkeypatch.setattr(nextcloud_module, "settings", dummy_settings) monkeypatch.setattr(nextcloud_module, "random_password", lambda *_args, **_kwargs: "pw") 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[0] == "user:info": return types.SimpleNamespace(stdout="User does not exist", stderr="", exit_code=1, ok=False) return types.SimpleNamespace(stdout="created", stderr="", exit_code=0, ok=True) monkeypatch.setattr(svc, "_occ_exec", fake_occ_exec) svc._ensure_nextcloud_user("alice", "alice@bstein.dev", "Alice") assert any(call[0][0] == "user:add" for call in calls) assert calls[-1][1]["OC_PASS"] == "pw" def test_nextcloud_ensure_user_lookup_error(monkeypatch) -> None: dummy_settings = types.SimpleNamespace( nextcloud_namespace="nextcloud", nextcloud_mail_sync_cronjob="nextcloud-mail-sync", nextcloud_mail_sync_wait_timeout_sec=90.0, nextcloud_mail_sync_job_ttl_sec=3600, 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="", mailu_domain="bstein.dev", mailu_host="mail.bstein.dev", ) monkeypatch.setattr(nextcloud_module, "settings", dummy_settings) svc = NextcloudService() def fake_occ_exec(args, env=None, check=True): return types.SimpleNamespace(stdout="permission denied", stderr="", exit_code=1, ok=False) monkeypatch.setattr(svc, "_occ_exec", fake_occ_exec) with pytest.raises(RuntimeError): svc._ensure_nextcloud_user("alice", "alice@bstein.dev", "Alice") def test_nextcloud_run_maintenance(monkeypatch) -> None: dummy_settings = types.SimpleNamespace( nextcloud_namespace="nextcloud", nextcloud_mail_sync_cronjob="nextcloud-mail-sync", nextcloud_mail_sync_wait_timeout_sec=90.0, nextcloud_mail_sync_job_ttl_sec=3600, 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://cloud.bstein.dev", nextcloud_admin_user="admin", nextcloud_admin_password="secret", mailu_domain="bstein.dev", mailu_host="mail.bstein.dev", ) monkeypatch.setattr(nextcloud_module, "settings", dummy_settings) svc = NextcloudService() occ_calls: list[list[str]] = [] monkeypatch.setattr(svc, "_occ", lambda args: occ_calls.append(args) or "") exec_calls = [] def fake_exec(*_args, **_kwargs): exec_calls.append(_args) return types.SimpleNamespace(stdout="ok", stderr="", exit_code=0, ok=True) monkeypatch.setattr(svc._executor, "exec", fake_exec) api_calls: list[tuple[str, str]] = [] def fake_api(method: str, path: str, data=None): api_calls.append((method, path)) if method == "GET": return {"ocs": {"data": [{"id": 1}]}} return {} monkeypatch.setattr(svc, "_external_api", fake_api) result = svc.run_maintenance() assert result["status"] == "ok" assert any(len(call) > 1 and call[1] == "theming" for call in occ_calls) assert any(method == "DELETE" for method, _ in api_calls)