54 lines
2.1 KiB
Python
54 lines
2.1 KiB
Python
from pathlib import Path
|
|
|
|
from idea192_civicpulse_offline_first.service import CivicPulseService
|
|
from idea192_civicpulse_offline_first.sync import deterministic_replay
|
|
|
|
|
|
def test_endorsement_and_audit(tmp_path: Path) -> None:
|
|
service = CivicPulseService(tmp_path / "ledger.sqlite3")
|
|
org = service.create_org("Tenant Action Network")
|
|
proposal = service.create_proposal(org.id, "Endorse rent stabilization", "Coalition policy brief")
|
|
endorsement = service.decide_endorsement(proposal.id, org.id, True, "Aligned with tenant protections")
|
|
|
|
audit = service.audit(proposal.id)
|
|
|
|
assert proposal.status == "approved"
|
|
assert endorsement.signature
|
|
assert any(item["action"] == "create" for item in audit)
|
|
assert any(item["action"] == "decide" for item in service.audit(endorsement.id))
|
|
|
|
|
|
def test_privacy_redacts_public_briefs(tmp_path: Path) -> None:
|
|
service = CivicPulseService(tmp_path / "ledger.sqlite3")
|
|
brief = service.create_policy_brief("Field brief", "Sensitive analysis", ["source-1"])
|
|
|
|
public_view = service.share_brief(brief.id, "public")
|
|
admin_view = service.share_brief(brief.id, "admin")
|
|
|
|
assert public_view["body"] == "[redacted]"
|
|
assert admin_view["body"] == "Sensitive analysis"
|
|
assert public_view["revision"] == 1
|
|
|
|
|
|
def test_sync_merges_ledger_events(tmp_path: Path) -> None:
|
|
left = CivicPulseService(tmp_path / "left.sqlite3")
|
|
right = CivicPulseService(tmp_path / "right.sqlite3")
|
|
|
|
org = left.create_org("Coalition A")
|
|
left.create_task("campaign-1", "Call supporters", org.id, fields={"region": "North"})
|
|
result = left.sync_with(right)
|
|
|
|
assert result.accepted > 0
|
|
assert right.ledger.count() > 0
|
|
|
|
|
|
def test_deterministic_replay_orders_by_timestamp_and_hash(tmp_path: Path) -> None:
|
|
service = CivicPulseService(tmp_path / "ledger.sqlite3")
|
|
org = service.create_org("Organization")
|
|
service.create_task("campaign", "Door knock", org.id)
|
|
|
|
records = service.ledger.all()
|
|
replayed = deterministic_replay(records)
|
|
|
|
assert replayed == sorted(records, key=lambda item: (item.timestamp, item.hash))
|