mirror of
https://git.hiabuto.net/C3MA/CCMA.git
synced 2026-07-01 19:26:53 +02:00
Marcel Peterkau
225 lines
9.6 KiB
Python
225 lines
9.6 KiB
Python
import json
|
|
import shutil
|
|
from datetime import date
|
|
|
|
import pytest
|
|
|
|
from ccma.rules.loader import RuleLoadError
|
|
from ccma.services.housekeeper import Housekeeper
|
|
from ccma.storage.repository import MemberRepository
|
|
|
|
|
|
def test_store_rule_overrides_builtin_rule_with_same_filename(tmp_path) -> None:
|
|
repository = MemberRepository(tmp_path)
|
|
repository.initialize()
|
|
member = repository.create_member(first_name="Override", last_name="Test")
|
|
(repository.root / "rules" / "birthdate_check.py").write_text(
|
|
"""
|
|
from ccma.rules.api import RuleAction
|
|
|
|
RULE_ID = "birthdate-check"
|
|
|
|
def evaluate(context):
|
|
return [RuleAction(
|
|
key=f"birthdate-check:{context.member.member_id}:override",
|
|
action="task",
|
|
member_id=context.member.member_id,
|
|
payload={
|
|
"code": "override_active",
|
|
"severity": "info",
|
|
"title": "Store-Override aktiv",
|
|
"detail": "Die eingebaute Regel wurde ersetzt.",
|
|
},
|
|
)]
|
|
""".strip(),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
findings = Housekeeper(repository).run(today=date(2026, 6, 21))
|
|
state = json.loads((repository.root / "housekeeper.json").read_text(encoding="utf-8"))
|
|
|
|
assert not (repository.root / "hausmeister.json").exists()
|
|
assert not (repository.root / ".hausmeister.lock").exists()
|
|
assert any(finding.code == "override_active" for finding in findings)
|
|
rule = next(item for item in state["rules"] if item["filename"] == "birthdate_check.py")
|
|
assert rule["source"] == "store-override"
|
|
assert rule["script_hash"].startswith("sha256:")
|
|
assert member.member_id in {item["member_id"] for item in state["items"]}
|
|
|
|
|
|
def test_housekeeper_claim_actions_are_idempotent(tmp_path) -> None:
|
|
repository = MemberRepository(tmp_path)
|
|
repository.initialize()
|
|
member = repository.create_member(first_name="Contribution", last_name="Test", birth_date="1990-01-01")
|
|
member.status = "active"
|
|
member.accepted_at = "2026-04-15"
|
|
member.membership_started_at = "2026-04-15"
|
|
member.payment_frequency = "semiannual"
|
|
repository.save_member(member)
|
|
housekeeper = Housekeeper(repository)
|
|
|
|
housekeeper.run(today=date(2026, 4, 15))
|
|
first_claims = repository.get_contributions(member.member_id).claims
|
|
housekeeper.run(today=date(2026, 4, 15))
|
|
second_claims = repository.get_contributions(member.member_id).claims
|
|
state = json.loads((repository.root / "housekeeper.json").read_text(encoding="utf-8"))
|
|
|
|
assert {claim["claim_key"] for claim in first_claims} == {
|
|
"admission-fee",
|
|
"membership-fee:2026:first-half",
|
|
"membership-fee:2026:second-half",
|
|
}
|
|
assert len(second_claims) == len(first_claims) == 3
|
|
amounts = {claim["claim_key"]: claim["amount"] for claim in first_claims}
|
|
assert amounts["membership-fee:2026:first-half"] == "37.50"
|
|
assert amounts["membership-fee:2026:second-half"] == "75.00"
|
|
assert state["run_counter"] == 2
|
|
assert state["last_completed_run"] == "2026-04-15:000002"
|
|
|
|
|
|
def test_housekeeper_resolves_tasks_not_seen_in_current_run(tmp_path) -> None:
|
|
repository = MemberRepository(tmp_path)
|
|
repository.initialize()
|
|
member = repository.create_member(first_name="Missing", last_name="Birthday")
|
|
housekeeper = Housekeeper(repository)
|
|
|
|
housekeeper.run(today=date(2026, 6, 21))
|
|
member.birth_date = "1990-01-01"
|
|
repository.save_member(member)
|
|
housekeeper.run(today=date(2026, 6, 21))
|
|
state = json.loads((repository.root / "housekeeper.json").read_text(encoding="utf-8"))
|
|
task = next(item for item in state["items"] if item["key"].endswith(":missing"))
|
|
|
|
assert task["status"] == "resolved"
|
|
assert task["first_seen_run"] == "2026-06-21:000001"
|
|
assert task["resolved_run"] == "2026-06-21:000002"
|
|
|
|
|
|
def test_failed_run_does_not_advance_persisted_run_id(tmp_path) -> None:
|
|
repository = MemberRepository(tmp_path)
|
|
repository.initialize()
|
|
repository.create_member(first_name="Failed", last_name="Rule")
|
|
housekeeper = Housekeeper(repository)
|
|
housekeeper.run(today=date(2026, 6, 21))
|
|
state_before = (repository.root / "housekeeper.json").read_bytes()
|
|
(repository.root / "rules" / "broken.py").write_text("this is not python !!!", encoding="utf-8")
|
|
|
|
with pytest.raises(RuleLoadError):
|
|
housekeeper.run(today=date(2026, 6, 21))
|
|
|
|
assert (repository.root / "housekeeper.json").read_bytes() == state_before
|
|
assert not (repository.root / ".housekeeper.lock").exists()
|
|
|
|
|
|
def test_broken_contributions_file_creates_task_without_overwriting_file(tmp_path) -> None:
|
|
repository = MemberRepository(tmp_path)
|
|
repository.initialize()
|
|
member = repository.create_member(first_name="Broken", last_name="Contributions", birth_date="1990-01-01")
|
|
member.status = "active"
|
|
member.accepted_at = "2026-01-01"
|
|
member.membership_started_at = "2026-01-01"
|
|
repository.save_member(member)
|
|
path = repository.members_root / member.member_id / "contributions.json"
|
|
path.write_bytes(b"")
|
|
|
|
findings = Housekeeper(repository).run(today=date(2026, 6, 21))
|
|
|
|
assert path.read_bytes() == b""
|
|
assert [finding.code for finding in findings] == ["invalid_member_record"]
|
|
state = json.loads((repository.root / "housekeeper.json").read_text(encoding="utf-8"))
|
|
task = next(item for item in state["items"] if item["code"] == "invalid_member_record")
|
|
assert task["status"] == "open"
|
|
|
|
|
|
def test_preflight_skips_all_rules_for_broken_member_file(tmp_path) -> None:
|
|
repository = MemberRepository(tmp_path)
|
|
repository.initialize()
|
|
member = repository.create_member(first_name="Broken", last_name="Member")
|
|
member_path = repository.members_root / member.member_id / "member.json"
|
|
contributions_path = repository.members_root / member.member_id / "contributions.json"
|
|
contributions_before = contributions_path.read_bytes()
|
|
member_path.write_text("{", encoding="utf-8")
|
|
|
|
findings = Housekeeper(repository).run(today=date(2026, 6, 21))
|
|
state = json.loads((repository.root / "housekeeper.json").read_text(encoding="utf-8"))
|
|
|
|
assert [finding.code for finding in findings] == ["invalid_member_record"]
|
|
assert len(state["items"]) == 1
|
|
assert contributions_path.read_bytes() == contributions_before
|
|
|
|
|
|
def test_preflight_blocks_rules_for_broken_event_log(tmp_path) -> None:
|
|
repository = MemberRepository(tmp_path)
|
|
repository.initialize()
|
|
member = repository.create_member(first_name="Broken", last_name="Events", birth_date="1990-01-01")
|
|
member.status = "active"
|
|
member.accepted_at = "2026-01-01"
|
|
member.membership_started_at = "2026-01-01"
|
|
repository.save_member(member)
|
|
events_path = repository.members_root / member.member_id / "events.jsonl"
|
|
with events_path.open("a", encoding="utf-8") as handle:
|
|
handle.write("not-json\n")
|
|
|
|
findings = Housekeeper(repository).run(today=date(2026, 6, 21))
|
|
|
|
assert [finding.code for finding in findings] == ["invalid_member_record"]
|
|
assert repository.get_contributions(member.member_id).claims == []
|
|
|
|
|
|
def test_preflight_task_resolves_after_record_is_repaired(tmp_path) -> None:
|
|
repository = MemberRepository(tmp_path)
|
|
repository.initialize()
|
|
member = repository.create_member(first_name="Repair", last_name="Record", birth_date="1990-01-01")
|
|
contributions_path = repository.members_root / member.member_id / "contributions.json"
|
|
original = contributions_path.read_bytes()
|
|
contributions_path.write_bytes(b"")
|
|
housekeeper = Housekeeper(repository)
|
|
housekeeper.run(today=date(2026, 6, 21))
|
|
contributions_path.write_bytes(original)
|
|
|
|
housekeeper.run(today=date(2026, 6, 21))
|
|
state = json.loads((repository.root / "housekeeper.json").read_text(encoding="utf-8"))
|
|
task = next(item for item in state["items"] if item["code"] == "invalid_member_record")
|
|
|
|
assert task["status"] == "resolved"
|
|
assert task["resolved_run"] == "2026-06-21:000002"
|
|
|
|
|
|
def test_housekeeper_task_can_be_deleted_and_returns_on_next_run(tmp_path) -> None:
|
|
repository = MemberRepository(tmp_path)
|
|
repository.initialize()
|
|
repository.create_member(first_name="Manual", last_name="Delete")
|
|
housekeeper = Housekeeper(repository)
|
|
findings = housekeeper.run(today=date(2026, 6, 21))
|
|
finding = next(item for item in findings if item.code == "missing_birth_date")
|
|
|
|
remaining = housekeeper.delete_task(finding.key)
|
|
state_after_delete = json.loads((repository.root / "housekeeper.json").read_text(encoding="utf-8"))
|
|
|
|
assert remaining == []
|
|
assert state_after_delete["run_counter"] == 1
|
|
assert state_after_delete["items"] == []
|
|
|
|
recreated = housekeeper.run(today=date(2026, 6, 21))
|
|
recreated_finding = next(item for item in recreated if item.code == "missing_birth_date")
|
|
state_after_run = json.loads((repository.root / "housekeeper.json").read_text(encoding="utf-8"))
|
|
recreated_task = next(item for item in state_after_run["items"] if item["key"] == recreated_finding.key)
|
|
|
|
assert recreated_finding.key == finding.key
|
|
assert recreated_task["first_seen_run"] == "2026-06-21:000002"
|
|
|
|
|
|
def test_housekeeper_removes_items_for_deleted_member_directory(tmp_path) -> None:
|
|
repository = MemberRepository(tmp_path)
|
|
repository.initialize()
|
|
member = repository.create_member(first_name="Deleted", last_name="Member")
|
|
housekeeper = Housekeeper(repository)
|
|
assert housekeeper.run(today=date(2026, 6, 21))
|
|
shutil.rmtree(repository.members_root / member.member_id)
|
|
|
|
findings = housekeeper.run(today=date(2026, 6, 21))
|
|
state = json.loads((repository.root / "housekeeper.json").read_text(encoding="utf-8"))
|
|
|
|
assert findings == []
|
|
assert not any(item.get("member_id") == member.member_id for item in state["items"])
|