"""Real row-lock coverage for admission and asynchronous progress events."""

from concurrent.futures import ThreadPoolExecutor
from threading import Barrier
from unittest.mock import MagicMock

import pytest
from apps.core.exceptions import ConflictError
from apps.core.models import BackgroundJob
from apps.parsers.source_cards import SourceCardService
from django.db import close_old_connections, connection

pytestmark = pytest.mark.django_db(transaction=True)


def _parallel(actions):
    barrier = Barrier(len(actions))

    def run(action):
        close_old_connections()
        try:
            barrier.wait(timeout=10)
            return action()
        finally:
            close_old_connections()

    with ThreadPoolExecutor(max_workers=len(actions)) as executor:
        return list(executor.map(run, actions))


@pytest.mark.parametrize(
    "slug,source",
    [
        ("sme-support-recipients-registry", "fns_sme_support_recipients"),
        ("budget-process-registry", "budget_ubpandnubp"),
        ("sro-membership-check", "sro_membership_check"),
    ],
)
def test_two_simultaneous_refreshes_dispatch_exactly_one_task(slug, source):
    if connection.vendor != "postgresql":
        pytest.skip("Requires PostgreSQL row locks")
    task = MagicMock()
    definition = SourceCardService.get_definition(slug)

    def refresh():
        try:
            SourceCardService._enqueue_refresh_group(
                definition,
                [(task, f"parsers.{source}.refresh", source)],
                requested_by_id=None,
                kwargs={},
            )
            return "queued"
        except ConflictError:
            return "conflict"

    assert sorted(_parallel([refresh, refresh])) == ["conflict", "queued"]
    assert task.apply_async.call_count == 1
    assert BackgroundJob.objects.count() == 1


def test_concurrent_progress_preserves_maximum_and_terminal_result():
    if connection.vendor != "postgresql":
        pytest.skip("Requires PostgreSQL concurrent updates")
    job = BackgroundJob.objects.create(
        task_id="concurrent-progress", task_name="test.task"
    )

    def progress(value):
        return lambda: BackgroundJob.objects.get(pk=job.pk).update_progress(value)

    _parallel([progress(value) for value in [10, 85, 30, 55]])
    job.refresh_from_db()
    assert job.progress == 85
    _parallel([lambda: job.fail("Source unavailable"), progress(25)])
    job.refresh_from_db()
    assert (job.status, job.progress, job.error) == (
        "failure",
        85,
        "Source unavailable",
    )