New_REP_SYSTEM/ERP-core
8
0
Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases Wiki Activity

adding complete db migrations from posgres

Browse Source
This commit is contained in:
ImanThiyanga
2026-10-05 15:57:55 +05:30
parent 9d710349c2
commit 65f58c4000
113 changed files with 24033 additions and 1007 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Backend/smoke/__pycache__/smoke_common.cpython-313.pyc
BIN
View File
Binary file not shown.
Backend/smoke/m8_concurrency.py
+144
View File
@@ -0,0 +1,144 @@
"""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())
Backend/smoke/m9_fifo.py
+149
View File
@@ -0,0 +1,149 @@
"""FIFO stock-consumption tests for the MariaDB port (R4.6 in the migration plan).
FifoCostingService.ConsumeAsync had to be redesigned, not translated: MariaDB has no
`FOR UPDATE OF <alias>` (MySQL 8.0.1+ only), and InnoDB's locking clause otherwise applies
to every table a statement reads — so a single-statement `... NOT EXISTS (grn_lines) ...
FOR UPDATE` would also X-lock grn_lines and batches, deadlocking against GrnService. It is
now a two-step: unlocked candidate SELECT, then a locking read scoped to stock_layers,
then a re-validation under the locks.
Usage:
python m9_fifo.py [--api URL] [--auth URL] [--user EMAIL] [--password PW]
"""
from __future__ import annotations
import concurrent.futures
import json
import urllib.error
import urllib.request
import uuid
from smoke_common import bootstrap
def _req(client, method: str, path: str, body=None, headers=None):
url = f"{client.api.rstrip('/')}{path}"
data = json.dumps(body).encode() if body is not None else None
r = urllib.request.Request(url, data=data, method=method)
r.add_header("Authorization", f"Bearer {client.token}")
if data is not None:
r.add_header("Content-Type", "application/json")
for k, v in (headers or {}).items():
r.add_header(k, v)
try:
with urllib.request.urlopen(r) as resp:
raw = resp.read()
return resp.status, (json.loads(raw) if raw else None)
except urllib.error.HTTPError as e:
raw = e.read()
try:
return e.code, (json.loads(raw) if raw else None)
except json.JSONDecodeError:
return e.code, raw.decode(errors="replace")
def main() -> int:
client, check, _args = bootstrap("MariaDB FIFO consumption tests")
sfx = uuid.uuid4().hex[:8]
print("\nFixtures")
_, branches = _req(client, "GET", "/branches?page=1&pageSize=1")
branch_id = ((branches or {}).get("items") or [{}])[0].get("branchId")
st, wh = _req(client, "POST", "/warehouses",
{"code": f"F-W-{sfx}", "name": f"FIFO WH {sfx}", "branchId": branch_id})
check.check("warehouse created", st, 201)
wh_id = (wh or {}).get("warehouseId")
_, cat = _req(client, "POST", "/categories", {"name": f"FIFO Cat {sfx}"})
_, uom = _req(client, "POST", "/uoms", {"name": f"F-U-{sfx}"})
st, item = _req(client, "POST", "/items", {
"sku": f"F-SKU-{sfx}", "name": f"FIFO Item {sfx}",
"categoryId": (cat or {}).get("categoryId"), "baseUomId": (uom or {}).get("uomId"),
"stockNature": "Stocked", "trackingMode": "None"})
check.check("item created", st, 201)
item_id = (item or {}).get("itemId")
st, vendor = _req(client, "POST", "/vendors",
{"code": f"F-V-{sfx}", "name": f"FIFO Vendor {sfx}",
"currency": "LKR", "branchId": branch_id})
check.check("vendor created", st, 201)
vendor_id = (vendor or {}).get("vendorId")
print("\nInbound: two FIFO layers at different costs")
grn_ids = []
for qty, cost in ((10, 100), (10, 150)):
st, grn = _req(client, "POST", "/grns", {
"vendorId": vendor_id, "warehouseId": wh_id, "branchId": branch_id,
"lines": [{"itemId": item_id, "qty": qty, "unitCost": cost}]})
if st != 201:
print(f" GRN create failed {st}: {grn}")
check.check(f"GRN qty={qty} @ {cost} created", st, 201)
break
grn_id = (grn or {}).get("grnId")
grn_ids.append(grn_id)
st2, _ = _req(client, "POST", f"/grns/{grn_id}/confirm", {})
check.check(f"GRN {grn_id} (qty {qty} @ {cost}) confirmed", st2 in (200, 204), True)
st, onhand = _req(client, "GET", f"/stock/on-hand?itemId={item_id}&warehouseId={wh_id}")
print(f" on-hand response: {onhand}")
print("\nFIFO consume via a negative adjustment (exercises the two-step locking read)")
_, reasons = _req(client, "GET", "/reason-codes?context=Adjustment")
rlist = (reasons or {}).get("items") or reasons or []
var = next((r for r in rlist if r.get("code") == "VAR"), None)
check.check("seeded reason code VAR present", var is not None, True)
reason_id = (var or {}).get("reasonCodeId")
# Consume 15 of 20: must take all 10 @100 then 5 @150 => cost 1750, oldest-first.
st, adj = _req(client, "POST", "/stock-adjustments", {
"warehouseId": wh_id, "branchId": branch_id, "reasonCodeId": reason_id,
"lines": [{"itemId": item_id, "qtyDelta": -15}]})
print(f" adjustment -> {st}: {json.dumps(adj)[:400] if adj else None}")
check.check("negative adjustment accepted (FIFO consume ran)", st, 201)
st, val = _req(client, "GET", f"/stock/valuation?itemId={item_id}&warehouseId={wh_id}")
print(f" valuation: {json.dumps(val)[:400] if val else None}")
if isinstance(val, dict):
check.check("5 units remain after consuming 15 of 20", val.get("totalQty"), 5)
# the survivor is the newer, dearer layer
check.check("remaining value is 5 @ 150 = 750", val.get("totalValue"), 750)
print("\nOver-issue must be blocked, not allowed negative")
st, body = _req(client, "POST", "/stock-adjustments", {
"warehouseId": wh_id, "branchId": branch_id, "reasonCodeId": reason_id,
"lines": [{"itemId": item_id, "qtyDelta": -999}]})
code = (body or {}).get("code") if isinstance(body, dict) else None
print(f" over-issue -> {st} code={code}")
check.check("over-issue rejected with 409", st, 409)
check.check("error code is STOCK_NEGATIVE_BLOCKED", code, "STOCK_NEGATIVE_BLOCKED")
print("\nConcurrent issue of the last 5 units: exactly one wins")
def issue(_):
s, b = _req(client, "POST", "/stock-adjustments", {
"warehouseId": wh_id, "branchId": branch_id, "reasonCodeId": reason_id,
"lines": [{"itemId": item_id, "qtyDelta": -5}]})
return s, ((b or {}).get("code") if isinstance(b, dict) else None)
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
out = list(pool.map(issue, range(2)))
print(f" results={out}")
oks = [s for s, _ in out if s == 201]
check.check("exactly one issue succeeded", len(oks), 1)
check.check("the loser was blocked, not a 500",
all(s in (201, 409) for s, _ in out), True)
st, val = _req(client, "GET", f"/stock/valuation?itemId={item_id}&warehouseId={wh_id}")
if isinstance(val, dict):
check.check("stock never went negative", (val.get("totalQty") or 0) >= 0, True)
print(f" final qty={val.get('totalQty')}")
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())