You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
72 lines
3.2 KiB
72 lines
3.2 KiB
"""Competing import commits use independent PostgreSQL connections.""" |
|
|
|
from concurrent.futures import ThreadPoolExecutor |
|
from threading import Barrier |
|
|
|
import pytest |
|
from django.db import close_old_connections, connection, connections |
|
|
|
from finance.models import AuditEvent, SourceRecord, Transaction |
|
from finance.services import Actor, DomainError, create_account |
|
from ingestion import services |
|
|
|
pytestmark = pytest.mark.django_db(transaction=True) |
|
ACTOR = Actor(kind="AGENT", identifier="synthetic-import-concurrency") |
|
DATA = b"Data,Documento,Historico,Agencia,Lote,Valor\n15/01/2026,1,Synthetic purchase,0,0,-12.50\n" |
|
|
|
|
|
def reviewed(bank): |
|
batch = services.create_preview(data=DATA, profile="BANK_CSV_V1", section_mapping={"1": str(bank.pk)}, actor=ACTOR) |
|
return services.apply_decisions(batch_id=batch.pk, revision=batch.revision, actor=ACTOR, |
|
decisions=[{"locator": batch.items.get().locator, "action": "CREATE"}]) |
|
|
|
|
|
def race(batches): |
|
assert connection.vendor == "postgresql" |
|
barrier = Barrier(2) |
|
|
|
def attempt(batch): |
|
close_old_connections() |
|
try: |
|
with connection.cursor() as cursor: |
|
cursor.execute("SET statement_timeout = '20s'") |
|
cursor.execute("SELECT pg_backend_pid()") |
|
backend_pid = cursor.fetchone()[0] |
|
barrier.wait(timeout=15) |
|
try: |
|
result = services.commit_preview(batch_id=batch.pk, revision=batch.revision, actor=ACTOR) |
|
return backend_pid, "success", result.result |
|
except DomainError as error: |
|
return backend_pid, "failure", error.code |
|
finally: |
|
connections.close_all() |
|
|
|
with ThreadPoolExecutor(max_workers=2) as pool: |
|
results = list(pool.map(attempt, batches)) |
|
assert len({item[0] for item in results}) == 2 |
|
return results |
|
|
|
|
|
def test_two_commits_of_same_batch_return_same_result_without_duplicate_audit(): |
|
bank = create_account(name="Synthetic concurrent bank", kind="BANK", actor=ACTOR) |
|
batch = reviewed(bank) |
|
previous_audits = set(AuditEvent.objects.values_list("pk", flat=True)) |
|
results = race([batch, batch]) |
|
assert [item[1] for item in results] == ["success", "success"] |
|
assert results[0][2] == results[1][2] |
|
assert Transaction.objects.count() == SourceRecord.objects.count() == 1 |
|
assert list(AuditEvent.objects.exclude(pk__in=previous_audits).values_list("outcome", flat=True)) == ["SUCCESS"] |
|
|
|
|
|
def test_identical_batches_have_one_success_and_one_stale_review(): |
|
bank = create_account(name="Synthetic concurrent bank", kind="BANK", actor=ACTOR) |
|
batches = [reviewed(bank), reviewed(bank)] |
|
previous_audits = set(AuditEvent.objects.values_list("pk", flat=True)) |
|
results = race(batches) |
|
assert sorted(item[1] for item in results) == ["failure", "success"] |
|
assert next(item[2] for item in results if item[1] == "failure") == "STALE_PREVIEW" |
|
assert Transaction.objects.count() == SourceRecord.objects.count() == 1 |
|
for batch in batches: |
|
batch.refresh_from_db() |
|
assert sorted(batch.state for batch in batches) == ["COMMITTED", "PREVIEW"] |
|
assert sorted(AuditEvent.objects.exclude(pk__in=previous_audits).values_list("outcome", flat=True)) == ["FAILURE", "SUCCESS"]
|
|
|