tegwick
d40e842052
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a140c4-354a-7823-8f0d-8f717397f786
130 lines
5.2 KiB
Python
130 lines
5.2 KiB
Python
"""Actual source clock, locking and original immutable receipt precision."""
|
|
from datetime import timedelta, timezone
|
|
import sqlite3
|
|
import threading
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
import pytest
|
|
|
|
from approval_engine.errors import Conflict, StoreUnavailable
|
|
from approval_engine.store import Engine, iso, utc_now
|
|
from tests.conftest import FROZEN, binding
|
|
|
|
DIGEST = "sha256:" + "ab" * 32
|
|
|
|
|
|
def issue(engine, start, end):
|
|
obj = engine.create(binding(), {"not_before": start.isoformat(), "expires_at": end.isoformat()})
|
|
return engine.add_entry(obj.id, "human", principal_type="human")
|
|
|
|
|
|
def uses(engine):
|
|
return [row for row in engine.undrained() if row["class"] == "use"]
|
|
|
|
|
|
def test_clock_and_iso_preserve_aware_precision_and_legacy_seconds():
|
|
now = FROZEN.replace(microsecond=750123)
|
|
offset = now.astimezone(timezone(timedelta(hours=2)))
|
|
assert utc_now(lambda: offset) == now
|
|
assert iso(offset) == "2026-08-29T12:00:00.750123+00:00"
|
|
assert iso(FROZEN) == "2026-08-29T12:00:00+00:00"
|
|
|
|
|
|
def test_fractional_expiry_refuses_actual_canonical_counterexample():
|
|
engine = Engine(clock=lambda: FROZEN)
|
|
obj = issue(engine, FROZEN-timedelta(seconds=1), FROZEN.replace(microsecond=500000))
|
|
engine.clock = lambda: FROZEN.replace(microsecond=750000)
|
|
assert engine.claim(obj.id)["valid_now"] is False
|
|
with pytest.raises(Conflict):
|
|
engine.consume(obj.id, DIGEST, decision_id="decision:original")
|
|
assert uses(engine) == []
|
|
|
|
|
|
def test_immediate_fractional_receipt_restart_and_outbox_use_one_instant(tmp_path):
|
|
current = [FROZEN.replace(microsecond=750123)]
|
|
path = tmp_path/"approval.sqlite"
|
|
engine = Engine(path, clock=lambda: current[0])
|
|
obj = issue(engine, FROZEN.replace(microsecond=700000), FROZEN+timedelta(seconds=1))
|
|
samples = []
|
|
def clock():
|
|
value = current[0] + timedelta(microseconds=len(samples))
|
|
samples.append(value)
|
|
return value
|
|
engine.clock = clock
|
|
result = engine.consume(obj.id, DIGEST, decision_id="decision:original")
|
|
assert samples == [current[0]]
|
|
assert result["consumed_at"] == current[0].isoformat()
|
|
assert result["consumed_at"] >= FROZEN.replace(microsecond=700001).isoformat()
|
|
event = uses(engine)[0]
|
|
assert event["created_at"] == event["payload"]["observed_at"] == result["consumed_at"]
|
|
engine.close()
|
|
engine = Engine(path, clock=lambda: FROZEN+timedelta(days=1), auto_migrate=False)
|
|
assert engine.get(obj.id).as_dict()["consumption"]["consumed_at"] == result["consumed_at"]
|
|
retry = engine.consume(obj.id, DIGEST, decision_id="decision:new")
|
|
assert retry == {**result, "idempotent": True}
|
|
assert uses(engine) == [event]
|
|
engine.close()
|
|
|
|
|
|
@pytest.mark.parametrize("after,accepted", [(450000, True), (750000, False)])
|
|
def test_consume_samples_after_real_sqlite_lock_wait(tmp_path, after, accepted):
|
|
current = [FROZEN.replace(microsecond=250000)]
|
|
path = tmp_path/"approval.sqlite"
|
|
engine = Engine(path, clock=lambda: current[0])
|
|
obj = issue(engine, FROZEN-timedelta(seconds=1), FROZEN.replace(microsecond=500000))
|
|
entered = threading.Event()
|
|
initialized = threading.Event()
|
|
start = threading.Event()
|
|
samples = []
|
|
def consume():
|
|
conn = engine._conn()
|
|
conn.set_trace_callback(lambda statement: entered.set() if statement == "BEGIN IMMEDIATE" else None)
|
|
initialized.set()
|
|
assert start.wait(2)
|
|
def clock():
|
|
samples.append(current[0])
|
|
return current[0]
|
|
engine.clock = clock
|
|
try:
|
|
return engine.consume(obj.id, DIGEST, decision_id="decision:original")
|
|
finally:
|
|
engine.close()
|
|
with ThreadPoolExecutor(max_workers=1) as pool:
|
|
future = pool.submit(consume)
|
|
assert initialized.wait(2)
|
|
lock = sqlite3.connect(path)
|
|
lock.execute("BEGIN IMMEDIATE")
|
|
start.set()
|
|
assert entered.wait(2)
|
|
current[0] = FROZEN.replace(microsecond=after)
|
|
lock.commit()
|
|
lock.close()
|
|
if accepted:
|
|
assert future.result(timeout=2)["consumed_at"] == current[0].isoformat()
|
|
else:
|
|
with pytest.raises(Conflict):
|
|
future.result(timeout=2)
|
|
assert samples == [current[0]]
|
|
assert len(uses(engine)) == int(accepted)
|
|
engine.close()
|
|
|
|
|
|
def test_old_second_precision_receipt_and_payload_are_never_backfilled(tmp_path):
|
|
engine = Engine(tmp_path/"old.sqlite", clock=lambda: FROZEN)
|
|
obj = issue(engine, FROZEN-timedelta(seconds=1), FROZEN+timedelta(seconds=1))
|
|
original = engine.consume(obj.id, DIGEST, decision_id="decision:old")
|
|
event = uses(engine)[0]
|
|
assert original["consumed_at"] == "2026-08-29T12:00:00+00:00"
|
|
engine.clock = lambda: FROZEN + timedelta(days=1, microseconds=123456)
|
|
assert engine.consume(obj.id, DIGEST, decision_id="decision:new") == {**original, "idempotent": True}
|
|
assert uses(engine) == [event]
|
|
|
|
|
|
def test_fractional_use_outbox_failure_rolls_back(engine):
|
|
engine.clock = lambda: FROZEN.replace(microsecond=123456)
|
|
obj = issue(engine, FROZEN, FROZEN+timedelta(seconds=1))
|
|
engine.fail_outbox = True
|
|
with pytest.raises(StoreUnavailable):
|
|
engine.consume(obj.id, DIGEST)
|
|
assert engine.get(obj.id).consumed_at is None
|
|
assert uses(engine) == []
|