CCMA/tests/test_read_only_store.py

import os
import stat
from datetime import date

import pytest

from ccma.domain.models import ContributionData
from ccma.services.documents import DocumentService
from ccma.services.housekeeper import Housekeeper
from ccma.services.reminder_mail import generate_and_send_reminder_mail
from ccma.storage.repository import MemberRepository, ReadOnlyStoreError

# Root ignores the permission bits this test relies on, so the read-only mount it
# stands in for cannot be simulated there.
pytestmark = pytest.mark.skipif(
    hasattr(os, "geteuid") and os.geteuid() == 0, reason="root bypasses file permissions"
)


def _set_writable(root, writable: bool) -> None:
    mode_add = stat.S_IWUSR
    for path in sorted(root.rglob("*"), reverse=True) + [root]:
        current = stat.S_IMODE(path.stat().st_mode)
        path.chmod(current | mode_add if writable else current & ~mode_add)


@pytest.fixture
def read_only_store(tmp_path):
    root = tmp_path / "store"
    repository = MemberRepository(root)
    repository.initialize()
    member = repository.create_member(
        first_name="Ada", last_name="Lovelace", birth_date="1990-01-01"
    )
    repository.save_contributions(
        member.member_id,
        ContributionData(
            claims=[
                {
                    "claim_id": "claim-1",
                    "claim_key": "overdue",
                    "title": "Offene Forderung",
                    "amount": "100.00",
                    "due_date": "2026-01-31",
                    "status": "open",
                }
            ]
        ),
    )
    _set_writable(root, False)
    yield MemberRepository(root), member
    _set_writable(root, True)


def test_a_writable_store_is_not_reported_as_read_only(tmp_path):
    repository = MemberRepository(tmp_path / "store")
    repository.initialize()

    assert repository.read_only is False


def test_store_directory_that_does_not_exist_yet_counts_as_writable(tmp_path):
    assert MemberRepository(tmp_path / "not" / "created" / "yet").read_only is False


def test_read_only_store_opens_and_stays_readable(read_only_store):
    repository, member = read_only_store

    assert repository.read_only is True
    repository.initialize()
    assert repository.validate() == []
    assert [item.member_id for item in repository.list_members()] == [member.member_id]
    assert repository.get_member(member.member_id).first_name == "Ada"
    assert repository.get_contributions(member.member_id).claims
    assert repository.get_events(member.member_id)
    assert repository.get_configuration()["organization"]["name"]


def test_every_write_is_refused_with_one_clear_message(read_only_store):
    repository, member = read_only_store
    stored = repository.get_member(member.member_id)

    for action in (
        lambda: repository.save_member(stored),
        lambda: repository.create_member(first_name="Grace", last_name="Hopper"),
        lambda: repository.create_manual_claim(
            member.member_id, title="Beitrag", amount="10.00", due_date="2026-09-30"
        ),
        lambda: repository.save_contributions(member.member_id, ContributionData()),
        lambda: repository.append_event(
            member.member_id, event_type="comment", summary="Test", actor_type="user"
        ),
        lambda: repository.save_organization({"name": "Test", "iban": "", "bic": ""}),
        lambda: repository.create_asset(label="Beamer"),
    ):
        with pytest.raises(ReadOnlyStoreError, match="schreibgeschützt"):
            action()


def test_housekeeper_refuses_to_run_instead_of_failing_on_its_lock(read_only_store):
    repository, _member = read_only_store

    with pytest.raises(ReadOnlyStoreError, match="schreibgeschützt"):
        Housekeeper(repository).run(today=date(2026, 2, 10))
    with pytest.raises(ReadOnlyStoreError, match="schreibgeschützt"):
        Housekeeper(repository).delete_task("some-key")
    assert not (repository.root / ".housekeeper.lock").exists()


def test_documents_and_mails_refuse_before_touching_the_member_file(read_only_store, tmp_path):
    repository, member = read_only_store

    service = DocumentService(repository)
    template = next(item for item in service.list_templates() if item.name == "Mitglied")
    with pytest.raises(ReadOnlyStoreError, match="schreibgeschützt"):
        service.generate(template, member.member_id, output_name="Test")
    with pytest.raises(ReadOnlyStoreError, match="schreibgeschützt"):
        generate_and_send_reminder_mail(
            repository,
            member.member_id,
            "claim-1",
            "reminder-1",
            delivery_mode="local",
            output_path=tmp_path / "Mahnung.eml",
            sender_name="Verwaltung",
            sender_email="verwaltung@example.org",
            signature="Der Vorstand",
        )


def test_uninitialized_read_only_store_reports_why_it_cannot_be_opened(tmp_path):
    root = tmp_path / "empty-store"
    root.mkdir()
    root.chmod(stat.S_IRUSR | stat.S_IXUSR)
    try:
        with pytest.raises(ReadOnlyStoreError, match="repository.json"):
            MemberRepository(root).initialize()
    finally:
        root.chmod(stat.S_IRWXU)


def test_remounting_writable_is_picked_up_without_a_restart(read_only_store):
    repository, _member = read_only_store
    assert repository.read_only is True

    _set_writable(repository.root, True)

    assert repository.read_only is True, "the cached answer must stay put until refreshed"
    assert repository.refresh_read_only() is False
    repository.append_event(
        _member.member_id,
        event_type="comment",
        summary="Wieder beschreibbar",
        actor_type="user",
    )


def test_startup_sequence_completes_without_the_housekeeper(read_only_store):
    """Mirrors what the splash screen's worker thread does -- this is the sequence
    that used to abort the whole start with a PermissionError."""
    repository, member = read_only_store

    repository.initialize()
    errors = repository.validate()
    members = repository.list_members()
    findings = [] if repository.read_only else Housekeeper(repository).run()

    assert errors == []
    assert [item.member_id for item in members] == [member.member_id]
    assert findings == []
    assert repository.member_count() == 1


def test_mail_templates_fall_back_to_the_shipped_defaults(read_only_store):
    """A store that predates mail templates has no templates/mail/ directory, and a
    read-only one cannot get it -- the shipped default has to carry the session."""
    from ccma.domain.mail_templates import default_mail_template

    repository, _member = read_only_store
    _set_writable(repository.root, True)
    for path in repository.mail_templates_root.glob("*.txt"):
        path.unlink()
    _set_writable(repository.root, False)

    assert repository.get_mail_template("welcome") == default_mail_template("welcome")
    assert not list(repository.mail_templates_root.glob("*.txt")), "nothing was written"
    with pytest.raises(ReadOnlyStoreError, match="schreibgeschützt"):
        repository.save_mail_template("welcome", subject="Moin", body="Kurz.")