222 lines
9.5 KiB
Python
222 lines
9.5 KiB
Python
 
|
"""Phase 5 shop writes under concurrency: weld and heat tag numbers come from the packet counter under the row
|
||
|
|
lock, so two sessions marking at once never collide; two supervisors verify two departments of one change order
|
||
|
|
at once and both land; two users tick different checklist keys and both stay; a stale expect is refused with the
|
||
|
|
message; a write to a packet filed in the meantime is refused."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import threading
|
||
|
|
from collections.abc import Callable, Iterator
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
from fakedir import FakeDirectory
|
||
|
|
from fastapi.testclient import TestClient
|
||
|
|
from pkthelpers import step
|
||
|
|
from pshelpers import needs_pg
|
||
|
|
from shophelpers import (
|
||
|
|
SIGN,
|
||
|
|
STALE,
|
||
|
|
Shop,
|
||
|
|
all_rows,
|
||
|
|
finish_stage,
|
||
|
|
fresh,
|
||
|
|
in_progress,
|
||
|
|
issue_co,
|
||
|
|
ok,
|
||
|
|
open_shop,
|
||
|
|
refused,
|
||
|
|
run,
|
||
|
|
to_station,
|
||
|
|
verify,
|
||
|
|
)
|
||
|
|
|
||
|
|
from packetstation.settings import Settings
|
||
|
|
|
||
|
|
pytestmark = needs_pg
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def shop(pg_settings: Settings, fake_dir: FakeDirectory, dev_visual: None) -> Iterator[Shop]:
|
||
|
|
yield from open_shop(pg_settings, fake_dir)
|
||
|
|
|
||
|
|
|
||
|
|
def together(calls: list[Callable[[], Any]]) -> list[Any]:
|
||
|
|
"""Run the calls in threads released by one barrier. Returns their answers in order."""
|
||
|
|
results: list[Any] = [None] * len(calls)
|
||
|
|
errors: list[BaseException] = []
|
||
|
|
barrier = threading.Barrier(len(calls))
|
||
|
|
|
||
|
|
def run_one(i: int) -> None:
|
||
|
|
try:
|
||
|
|
barrier.wait(10)
|
||
|
|
results[i] = calls[i]()
|
||
|
|
except BaseException as e: # reported below
|
||
|
|
errors.append(e)
|
||
|
|
|
||
|
|
threads = [threading.Thread(target=run_one, args=(i,)) for i in range(len(calls))]
|
||
|
|
for t in threads:
|
||
|
|
t.start()
|
||
|
|
for t in threads:
|
||
|
|
t.join(60)
|
||
|
|
assert not errors, errors
|
||
|
|
return results
|
||
|
|
|
||
|
|
|
||
|
|
def weld_call(c: TestClient, pid: int, fx: float) -> Callable[[], Any]:
|
||
|
|
body = {"page": 1, "fx": fx, "fy": 0.5, "items": [1], "type": "Fillet", "process": "GMAW"}
|
||
|
|
return lambda: c.post(f"/api/packets/{pid}/welds", json=body)
|
||
|
|
|
||
|
|
|
||
|
|
def tag_call(c: TestClient, pid: int, fx: float) -> Callable[[], Any]:
|
||
|
|
return lambda: c.post(f"/api/packets/{pid}/heat-tags", json={"page": 1, "item": 2, "fx": fx, "fy": 0.4})
|
||
|
|
|
||
|
|
|
||
|
|
def test_concurrent_welds_never_collide(shop: Shop) -> None:
|
||
|
|
pkt = in_progress(shop)
|
||
|
|
pid = pkt["id"]
|
||
|
|
# weld.andy on two sessions (two tablets) and sup.wes, three welds each, all at once
|
||
|
|
clients = [shop.fresh("weld.andy"), shop.fresh("weld.andy"), shop.fresh("sup.wes")]
|
||
|
|
calls = [weld_call(c, pid, 0.1 + 0.1 * i + 0.01 * k) for i, c in enumerate(clients) for k in range(3)]
|
||
|
|
answers = together(calls)
|
||
|
|
assert [r.status_code for r in answers] == [200] * len(calls), [r.text for r in answers if r.status_code != 200]
|
||
|
|
nos = sorted(w["weld_no"] for w in fresh(shop("sup.wes"), pid)["welds"])
|
||
|
|
assert nos == list(range(1, len(calls) + 1))
|
||
|
|
rows = all_rows(shop.settings, "SELECT weld_no FROM welds WHERE packet_id = %s ORDER BY weld_no", (pid,))
|
||
|
|
assert [r[0] for r in rows] == nos
|
||
|
|
marked = sorted(
|
||
|
|
r[0]
|
||
|
|
for r in all_rows(shop.settings, "SELECT action FROM packet_log WHERE packet_id = %s", (pid,))
|
||
|
|
if r[0].endswith(" marked") and r[0].startswith("Weld W")
|
||
|
|
)
|
||
|
|
assert marked == sorted(f"Weld W{n} marked" for n in nos)
|
||
|
|
assert (
|
||
|
|
int(all_rows(shop.settings, "SELECT next_weld_no FROM packets WHERE id = %s", (pid,))[0][0]) == len(calls) + 1
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def test_two_welds_at_once_are_w1_and_w2(shop: Shop) -> None:
|
||
|
|
pkt = in_progress(shop)
|
||
|
|
a, b = together(
|
||
|
|
[weld_call(shop.fresh("weld.andy"), pkt["id"], 0.2), weld_call(shop.fresh("sup.wes"), pkt["id"], 0.7)]
|
||
|
|
)
|
||
|
|
assert a.status_code == 200 and b.status_code == 200, (a.text, b.text)
|
||
|
|
got = {w["fx"]: w["weld_no"] for w in fresh(shop("sup.wes"), pkt["id"])["welds"]}
|
||
|
|
assert sorted(got.values()) == [1, 2] and got[0.2] != got[0.7]
|
||
|
|
|
||
|
|
|
||
|
|
def test_concurrent_heat_tags_never_collide(shop: Shop) -> None:
|
||
|
|
pkt = in_progress(shop)
|
||
|
|
pid = pkt["id"]
|
||
|
|
clients = [shop.fresh("weld.andy"), shop.fresh("sup.wes"), shop.fresh("lead.lou")]
|
||
|
|
calls = [tag_call(c, pid, 0.1 + 0.1 * i + 0.01 * k) for i, c in enumerate(clients) for k in range(3)]
|
||
|
|
answers = together(calls)
|
||
|
|
assert [r.status_code for r in answers] == [200] * len(calls), [r.text for r in answers if r.status_code != 200]
|
||
|
|
nos = sorted(t["tag_no"] for t in fresh(shop("sup.wes"), pid)["heat_tags"])
|
||
|
|
assert nos == list(range(1, len(calls) + 1))
|
||
|
|
|
||
|
|
|
||
|
|
def test_two_supervisors_verify_at_once(shop: Shop) -> None:
|
||
|
|
pkt = in_progress(shop)
|
||
|
|
pkt = issue_co(shop, pkt, ["03", "06"])
|
||
|
|
assert pkt["change_orders"][0]["required"] == ["03", "06"]
|
||
|
|
wes, mia = shop.fresh("sup.wes"), shop.fresh("multi.mia")
|
||
|
|
|
||
|
|
def call(c: TestClient, dept: str) -> Callable[[], Any]:
|
||
|
|
body = {"dept": dept, "note": f"Checked in {dept}", "sign": SIGN}
|
||
|
|
return lambda: c.post(f"/api/packets/{pkt['id']}/change-orders/1/verify", json=body)
|
||
|
|
|
||
|
|
a, b = together([call(wes, "03"), call(mia, "06")])
|
||
|
|
assert a.status_code == 200 and b.status_code == 200, (a.text, b.text)
|
||
|
|
out = fresh(shop("sup.wes"), pkt["id"])
|
||
|
|
co = out["change_orders"][0]
|
||
|
|
assert set(co["verified"]) == {"03", "06"} and co["pending_depts"] == [] and co["open"] is False
|
||
|
|
assert out["open_cos"] == []
|
||
|
|
rows = all_rows(shop.settings, "SELECT dept_code FROM co_verifications ORDER BY dept_code")
|
||
|
|
assert [r[0] for r in rows] == ["03", "06"]
|
||
|
|
notes = sorted(
|
||
|
|
r[0]
|
||
|
|
for r in all_rows(
|
||
|
|
shop.settings,
|
||
|
|
"SELECT note FROM packet_log WHERE packet_id = %s AND action LIKE 'Change order CO-1 verified by %%'",
|
||
|
|
(pkt["id"],),
|
||
|
|
)
|
||
|
|
)
|
||
|
|
# one of them landed first and still waited on the other; the second fully verified it
|
||
|
|
assert len(notes) == 2
|
||
|
|
assert any(n.endswith("change order fully verified") for n in notes)
|
||
|
|
assert any("still waiting on" in n for n in notes)
|
||
|
|
# the same department twice at once: one lands, one is told it is done
|
||
|
|
pkt2 = issue_co(shop, out, ["03"])
|
||
|
|
no = pkt2["change_orders"][-1]["co_no"]
|
||
|
|
|
||
|
|
def again(c: TestClient) -> Callable[[], Any]:
|
||
|
|
return lambda: c.post(
|
||
|
|
f"/api/packets/{pkt['id']}/change-orders/{no}/verify", json={"dept": "03", "note": "x", "sign": SIGN}
|
||
|
|
)
|
||
|
|
|
||
|
|
a, b = together([again(shop.fresh("sup.wes")), again(shop.fresh("sup.wes"))])
|
||
|
|
assert sorted([a.status_code, b.status_code]) == [200, 409], (a.text, b.text)
|
||
|
|
loser = a if a.status_code == 409 else b
|
||
|
|
assert loser.json()["code"] == "already_verified"
|
||
|
|
|
||
|
|
|
||
|
|
def test_checklist_keys_merge(shop: Shop) -> None:
|
||
|
|
pkt = in_progress(shop)
|
||
|
|
andy, wes = shop.fresh("weld.andy"), shop.fresh("sup.wes")
|
||
|
|
url = f"/api/packets/{pkt['id']}/stage/checklist"
|
||
|
|
pairs = [(andy, "dwg"), (wes, "heat"), (andy, "wps"), (wes, "fit")]
|
||
|
|
answers = together([(lambda c=c, k=k: c.put(f"{url}/{k}", json={"done": True})) for c, k in pairs])
|
||
|
|
assert [r.status_code for r in answers] == [200] * 4, [r.text for r in answers]
|
||
|
|
ck = fresh(shop("sup.wes"), pkt["id"])["stage"]["checklist"]
|
||
|
|
assert {k for k, v in ck.items() if v.get("done")} == {"dwg", "heat", "wps", "fit"}
|
||
|
|
assert ck["dwg"]["by"] == shop.uid("weld.andy") and ck["heat"]["by"] == shop.uid("sup.wes")
|
||
|
|
|
||
|
|
|
||
|
|
def test_stale_expect_is_refused(shop: Shop) -> None:
|
||
|
|
pkt = finish_stage(shop, in_progress(shop))
|
||
|
|
andy_view = fresh(shop("weld.andy"), pkt["id"])
|
||
|
|
ok(step(shop("sup.wes"), pkt, "pull_back"))
|
||
|
|
r = refused(step(shop("weld.andy"), andy_view, "submit", sign=SIGN), 409, "stale_step", STALE)
|
||
|
|
assert r["current"]["status"] == "ready" and r["current"]["dept"] == "03"
|
||
|
|
assert r["current"]["version"] == fresh(shop("sup.wes"), pkt["id"])["version"]
|
||
|
|
# pull back and start at once: one of them is stale
|
||
|
|
pkt = to_station(shop, fresh(shop("sup.wes"), pkt["id"]))
|
||
|
|
last = all_rows(
|
||
|
|
shop.settings, "SELECT coalesce(max(id), 0) FROM packet_transitions WHERE packet_id = %s", (pkt["id"],)
|
||
|
|
)
|
||
|
|
wes, andy = shop.fresh("sup.wes"), shop.fresh("weld.andy")
|
||
|
|
a, b = together([lambda: step(wes, pkt, "unassign"), lambda: step(andy, pkt, "start")])
|
||
|
|
assert sorted([a.status_code, b.status_code]) == [200, 409], (a.text, b.text)
|
||
|
|
loser = a if a.status_code == 409 else b
|
||
|
|
assert loser.json()["code"] == "stale_step" and loser.json()["detail"] == STALE
|
||
|
|
n = all_rows(
|
||
|
|
shop.settings,
|
||
|
|
"SELECT count(*) FROM packet_transitions WHERE packet_id = %s AND id > %s",
|
||
|
|
(pkt["id"], last[0][0]),
|
||
|
|
)[0][0]
|
||
|
|
assert n == 1
|
||
|
|
|
||
|
|
|
||
|
|
def test_filed_in_the_meantime_is_refused(shop: Shop) -> None:
|
||
|
|
pkt = in_progress(shop)
|
||
|
|
pkt = issue_co(shop, pkt, ["03"])
|
||
|
|
seen = fresh(shop("sup.wes"), pkt["id"])
|
||
|
|
run(shop.settings, "UPDATE packets SET status = 'filed', version = version + 1 WHERE id = %s", (pkt["id"],))
|
||
|
|
msg = "This packet is filed and locked. IT can reopen it if a correction is needed."
|
||
|
|
refused(verify(shop, seen, 1, "03", "sup.wes"), 403, "filed", msg)
|
||
|
|
refused(
|
||
|
|
shop("clerk.clara").patch(f"/api/packets/{pkt['id']}", json={"set": {"drawing": "D-9"}, "base": {}}),
|
||
|
|
403,
|
||
|
|
"filed",
|
||
|
|
msg,
|
||
|
|
)
|
||
|
|
refused(verify(shop, seen, 1, "03", "admin.ann", it_reason="Ticket 5502: verify after filing"), 403, "filed", msg)
|
||
|
|
# a shop edit is not open on a filed packet either
|
||
|
|
r = shop("weld.andy").post(
|
||
|
|
f"/api/packets/{pkt['id']}/welds", json={"page": 1, "fx": 0.5, "fy": 0.5, "items": [1], "type": "Fillet"}
|
||
|
|
)
|
||
|
|
assert r.status_code == 403 and r.json()["code"] in ("filed", "wrong_status"), r.text
|
||
|
|
r = step(shop("weld.andy"), seen, "submit", sign=SIGN)
|
||
|
|
assert r.status_code == 409 and r.json()["code"] == "stale_step", r.text
|