304 lines
13 KiB
Python
304 lines
13 KiB
Python
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")
|
|
|
|
|
|
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") == {}
|