"""On DataK3 the loser of a write race on a claim row (a version claim, a step claim, an idempotency key) can fail
at COMMIT, after the use case returned: that is the retry-safe 409, never a 500 — and a claim that a caller checks
for (engine.claim) is still read as lost, never a crash."""
from contextlib import contextmanager
import pytest
from erp_platform.bricks import Refused
from sqlalchemy.exc import IntegrityError
from brick_hr.application import use_cases
from brick_hr.infrastructure.repositories import is_duplicate
class _Engine:
def __init__(self, text):
self.text = text
@contextmanager
def begin(self):
yield object()
raise IntegrityError("COMMIT", {}, Exception(self.text))
def test_a_unique_violation_at_commit_is_the_409_and_any_other_integrity_error_is_not_hidden():
with pytest.raises(Refused) as lost:
with use_cases._on_engine(_Engine("duplicate key value violates unique constraint (SQLSTATE 23505)")):
pass
assert lost.value.status == 409
assert is_duplicate(lost.value) # a step claim lost at COMMIT is a claim lost
with pytest.raises(IntegrityError):
with use_cases._on_engine(_Engine("null value in column violates not-null constraint")):
pass