"""Concurrency smoke tests for the MySQL/MariaDB port (R4 in the migration plan). These cover what no other suite can: on PostgreSQL the ETag token was the `xmin` system column, maintained by the engine. On MariaDB it is a real `RowVersion int unsigned` column incremented by ErpDbContext.BumpRowVersions(). The InMemory unit-test provider does not implement optimistic concurrency at all, so it stays green whether that works or not. Run order matters: R4.1 gates everything else. If the ETag does not change across an update, nothing downstream is meaningful. Usage: python m8_concurrency.py [--api URL] [--auth URL] [--user EMAIL] [--password PW] """ from __future__ import annotations import concurrent.futures import json import uuid import urllib.error import urllib.request from smoke_common import bootstrap def uniqueish() -> str: return uuid.uuid4().hex[:8] def _raw_request(client, method: str, path: str, body=None, headers=None): """Request that returns (status, body, headers) without raising on 4xx.""" url = f"{client.api.rstrip('/')}{path}" data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(url, data=data, method=method) req.add_header("Authorization", f"Bearer {client.token}") if data is not None: req.add_header("Content-Type", "application/json") for k, v in (headers or {}).items(): req.add_header(k, v) try: with urllib.request.urlopen(req) as r: raw = r.read() return r.status, (json.loads(raw) if raw else None), dict(r.headers) except urllib.error.HTTPError as e: raw = e.read() try: parsed = json.loads(raw) if raw else None except json.JSONDecodeError: parsed = raw.decode(errors="replace") return e.code, parsed, dict(e.headers) def main() -> int: client, check, args = bootstrap("MySQL port concurrency tests") print("\nR4.1 ETag round-trip (gates every test below)") suffix = uniqueish() status, brand, headers = _raw_request(client, "POST", "/brands", {"name": f"E2E Brand {suffix}"}) check.check("POST /brands -> 201", status, 201) brand_id = (brand or {}).get("brandId") etag_1 = headers.get("ETag") print(f" brandId={brand_id} ETag={etag_1}") check.check("create returns an ETag", etag_1 is not None, True) # Added -> 1 means the token is never the all-zero "AAAAAA==" a default uint produces. check.check('new row ETag is not the all-zero token', etag_1 != '"AAAAAA=="', True) status, _, headers = _raw_request( client, "PUT", f"/brands/{brand_id}", {"name": f"E2E Brand {suffix} renamed"}, {"If-Match": etag_1}) check.check("PUT with current ETag -> 200", status, 200) etag_2 = headers.get("ETag") print(f" new ETag={etag_2}") check.check("ETag CHANGED after update (the whole point)", etag_2 != etag_1, True) status, _, _ = _raw_request( client, "PUT", f"/brands/{brand_id}", {"name": f"E2E Brand {suffix} again"}, {"If-Match": etag_1}) check.check("PUT with the STALE ETag -> 412", status, 412) status, _, _ = _raw_request( client, "PUT", f"/brands/{brand_id}", {"name": f"E2E Brand {suffix} no match"}) check.check("PUT with no If-Match -> 428", status, 428) print("\nR4.2 Lost update: two concurrent PUTs with the same valid ETag") status, brand, headers = _raw_request(client, "POST", "/brands", {"name": f"E2E Race {suffix}"}) race_id = (brand or {}).get("brandId") race_etag = headers.get("ETag") def put(n): return _raw_request(client, "PUT", f"/brands/{race_id}", {"name": f"E2E Race {suffix} w{n}"}, {"If-Match": race_etag})[0] with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool: codes = sorted(pool.map(put, range(2))) print(f" status codes={codes}") check.check("exactly one 200", codes.count(200), 1) check.check("exactly one 412", codes.count(412), 1) print("\nR4.3 Document numbers are unique under concurrency (upsert + RETURNING)") # Requisitions mint a doc number through NumberSequenceService inside a UoW transaction. # N parallel creates must yield N DISTINCT numbers. This is the test that would catch a # broken upsert: the LAST_INSERT_ID variant, or an INSERT-then-SELECT split, hands the # same number to two callers. _, branches, _ = _raw_request(client, "GET", "/branches?page=1&pageSize=1") branch_id = ((branches or {}).get("items") or [{}])[0].get("branchId") check.check("seeded default branch is reachable", branch_id is not None, True) _, cat, _ = _raw_request(client, "POST", "/categories", {"name": f"E2E Cat {suffix}"}) _, uom, _ = _raw_request(client, "POST", "/uoms", {"name": f"E2E-U-{suffix}"}) st, item, _ = _raw_request(client, "POST", "/items", { "sku": f"E2E-SKU-{suffix}", "name": f"E2E Item {suffix}", "categoryId": (cat or {}).get("categoryId"), "baseUomId": (uom or {}).get("uomId"), "stockNature": "Stocked", "trackingMode": "None"}) check.check("fixture item created", st, 201) item_id = (item or {}).get("itemId") n_parallel = 8 def make_requisition(_): s, b, _h = _raw_request(client, "POST", "/requisitions", { "branchId": branch_id, "lines": [{"itemId": item_id, "qty": 1}]}) return s, (b or {}).get("docNo") with concurrent.futures.ThreadPoolExecutor(max_workers=n_parallel) as pool: results = list(pool.map(make_requisition, range(n_parallel))) statuses = [s for s, _ in results] doc_nos = [d for s, d in results if s == 201 and d] print(f" statuses={statuses}") print(f" docNos={doc_nos}") check.check(f"all {n_parallel} requisitions created", len(doc_nos), n_parallel) check.check("every document number is DISTINCT", len(set(doc_nos)), len(doc_nos)) check.check("doc numbers carry the branch code", all("BR-000" in d for d in doc_nos) if doc_nos else False, True) print("\n" + "=" * 62) print(f" PASSED {check.passed} FAILED {check.failed}") print("=" * 62) return 1 if check.failed else 0 if __name__ == "__main__": raise SystemExit(main())