mirror of
https://git.hiabuto.net/C3MA/CCMA.git
synced 2026-07-01 11:14:52 +02:00
Marcel Peterkau
109 lines
4.3 KiB
Python
109 lines
4.3 KiB
Python
import json
|
|
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 / "hausmeister.json").read_text(encoding="utf-8"))
|
|
|
|
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 / "hausmeister.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 / "hausmeister.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 / "hausmeister.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 / "hausmeister.json").read_bytes() == state_before
|
|
assert not (repository.root / ".hausmeister.lock").exists()
|