283 lines
9.7 KiB
Python
283 lines
9.7 KiB
Python
# 入站单测用的假连接：录 SQL、按 oms_id / fid 存行，不连真库。
|
||
|
||
from __future__ import annotations
|
||
|
||
from contextlib import contextmanager
|
||
from types import SimpleNamespace
|
||
|
||
|
||
# 构造带 oracledb 风格 args[0].code / .message 的假异常。
|
||
def ora_error(code: int, message: str) -> Exception:
|
||
info = SimpleNamespace(code=code, message=message)
|
||
exc = Exception(message)
|
||
exc.args = (info,)
|
||
return exc
|
||
|
||
|
||
# 构造指定约束名的 ORA-00001。
|
||
def unique_error(constraint: str) -> Exception:
|
||
return ora_error(
|
||
1,
|
||
f"ORA-00001: unique constraint (RPAINIT.{constraint}) violated",
|
||
)
|
||
|
||
|
||
# oracledb cursor.getbatcherrors() 单项的替身。
|
||
class BatchError:
|
||
# 记下失败行在本批里的下标、报文和 ORA 码。
|
||
def __init__(self, offset: int, message: str, code: int = 1):
|
||
self.offset = offset
|
||
self.message = message
|
||
self.code = code
|
||
|
||
|
||
_CART_COLS = (
|
||
"fid",
|
||
"operate_type",
|
||
"oms_id",
|
||
"platform_type",
|
||
"supplier_name",
|
||
"bar_code",
|
||
"purchase_price",
|
||
"purchase_qty",
|
||
"purchase_total",
|
||
"status",
|
||
"fail_reason",
|
||
"pay_order_no",
|
||
"pay_order_dtl_id",
|
||
"org_code",
|
||
"push_status",
|
||
)
|
||
|
||
_AGREEMENT_COLS = (
|
||
"fid",
|
||
"oms_id",
|
||
"action_code",
|
||
"contract_status",
|
||
"contract_fail_reason",
|
||
"contract_no",
|
||
"agreement_status",
|
||
"agreement_fail_reason",
|
||
"agreement_id",
|
||
"agreement_dtl_id",
|
||
"org_code",
|
||
"push_status",
|
||
)
|
||
|
||
|
||
# 一块假库：销售商品、采购合同、入站审计、表 2 / 表 4，外加 SQL 记录。
|
||
class FakeDb:
|
||
# 空库，默认不注入故障。
|
||
def __init__(self):
|
||
self.sales: dict[int, dict] = {}
|
||
self.contracts: dict[int, dict] = {}
|
||
self.carts: dict[int, dict] = {}
|
||
self.agreements: dict[int, dict] = {}
|
||
self.audits: list[dict] = []
|
||
self.calls: list[tuple] = []
|
||
self.commits = 0
|
||
self.rollbacks = 0
|
||
self.closes = 0
|
||
self.collide_sales: set[int] = set()
|
||
self.collide_contracts: set[int] = set()
|
||
self.collide_constraint: str | None = None
|
||
self.raise_execute: BaseException | None = None
|
||
|
||
# 新开一条假连接，读写同一块数据。
|
||
def connection(self) -> "FakeConn":
|
||
return FakeConn(self)
|
||
|
||
|
||
# 假 Oracle 连接：cursor / commit / rollback / close。
|
||
class FakeConn:
|
||
# 挂到同一块 FakeDb 上。
|
||
def __init__(self, db: FakeDb):
|
||
self.db = db
|
||
|
||
# 打开一条假游标。
|
||
def cursor(self) -> "FakeCursor":
|
||
return FakeCursor(self.db)
|
||
|
||
# 记下提交次数。
|
||
def commit(self) -> None:
|
||
self.db.commits += 1
|
||
|
||
# 记下回滚次数。
|
||
def rollback(self) -> None:
|
||
self.db.rollbacks += 1
|
||
|
||
# 记下归还次数。
|
||
def close(self) -> None:
|
||
self.db.closes += 1
|
||
|
||
|
||
# 按 SQL 关键字把 SELECT / INSERT / executemany 落到 FakeDb。
|
||
class FakeCursor:
|
||
# 空结果、无批错误。
|
||
def __init__(self, db: FakeDb):
|
||
self.db = db
|
||
self._rows: list[tuple] = []
|
||
self._batch_errors: list[BatchError] = []
|
||
self.description: tuple = ()
|
||
|
||
# with 语句返回自身。
|
||
def __enter__(self) -> "FakeCursor":
|
||
return self
|
||
|
||
# 不吞异常。
|
||
def __exit__(self, *args) -> bool:
|
||
return False
|
||
|
||
# 执行单条 SQL：查重、插入、审计，或表 2 / 表 4 取数回写。
|
||
def execute(self, sql: str, params=None) -> None:
|
||
self.db.calls.append(("execute", sql, params))
|
||
if self.db.raise_execute is not None:
|
||
raise self.db.raise_execute
|
||
self._dispatch(sql, params, offset=None, batcherrors=False)
|
||
|
||
# 按行插入；batcherrors=True 时把唯一冲突记进批错误。
|
||
def executemany(self, sql: str, seq_of_params, batcherrors: bool = False) -> None:
|
||
self.db.calls.append(("executemany", sql, seq_of_params, batcherrors))
|
||
if self.db.raise_execute is not None:
|
||
raise self.db.raise_execute
|
||
self._batch_errors = []
|
||
for offset, params in enumerate(seq_of_params):
|
||
try:
|
||
self._dispatch(sql, params, offset=offset, batcherrors=batcherrors)
|
||
except Exception as exc:
|
||
if batcherrors and _is_unique(exc):
|
||
self._batch_errors.append(
|
||
BatchError(offset, str(exc), code=1)
|
||
)
|
||
continue
|
||
raise
|
||
|
||
# 返回最近一次 SELECT 的行。
|
||
def fetchall(self) -> list[tuple]:
|
||
return list(self._rows)
|
||
|
||
# 返回最近一次 executemany 的批错误。
|
||
def getbatcherrors(self) -> list[BatchError]:
|
||
return list(self._batch_errors)
|
||
|
||
# 按表名把 SQL 分到销售、合同、审计、表 2 或表 4。
|
||
def _dispatch(self, sql: str, params, offset, batcherrors: bool) -> None:
|
||
norm = " ".join(sql.split()).upper()
|
||
binds = params or {}
|
||
if norm.startswith("SELECT") and "RPA_SALES_GOODS_MAPPING_DATA" in norm:
|
||
ids = list(binds.values()) if isinstance(binds, dict) else list(binds)
|
||
self._rows = [(oms_id,) for oms_id in ids if oms_id in self.db.sales]
|
||
return
|
||
if norm.startswith("SELECT") and "RPA_PURCHASE_CONTRACT_DATA" in norm:
|
||
ids = list(binds.values()) if isinstance(binds, dict) else list(binds)
|
||
self._rows = [(fid,) for fid in ids if fid in self.db.contracts]
|
||
return
|
||
if "RPA_OMS_INBOUND_AUDIT" in norm:
|
||
self.db.audits.append(dict(binds))
|
||
return
|
||
if "RPA_CART_SETTLEMENT_GOODS_DATA" in norm:
|
||
if "UPDATE" in norm:
|
||
self._update_pushed(self.db.carts, binds)
|
||
return
|
||
self._select_cart(sql, binds)
|
||
return
|
||
if "RPA_CONTRACT_AGREEMENT_RESULT" in norm:
|
||
if "UPDATE" in norm:
|
||
self._update_pushed(self.db.agreements, binds)
|
||
return
|
||
self._select_agreement(sql, binds)
|
||
return
|
||
if "RPA_SALES_GOODS_MAPPING_DATA" in norm:
|
||
self._insert_sales(binds)
|
||
return
|
||
if "RPA_PURCHASE_CONTRACT_DATA" in norm:
|
||
self._insert_contract(binds)
|
||
return
|
||
raise AssertionError(f"unexpected sql: {sql}")
|
||
|
||
# 捞表 2 push_status=0；SQL 带 PARTITION BY 且绑了 limit 时按 oms_id 整组截断。
|
||
def _select_cart(self, sql: str, binds: dict) -> None:
|
||
rows = [dict(row) for row in self.db.carts.values() if row.get("push_status", 0) == 0]
|
||
rows.sort(key=lambda row: (row["oms_id"], row["fid"]))
|
||
if "PARTITION BY" in sql.upper() and "limit" in binds:
|
||
from app.outbound.cart_settlement import _take_complete_groups
|
||
|
||
rows = _take_complete_groups(rows, int(binds["limit"]))
|
||
self.description = tuple((name.upper(),) for name in _CART_COLS)
|
||
self._rows = [tuple(row.get(name) for name in _CART_COLS) for row in rows]
|
||
|
||
# 捞表 4 push_status=0；SQL 带 PARTITION BY 且绑了 limit 时按合同号整组截断。
|
||
def _select_agreement(self, sql: str, binds: dict) -> None:
|
||
rows = [
|
||
dict(row) for row in self.db.agreements.values() if row.get("push_status", 0) == 0
|
||
]
|
||
rows.sort(key=lambda row: row["fid"])
|
||
if "PARTITION BY" in sql.upper() and "limit" in binds:
|
||
from app.outbound.contract_agreement import _take_complete_groups
|
||
|
||
rows = _take_complete_groups(rows, int(binds["limit"]))
|
||
self.description = tuple((name.upper(),) for name in _AGREEMENT_COLS)
|
||
self._rows = [tuple(row.get(name) for name in _AGREEMENT_COLS) for row in rows]
|
||
|
||
# 按 :id* 回写 push_status / push_result。
|
||
def _update_pushed(self, table: dict[int, dict], binds: dict) -> None:
|
||
status = binds.get("status")
|
||
reason = binds.get("reason")
|
||
fids = [value for key, value in binds.items() if str(key).startswith("id")]
|
||
for fid in fids:
|
||
row = table.get(fid)
|
||
if row is None:
|
||
continue
|
||
row["push_status"] = status
|
||
row["push_result"] = reason
|
||
|
||
# 按 oms_id 写入表 1 替身，冲突时抛 ORA-00001。
|
||
def _insert_sales(self, params: dict) -> None:
|
||
oms_id = params["oms_id"]
|
||
constraint = self.db.collide_constraint
|
||
if constraint and (oms_id in self.db.collide_sales or oms_id in self.db.sales):
|
||
raise unique_error(constraint)
|
||
if oms_id in self.db.sales or oms_id in self.db.collide_sales:
|
||
raise unique_error("UK_RPA_SGMD_OMS_ID")
|
||
row = dict(params)
|
||
row.setdefault("deal_flag", 0)
|
||
self.db.sales[oms_id] = row
|
||
|
||
# 按 fid 写入表 3 替身，冲突时抛 ORA-00001。
|
||
def _insert_contract(self, params: dict) -> None:
|
||
fid = params["fid"]
|
||
constraint = self.db.collide_constraint
|
||
if constraint and (fid in self.db.collide_contracts or fid in self.db.contracts):
|
||
raise unique_error(constraint)
|
||
if fid in self.db.contracts or fid in self.db.collide_contracts:
|
||
raise unique_error("PK_RPA_PCD")
|
||
row = dict(params)
|
||
row.setdefault("process_status", 0)
|
||
self.db.contracts[fid] = row
|
||
|
||
|
||
# 假异常是否为 ORA-00001。
|
||
def _is_unique(exc: BaseException) -> bool:
|
||
if not exc.args:
|
||
return False
|
||
info = exc.args[0]
|
||
return getattr(info, "code", None) == 1
|
||
|
||
|
||
# get_pool 替身：每次 acquire 都借同一块 FakeDb 上的新连接。
|
||
class FakePool:
|
||
# 默认自带一块空库。
|
||
def __init__(self, db: FakeDb | None = None):
|
||
self.db = db or FakeDb()
|
||
self.acquires = 0
|
||
|
||
# 借一条假连接，退出 with 时 close。
|
||
@contextmanager
|
||
def acquire(self):
|
||
self.acquires += 1
|
||
conn = FakeConn(self.db)
|
||
try:
|
||
yield conn
|
||
finally:
|
||
conn.close()
|