PacketStation/tests/server/test_pipeline_concurrency.py

"""The pipeline under concurrency: the FOR UPDATE lock serialises writers; counters never collide; the audit chain
holds after a full packet life; packet_log is append-only for the app role."""

from __future__ import annotations

import threading
from collections.abc import Iterator
from typing import Any

import psycopg
import pytest
from fakedir import FakeDirectory
from fastapi.testclient import TestClient
from pkthelpers import as_user, happy_path, make_unit, phase2_client, ready_draft, seed_users, step
from pshelpers import app_conn, needs_pg

from packetstation import audit
from packetstation.domain.pipeline import next_counter
from packetstation.settings import Settings

pytestmark = needs_pg


@pytest.fixture
def pg_client2(pg_settings: Settings, fake_dir: FakeDirectory, dev_visual: None) -> Iterator[TestClient]:
    with phase2_client(pg_settings, fake_dir) as client:
        seed_users(client)
        yield client


def test_same_step_twice_one_wins(pg_client2: TestClient, pg_settings: Settings, fake_dir: FakeDirectory) -> None:
    with app_conn(pg_settings) as conn:
        _, unit_id = make_unit(conn)
    as_user(pg_client2, "clerk.clara")
    pkt = ready_draft(pg_client2, unit_id)
    with phase2_client(pg_settings, fake_dir) as other:
        as_user(other, "clerk.clara")
        results: list[Any] = [None, None]
        barrier = threading.Barrier(2)

        def run(i: int, client: TestClient) -> None:
            barrier.wait()
            results[i] = step(client, pkt, "send_to_warehouse")

        threads = [threading.Thread(target=run, args=(i, c)) for i, c in enumerate((pg_client2, other))]
        for t in threads:
            t.start()
        for t in threads:
            t.join(30)
    codes = sorted(r.status_code for r in results)
    assert codes == [200, 409], [r.text for r in results]
    loser = next(r for r in results if r.status_code == 409)
    assert loser.json()["code"] == "stale_step"
    assert loser.json()["current"]["status"] == "warehouse" and loser.json()["current"]["version"] == pkt["version"] + 1
    with app_conn(pg_settings) as conn:
        n = conn.execute("SELECT count(*) FROM packet_transitions WHERE packet_id = %s", (pkt["id"],)).fetchone()[0]
        wh = conn.execute("SELECT count(*) FROM packet_wh WHERE packet_id = %s", (pkt["id"],)).fetchone()[0]
    assert n == 1 and wh == 1


def test_next_counter_concurrent(pg_client2: TestClient, pg_settings: Settings) -> None:
    with app_conn(pg_settings) as conn:
        _, unit_id = make_unit(conn)
        pid = conn.execute(
            "INSERT INTO packets (unit_id, dept_code) VALUES (%s, '03') RETURNING id", (unit_id,)
        ).fetchone()[0]
    got: list[int] = []
    lock = threading.Lock()
    barrier = threading.Barrier(8)

    def take() -> None:
        with app_conn(pg_settings) as conn:
            barrier.wait()
            with conn.transaction():
                n = next_counter(conn, pid, "weld")
            with lock:
                got.append(n)

    threads = [threading.Thread(target=take) for _ in range(8)]
    for t in threads:
        t.start()
    for t in threads:
        t.join(30)
    assert sorted(got) == list(range(1, 9))
    with app_conn(pg_settings) as conn:
        assert conn.execute("SELECT next_weld_no FROM packets WHERE id = %s", (pid,)).fetchone()[0] == 9
        with conn.transaction():
            assert next_counter(conn, pid, "page", 3) == 1
            assert next_counter(conn, pid, "page") == 4


def test_audit_chain_after_happy_path(pg_client2: TestClient, pg_settings: Settings) -> None:
    with app_conn(pg_settings) as conn:
        _, unit_id = make_unit(conn)
    happy_path(pg_client2, pg_settings, unit_id)
    with app_conn(pg_settings) as conn:
        ok, n, detail = audit.verify_chain(conn)
        actions = {r[0] for r in conn.execute("SELECT action FROM audit_events")}
    assert ok, detail
    assert n > 20
    assert {"packet.created", "packet.file", "packet.qc_accept", "sig.signed", "asset.uploaded"} <= actions


def test_packet_log_append_only(pg_client2: TestClient, pg_settings: Settings) -> None:
    with app_conn(pg_settings) as conn:
        _, unit_id = make_unit(conn)
    as_user(pg_client2, "clerk.clara")
    pkt = ready_draft(pg_client2, unit_id)
    for sql in (
        "UPDATE packet_log SET note = 'x' WHERE packet_id = %s",
        "DELETE FROM packet_log WHERE packet_id = %s",
        "UPDATE packet_transitions SET reason = 'x' WHERE packet_id = %s",
    ):
        with app_conn(pg_settings) as conn, pytest.raises(psycopg.Error):
            conn.execute(sql, (pkt["id"],))
    with app_conn(pg_settings) as conn, pytest.raises(psycopg.Error):
        conn.execute("TRUNCATE packet_log")