tegwick
c97e5ea92d
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02b6f-7db1-7222-918b-e813a6bda38d
339 lines
11 KiB
Python
339 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import math
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
import rein_aharness.close_outbox as close_outbox_module
|
|
from rein_aharness.close_outbox import (
|
|
CloseOutbox,
|
|
CloseRequest,
|
|
InvalidCloseRequestError,
|
|
OutboxConflictError,
|
|
OutboxCorruptError,
|
|
)
|
|
|
|
|
|
def _request(
|
|
*,
|
|
run_id: str = "run-1",
|
|
transaction_id: str = "tx-1",
|
|
action: str = "complete",
|
|
result: dict[str, object] | None = None,
|
|
error: str = "",
|
|
reopen: bool = False,
|
|
) -> CloseRequest:
|
|
return CloseRequest(
|
|
run_id=run_id,
|
|
transaction_id=transaction_id,
|
|
action=action,
|
|
result=result or {"ok": True, "accepted_commit": "a" * 40},
|
|
error=error,
|
|
reopen=reopen,
|
|
)
|
|
|
|
|
|
def test_enqueue_uses_private_external_atomic_record(tmp_path: Path) -> None:
|
|
checkout = tmp_path / "checkout"
|
|
checkout.mkdir()
|
|
outbox = CloseOutbox(state_dir=tmp_path / "state")
|
|
|
|
receipt = outbox.enqueue(_request())
|
|
|
|
path = outbox.pending_dir / f"{receipt.entry_id}.json"
|
|
assert receipt.created is True
|
|
assert receipt.state == "pending"
|
|
assert path.is_file()
|
|
assert not path.is_relative_to(checkout)
|
|
assert path.stat().st_mode & 0o777 == 0o600
|
|
assert outbox.root.stat().st_mode & 0o777 == 0o700
|
|
assert outbox.pending_dir.stat().st_mode & 0o777 == 0o700
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
assert payload["state"] == "pending"
|
|
assert payload["attempts"] == 0
|
|
assert payload["entry_id"] == receipt.entry_id
|
|
assert not tuple(outbox.pending_dir.glob("*.tmp"))
|
|
|
|
|
|
def test_identical_enqueue_is_suppressed_without_rewrite(tmp_path: Path) -> None:
|
|
outbox = CloseOutbox(state_dir=tmp_path / "state")
|
|
request = _request()
|
|
first = outbox.enqueue(request)
|
|
path = outbox.pending_dir / f"{first.entry_id}.json"
|
|
before = path.read_bytes()
|
|
|
|
second = outbox.enqueue(request)
|
|
|
|
assert second.created is False
|
|
assert second.state == "pending"
|
|
assert path.read_bytes() == before
|
|
assert outbox.pending_count() == 1
|
|
|
|
|
|
def test_identity_reuse_with_different_intent_is_rejected(tmp_path: Path) -> None:
|
|
outbox = CloseOutbox(state_dir=tmp_path / "state")
|
|
outbox.enqueue(_request(result={"ok": True}))
|
|
|
|
with pytest.raises(OutboxConflictError, match="different close intent"):
|
|
outbox.enqueue(_request(result={"ok": False}))
|
|
|
|
|
|
def test_successful_replay_moves_entry_and_never_redelivers(tmp_path: Path) -> None:
|
|
outbox = CloseOutbox(state_dir=tmp_path / "state")
|
|
receipt = outbox.enqueue(_request())
|
|
delivered: list[CloseRequest] = []
|
|
|
|
report = outbox.replay(delivered.append)
|
|
|
|
assert report.attempted == 1
|
|
assert report.delivered == 1
|
|
assert report.failed == 0
|
|
assert report.remaining == 0
|
|
assert len(delivered) == 1
|
|
path = outbox.delivered_dir / f"{receipt.entry_id}.json"
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
assert payload["state"] == "delivered"
|
|
assert payload["attempts"] == 1
|
|
assert payload["last_error"] is None
|
|
|
|
second = outbox.replay(delivered.append)
|
|
duplicate = outbox.enqueue(_request())
|
|
assert second.attempted == 0
|
|
assert len(delivered) == 1
|
|
assert duplicate.created is False
|
|
assert duplicate.state == "delivered"
|
|
|
|
|
|
def test_failed_delivery_stays_pending_then_replays_same_intent(tmp_path: Path) -> None:
|
|
outbox = CloseOutbox(state_dir=tmp_path / "state")
|
|
request = _request(
|
|
action="fail",
|
|
result={"ok": False},
|
|
error="adapter failed",
|
|
reopen=False,
|
|
)
|
|
receipt = outbox.enqueue(request)
|
|
|
|
def unavailable(close_request: CloseRequest) -> None:
|
|
close_request.result["mutated-by-callback"] = True
|
|
raise RuntimeError("API down\n" + ("x" * 2000))
|
|
|
|
first = outbox.replay(unavailable)
|
|
path = outbox.pending_dir / f"{receipt.entry_id}.json"
|
|
retained = json.loads(path.read_text(encoding="utf-8"))
|
|
assert first.failed == 1
|
|
assert first.remaining == 1
|
|
assert retained["attempts"] == 1
|
|
assert retained["last_error"] == "RuntimeError: close delivery failed"
|
|
assert "API down" not in retained["last_error"]
|
|
assert "x" * 20 not in retained["last_error"]
|
|
assert "mutated-by-callback" not in retained["result"]
|
|
|
|
delivered: list[CloseRequest] = []
|
|
second = outbox.replay(delivered.append)
|
|
final = json.loads(
|
|
(outbox.delivered_dir / path.name).read_text(encoding="utf-8")
|
|
)
|
|
assert second.delivered == 1
|
|
assert final["attempts"] == 2
|
|
assert final["last_error"] is None
|
|
assert delivered[0].intent_digest == request.intent_digest
|
|
|
|
|
|
def test_process_interrupt_leaves_attempt_durable_and_pending(tmp_path: Path) -> None:
|
|
outbox = CloseOutbox(state_dir=tmp_path / "state")
|
|
receipt = outbox.enqueue(_request())
|
|
|
|
def interrupted(_: CloseRequest) -> None:
|
|
raise KeyboardInterrupt
|
|
|
|
with pytest.raises(KeyboardInterrupt):
|
|
outbox.replay(interrupted)
|
|
|
|
payload = json.loads(
|
|
(outbox.pending_dir / f"{receipt.entry_id}.json").read_text(encoding="utf-8")
|
|
)
|
|
assert payload["state"] == "pending"
|
|
assert payload["attempts"] == 1
|
|
assert payload["last_attempt_at"]
|
|
assert payload["last_error"] is None
|
|
|
|
|
|
def test_corrupt_pending_record_is_preserved_in_quarantine(tmp_path: Path) -> None:
|
|
outbox = CloseOutbox(state_dir=tmp_path / "state")
|
|
request = _request()
|
|
receipt = outbox.enqueue(request)
|
|
path = outbox.pending_dir / f"{receipt.entry_id}.json"
|
|
path.write_text("{not-json\n", encoding="utf-8")
|
|
called = False
|
|
|
|
def deliver(_: CloseRequest) -> None:
|
|
nonlocal called
|
|
called = True
|
|
|
|
report = outbox.replay(deliver)
|
|
|
|
assert report.quarantined == 1
|
|
assert report.attempted == 0
|
|
assert report.remaining == 0
|
|
assert called is False
|
|
quarantined = tuple(outbox.quarantine_dir.glob(f"{receipt.entry_id}.*.json"))
|
|
assert len(quarantined) == 2
|
|
assert any(item.name.endswith(".error.json") for item in quarantined)
|
|
assert any(item.read_text(encoding="utf-8") == "{not-json\n" for item in quarantined)
|
|
with pytest.raises(OutboxCorruptError, match="quarantined"):
|
|
outbox.enqueue(request)
|
|
|
|
|
|
def test_pending_record_with_delivered_state_is_quarantined(tmp_path: Path) -> None:
|
|
outbox = CloseOutbox(state_dir=tmp_path / "state")
|
|
receipt = outbox.enqueue(_request())
|
|
path = outbox.pending_dir / f"{receipt.entry_id}.json"
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
payload["state"] = "delivered"
|
|
path.write_text(json.dumps(payload), encoding="utf-8")
|
|
|
|
report = outbox.replay(lambda _: pytest.fail("must not deliver corrupt state"))
|
|
|
|
assert report.quarantined == 1
|
|
assert report.attempted == 0
|
|
assert report.remaining == 0
|
|
|
|
|
|
def test_atomic_enqueue_failure_leaves_no_partial_or_temp_record(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
outbox = CloseOutbox(state_dir=tmp_path / "state")
|
|
|
|
def fail_replace(source: Path, destination: Path) -> None:
|
|
raise OSError("simulated rename failure")
|
|
|
|
monkeypatch.setattr(close_outbox_module.os, "replace", fail_replace)
|
|
with pytest.raises(OSError, match="rename failure"):
|
|
outbox.enqueue(_request())
|
|
|
|
assert not tuple(outbox.pending_dir.iterdir())
|
|
|
|
|
|
def test_two_processes_enqueue_one_durable_intent(tmp_path: Path) -> None:
|
|
state_dir = tmp_path / "state"
|
|
child = """
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from rein_aharness.close_outbox import CloseOutbox, CloseRequest
|
|
outbox = CloseOutbox(state_dir=Path(sys.argv[1]))
|
|
receipt = outbox.enqueue(CloseRequest(
|
|
run_id="run-shared",
|
|
transaction_id="tx-shared",
|
|
action="complete",
|
|
result={"ok": True},
|
|
))
|
|
print(json.dumps({"created": receipt.created, "entry_id": receipt.entry_id}))
|
|
"""
|
|
env = os.environ.copy()
|
|
processes = [
|
|
subprocess.Popen(
|
|
[sys.executable, "-c", child, str(state_dir)],
|
|
cwd=Path(__file__).resolve().parents[1],
|
|
env=env,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
for _ in range(2)
|
|
]
|
|
outputs = [process.communicate(timeout=30) for process in processes]
|
|
|
|
assert [process.returncode for process in processes] == [0, 0]
|
|
receipts = [json.loads(stdout) for stdout, _ in outputs]
|
|
assert sorted(item["created"] for item in receipts) == [False, True]
|
|
assert receipts[0]["entry_id"] == receipts[1]["entry_id"]
|
|
outbox = CloseOutbox(state_dir=state_dir)
|
|
assert outbox.pending_count() == 1
|
|
assert len(tuple(outbox.pending_dir.glob("*.json"))) == 1
|
|
|
|
|
|
def test_replay_limit_is_bounded_and_leaves_remaining_entries(tmp_path: Path) -> None:
|
|
outbox = CloseOutbox(state_dir=tmp_path / "state")
|
|
for number in range(3):
|
|
outbox.enqueue(_request(run_id=f"run-{number}", transaction_id=f"tx-{number}"))
|
|
|
|
report = outbox.replay(lambda _: None, limit=2)
|
|
|
|
assert report.attempted == 2
|
|
assert report.delivered == 2
|
|
assert report.remaining == 1
|
|
with pytest.raises(ValueError, match="between 1 and 1000"):
|
|
outbox.replay(lambda _: None, limit=0)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"updates,match",
|
|
[
|
|
({"run_id": "bad/id"}, "run_id"),
|
|
({"transaction_id": "with space"}, "transaction_id"),
|
|
({"action": "cancel"}, "action"),
|
|
({"action": "complete", "error": "not allowed"}, "cannot carry"),
|
|
({"action": "complete", "reopen": True}, "cannot carry"),
|
|
({"action": "fail", "error": ""}, "require"),
|
|
],
|
|
)
|
|
def test_close_request_rejects_ambiguous_identity_or_action(
|
|
updates: dict[str, object],
|
|
match: str,
|
|
) -> None:
|
|
values: dict[str, object] = {
|
|
"run_id": "run-1",
|
|
"transaction_id": "tx-1",
|
|
"action": "complete",
|
|
"result": {"ok": True},
|
|
"error": "",
|
|
"reopen": False,
|
|
}
|
|
values.update(updates)
|
|
with pytest.raises(InvalidCloseRequestError, match=match):
|
|
CloseRequest(**values) # type: ignore[arg-type]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"result,match",
|
|
[
|
|
({"text": "x" * 4001}, "string exceeds"),
|
|
({"items": list(range(201))}, "array exceeds"),
|
|
({"value": math.nan}, "finite"),
|
|
({"value": 2**63}, "64-bit"),
|
|
({"value": {1, 2}}, "unsupported JSON type"),
|
|
({1: "not-string"}, "field names"),
|
|
(
|
|
{f"field-{number}": "x" * 4000 for number in range(20)},
|
|
"encoded bytes",
|
|
),
|
|
],
|
|
)
|
|
def test_close_request_rejects_unbounded_or_non_json_result(
|
|
result: dict[object, object],
|
|
match: str,
|
|
) -> None:
|
|
with pytest.raises(InvalidCloseRequestError, match=match):
|
|
CloseRequest(
|
|
run_id="run-1",
|
|
transaction_id="tx-1",
|
|
action="complete",
|
|
result=result, # type: ignore[arg-type]
|
|
)
|
|
|
|
|
|
def test_close_request_rejects_excessive_json_depth() -> None:
|
|
value: dict[str, object] = {"leaf": True}
|
|
for _ in range(10):
|
|
value = {"nested": value}
|
|
|
|
with pytest.raises(InvalidCloseRequestError, match="depth bound"):
|
|
_request(result=value)
|