dodil/erp-brick-inventory

DODIL ERP brick: inventory

tests/test_commit_conflict.py

"""On DataK3 the loser of a write race on a claim row can fail at COMMIT, after the body returned: the unit retries on a fresh read and, when it
keeps losing, answers the retry-safe 409 - never a 500. Any other integrity failure is not hidden."""
from contextlib import contextmanager

import pytest
from erp_platform.bricks import Refused
from sqlalchemy.exc import IntegrityError

from brick_inventory.application import unit


def _commit_fails(monkeypatch, text, seen):
    @contextmanager
    def repo():
        seen.append(1)
        yield object()
        raise IntegrityError("COMMIT", {}, Exception(text))
    monkeypatch.setattr(unit, "repository", repo)


def test_a_unique_violation_at_commit_is_retried_then_the_409(monkeypatch):
    seen = []
    _commit_fails(monkeypatch, "duplicate key value violates unique constraint (SQLSTATE 23505)", seen)
    with pytest.raises(Refused) as lost:
        unit.run({}, lambda u: "done")
    assert lost.value.status == 409 and len(seen) == unit.ATTEMPTS and "nothing was written" in str(lost.value)


def test_any_other_integrity_error_at_commit_is_not_hidden(monkeypatch):
    seen = []
    _commit_fails(monkeypatch, "null value in column violates not-null constraint", seen)
    with pytest.raises(IntegrityError):
        unit.run({}, lambda u: "done")
    assert len(seen) == 1