| 1234567891011121314151617181920212223242526 |
- from app.runner.ledger import InMemoryTaskLedger
- def test_a_token_can_be_claimed_only_once_and_keeps_bounded_outcome():
- ledger = InMemoryTaskLedger()
- binding = {
- "task_uid": "task-1",
- "node_id": "write_orders",
- "data_source_uid": "source-1",
- "idempotency_key": "orders:2026-07-19",
- }
- assert ledger.claim("jti-1", binding, expires_at=2_000) is True
- assert ledger.claim("jti-1", binding, expires_at=2_000) is False
- ledger.finish(
- "jti-1",
- status="success",
- commit_outcome="committed",
- safe_detail="3 rows affected",
- )
- record = ledger.get("jti-1")
- assert record.status == "success"
- assert record.commit_outcome == "committed"
- assert record.safe_detail == "3 rows affected"
- assert "password" not in repr(record).lower()
|
