498 lines
16 KiB
Python
498 lines
16 KiB
Python
# 选品 Oracle 仓库：.env 填充、领取 SQL、落表一条连接。不连真库。
|
||
|
||
from __future__ import annotations
|
||
|
||
from contextlib import contextmanager
|
||
import os
|
||
|
||
import pytest
|
||
|
||
from rpa_oracle import RepoUnavailable, load_job_env
|
||
from repo import (
|
||
FinishWrite,
|
||
OracleSelectRepo,
|
||
QueryAudit,
|
||
_SELECT_PENDING,
|
||
_SELECT_PENDING_LIMIT,
|
||
)
|
||
|
||
|
||
# 选品 .env 空缺时补 oms .env，已有非空环境变量不覆盖。
|
||
def test_load_runtime_env_fills_blank_from_oms_env(tmp_path, monkeypatch):
|
||
monkeypatch.delenv("RPA_ORACLE_HOST", raising=False)
|
||
monkeypatch.delenv("RPA_ORACLE_PASSWORD", raising=False)
|
||
monkeypatch.setenv("RPA_ORACLE_USER", "keep-me")
|
||
select_env = tmp_path / "select.env"
|
||
oms_env = tmp_path / "oms.env"
|
||
select_env.write_text(
|
||
"RPA_ORACLE_HOST=192.168.12.244\nRPA_ORACLE_PASSWORD=\nRPA_ORACLE_USER=from-select\n",
|
||
encoding="utf-8",
|
||
)
|
||
oms_env.write_text(
|
||
"RPA_ORACLE_PASSWORD=not-for-git\nRPA_ORACLE_USER=from-oms\n",
|
||
encoding="utf-8",
|
||
)
|
||
load_job_env(tmp_path, [select_env, oms_env])
|
||
assert os.environ["RPA_ORACLE_HOST"] == "192.168.12.244"
|
||
assert os.environ["RPA_ORACLE_PASSWORD"] == "not-for-git"
|
||
assert os.environ["RPA_ORACLE_USER"] == "keep-me"
|
||
|
||
|
||
# 缺 HOST/用户/密码时 from_env 抛数据库不可用。
|
||
def test_from_env_without_user_raises(monkeypatch):
|
||
monkeypatch.delenv("RPA_ORACLE_USER", raising=False)
|
||
monkeypatch.delenv("RPA_ORACLE_PASSWORD", raising=False)
|
||
monkeypatch.delenv("RPA_ORACLE_HOST", raising=False)
|
||
with pytest.raises(RepoUnavailable):
|
||
OracleSelectRepo.from_env()
|
||
|
||
|
||
# RETURNING 输出。仓库读 getvalue()[0]。
|
||
class _OutBox:
|
||
def __init__(self):
|
||
self.value = None
|
||
|
||
def getvalue(self):
|
||
return [self.value]
|
||
|
||
|
||
# 录 SQL 的假游标，fetchall 返回预设行。
|
||
class _FakeCursor:
|
||
# 挂上共享的 executed 列表和预设查询结果。
|
||
def __init__(
|
||
self,
|
||
executed: list,
|
||
rows: list | None = None,
|
||
description=None,
|
||
*,
|
||
many: list | None = None,
|
||
fail_on_many: bool = False,
|
||
fail_update_at: int | None = None,
|
||
):
|
||
self._executed = executed
|
||
self._rows = rows or []
|
||
self._many = many if many is not None else []
|
||
self._fail_on_many = fail_on_many
|
||
self._fail_update_at = fail_update_at
|
||
self._updates = 0
|
||
self._next_fid = 1000
|
||
self.description = description
|
||
self.rowcount = 1
|
||
|
||
# 进入 with，返回自身。
|
||
def __enter__(self):
|
||
return self
|
||
|
||
# 退出 with，不吞异常。
|
||
def __exit__(self, *args):
|
||
return False
|
||
|
||
# 记下 RETURNING 的输出变量。getvalue 给成单元素列表。
|
||
def var(self, *_args, **_kwargs):
|
||
return _OutBox()
|
||
|
||
# 记下 SQL 和 bind。一条 IN 超过 1000 个 fid 时抛错，模拟 ORA-01795。
|
||
def execute(self, sql, params=None):
|
||
params = dict(params or {})
|
||
out = params.pop("out_fid", None)
|
||
_reject_oversized_in(params)
|
||
text = str(sql).lstrip().lower()
|
||
if self._fail_on_many and "insert into rpa_cart_settlement_goods_data" in text:
|
||
raise RuntimeError("batch failed")
|
||
if text.startswith("update"):
|
||
self._updates = getattr(self, "_updates", 0) + 1
|
||
if self._fail_update_at is not None and self._updates == self._fail_update_at:
|
||
raise RuntimeError("chunk failed")
|
||
self._last_sql = sql
|
||
self._executed.append((sql, params))
|
||
if out is not None:
|
||
self._next_fid += 1
|
||
out.value = self._next_fid
|
||
|
||
# 记下批量 SQL。fail_on_many 时抛错，用来点名整批回滚。
|
||
def executemany(self, sql, seq):
|
||
if self._fail_on_many:
|
||
raise RuntimeError("batch failed")
|
||
self._many.append((sql, [dict(item) for item in seq]))
|
||
|
||
def fetchall(self):
|
||
"""领取查询返回预设行。落库再读默认仍是 deal_flag=3、表 2 还没有这单。"""
|
||
text = " ".join(str(getattr(self, "_last_sql", "")).split()).lower()
|
||
if text.startswith("select") and "for update" not in text:
|
||
if "count" in text and "rpa_cart_settlement_goods_data" in text:
|
||
return [(0,)]
|
||
if "deal_flag" in text and "where fid" in text:
|
||
return [(3,)]
|
||
return list(self._rows)
|
||
|
||
|
||
# 假连接：同一实例上的 cursor 共享 executed 列表。
|
||
class _FakeConn:
|
||
# 可选带上领取查询要返回的行。
|
||
def __init__(
|
||
self,
|
||
rows: list | None = None,
|
||
description=None,
|
||
*,
|
||
fail_on_many: bool = False,
|
||
fail_update_at: int | None = None,
|
||
):
|
||
self.executed: list[tuple[str, dict]] = []
|
||
self.many: list[tuple[str, list]] = []
|
||
self.committed = False
|
||
self.rolled_back = False
|
||
self.closed = 0
|
||
self._rows = rows or []
|
||
self._description = description
|
||
self._fail_on_many = fail_on_many
|
||
self._fail_update_at = fail_update_at
|
||
|
||
# 新开一个录 SQL 的游标。
|
||
def cursor(self):
|
||
return _FakeCursor(
|
||
self.executed,
|
||
self._rows,
|
||
self._description,
|
||
many=self.many,
|
||
fail_on_many=self._fail_on_many,
|
||
fail_update_at=self._fail_update_at,
|
||
)
|
||
|
||
# 记下已提交。
|
||
def commit(self):
|
||
self.committed = True
|
||
|
||
# 记下已回滚。
|
||
def rollback(self):
|
||
self.rolled_back = True
|
||
|
||
# 记下关闭次数。
|
||
def close(self):
|
||
self.closed += 1
|
||
|
||
|
||
# 只交出一条连接，并记下借了几次。
|
||
class _FakePool:
|
||
# 挂上要借出的那条假连接。
|
||
def __init__(self, conn):
|
||
self._conn = conn
|
||
self.acquires = 0
|
||
|
||
# 借出连接并给 acquires 加一。
|
||
@contextmanager
|
||
def acquire(self):
|
||
self.acquires += 1
|
||
yield self._conn
|
||
|
||
|
||
_CLAIM_DESCRIPTION = (
|
||
("FID",),
|
||
("OMS_ID",),
|
||
("BAR_CODE",),
|
||
("SALE_PRICE",),
|
||
("PLAN_PURCHASE_QTY",),
|
||
("APPROVAL_NO",),
|
||
("GOODS_NAME",),
|
||
("GOODS_SPEC",),
|
||
("MANUFACTURER",),
|
||
("ORG_CODE",),
|
||
("DEAL_FLAG",),
|
||
)
|
||
|
||
|
||
# 构造一条表 2 成功行 bind。
|
||
def _cart_row(**overrides) -> dict:
|
||
row = {
|
||
"operate_type": 1,
|
||
"oms_id": 2026100201,
|
||
"platform_type": 66,
|
||
"supplier_name": "广东壹号药业有限公司",
|
||
"bar_code": "06901234567890",
|
||
"purchase_price": 90,
|
||
"purchase_qty": 10,
|
||
"purchase_total": 900,
|
||
"status": 1,
|
||
"fail_reason": None,
|
||
"push_status": 0,
|
||
"org_code": "907",
|
||
"multi_platform_flag": 0,
|
||
}
|
||
row.update(overrides)
|
||
return row
|
||
|
||
|
||
# 领取 SQL 带 FOR UPDATE SKIP LOCKED，过期 deal_flag=3 可再领。
|
||
def test_claim_sql_uses_skip_locked_and_lease():
|
||
from repo import LEASE_MINUTES
|
||
|
||
sql = " ".join(_SELECT_PENDING.split()).lower()
|
||
assert "for update skip locked" in sql
|
||
assert "deal_flag = 0" in sql
|
||
assert "deal_flag = 3" in sql
|
||
assert "deal_time" in sql
|
||
assert "numtodsinterval(:lease_minutes" in sql
|
||
assert "skip locked" in " ".join(_SELECT_PENDING_LIMIT.split()).lower()
|
||
assert LEASE_MINUTES == 30
|
||
|
||
|
||
# 领取一次借连接：SKIP LOCKED 选出后盖成 3。
|
||
def test_claim_pending_locks_and_stamps_in_one_acquire():
|
||
conn = _FakeConn(
|
||
rows=[
|
||
(1, 2026100201, "06901234567890", 100, 10, "", "药", "0.25g", "厂", "907", 0),
|
||
],
|
||
description=_CLAIM_DESCRIPTION,
|
||
)
|
||
pool = _FakePool(conn)
|
||
repo = OracleSelectRepo(pool)
|
||
claimed = repo.claim_pending(limit=5)
|
||
assert pool.acquires == 1
|
||
assert [row.oms_id for row in claimed] == [2026100201]
|
||
assert claimed[0].deal_flag == 3
|
||
assert conn.committed is True
|
||
select_sql, select_binds = conn.executed[0]
|
||
assert "skip locked" in select_sql.lower()
|
||
assert select_binds["lease_minutes"] == 30
|
||
assert select_binds["limit"] == 5
|
||
update_sql = " ".join(conn.executed[1][0].split()).lower()
|
||
assert "deal_flag = 3" in update_sql
|
||
assert "fid" in update_sql
|
||
|
||
|
||
# 先插表 2 再改表 1，同一连接同一事务。
|
||
def test_save_result_inserts_cart_then_updates_mapping():
|
||
conn = _FakeConn()
|
||
pool = _FakePool(conn)
|
||
repo = OracleSelectRepo(pool)
|
||
repo.save_result(2026100201, 1, "选品成功：1号药城", [_cart_row()], fid=42)
|
||
assert conn.committed is True
|
||
assert conn.rolled_back is False
|
||
assert pool.acquires == 1
|
||
inserts = _cart_inserts(conn)
|
||
assert len(inserts) == 1
|
||
sql = " ".join(inserts[0][0].split())
|
||
assert "rpa_cart_settlement_goods_data" in sql.lower()
|
||
assert "create_user_id" in sql
|
||
assert "create_user_name" in sql
|
||
assert "update_time" in sql
|
||
assert "update_user_id" in sql
|
||
assert "update_user_name" in sql
|
||
assert "SYSDATE, 0, '系统管理员'" in sql
|
||
assert ":update_user_name" in sql
|
||
assert "returning fid into :out_fid" in sql.lower()
|
||
assert sql.count("SYSDATE") == 2
|
||
assert inserts[0][1]["oms_id"] == 2026100201
|
||
assert inserts[0][1]["update_user_name"] == "系统管理员"
|
||
assert "create_user_name" not in inserts[0][1]
|
||
assert "out_fid" not in inserts[0][1]
|
||
assert len(conn.many) == 1
|
||
update_sql = " ".join(conn.many[0][0].split()).lower()
|
||
assert "rpa_sales_goods_mapping_data" in update_sql
|
||
assert "where fid = :fid" in update_sql
|
||
assert conn.many[0][1][0]["fid"] == 42
|
||
assert conn.many[0][1][0]["deal_flag"] == 1
|
||
|
||
|
||
# 一批两行表 2 只从池里借一次连接。
|
||
def test_save_result_acquires_once_for_two_cart_rows():
|
||
conn = _FakeConn()
|
||
pool = _FakePool(conn)
|
||
repo = OracleSelectRepo(pool)
|
||
repo.save_result(
|
||
2026100201,
|
||
1,
|
||
"选品成功：1号药城、药帮忙",
|
||
[_cart_row(purchase_qty=7), _cart_row(platform_type=48, purchase_qty=3)],
|
||
fid=42,
|
||
)
|
||
assert pool.acquires == 1
|
||
inserts = _cart_inserts(conn)
|
||
updates = [item for item in conn.many if "rpa_sales_goods_mapping_data" in item[0].lower()]
|
||
assert len(inserts) == 2
|
||
assert [item[1]["purchase_qty"] for item in inserts] == [7, 3]
|
||
assert len(updates) == 1
|
||
assert len(updates[0][1]) == 1
|
||
assert conn.committed is True
|
||
|
||
|
||
# 两单的表 2 按行取回 fid，表 1 完成更新一次 executemany，只借一条连接。
|
||
def test_save_results_batches_two_orders_in_one_acquire():
|
||
conn = _FakeConn()
|
||
pool = _FakePool(conn)
|
||
repo = OracleSelectRepo(pool)
|
||
fids = repo.save_results(
|
||
[
|
||
FinishWrite(
|
||
oms_id=2026100201,
|
||
fid=42,
|
||
deal_flag=1,
|
||
deal_result="选品成功：1号药城、药帮忙",
|
||
cart_rows=[
|
||
_cart_row(purchase_qty=7, update_user_name="加购中"),
|
||
_cart_row(platform_type=48, purchase_qty=3, update_user_name="加购中"),
|
||
],
|
||
),
|
||
FinishWrite(
|
||
oms_id=2026100202,
|
||
fid=43,
|
||
deal_flag=2,
|
||
deal_result="缺少条形码，无法检索平台",
|
||
cart_rows=[_cart_row(oms_id=2026100202, status=2, purchase_qty=10)],
|
||
),
|
||
]
|
||
)
|
||
assert fids == [[1001, 1002], [1003]]
|
||
assert pool.acquires == 1
|
||
assert conn.committed is True
|
||
assert conn.rolled_back is False
|
||
assert len(conn.many) == 1
|
||
inserts = _cart_inserts(conn)
|
||
update_sql, update_binds = conn.many[0]
|
||
assert [row["oms_id"] for _sql, row in inserts] == [
|
||
2026100201,
|
||
2026100201,
|
||
2026100202,
|
||
]
|
||
assert [row["update_user_name"] for _sql, row in inserts] == [
|
||
"加购中",
|
||
"加购中",
|
||
"系统管理员",
|
||
]
|
||
assert "where fid = :fid" in " ".join(update_sql.split()).lower()
|
||
assert [row["fid"] for row in update_binds] == [42, 43]
|
||
assert [row["deal_flag"] for row in update_binds] == [1, 2]
|
||
|
||
|
||
# 批量写入失败则整批回滚，不提交。
|
||
def test_save_results_rolls_back_the_batch_when_write_fails():
|
||
conn = _FakeConn(fail_on_many=True)
|
||
pool = _FakePool(conn)
|
||
repo = OracleSelectRepo(pool)
|
||
with pytest.raises(RepoUnavailable):
|
||
repo.save_results(
|
||
[
|
||
FinishWrite(
|
||
oms_id=2026100201,
|
||
fid=42,
|
||
deal_flag=1,
|
||
deal_result="选品成功：1号药城",
|
||
cart_rows=[_cart_row()],
|
||
)
|
||
]
|
||
)
|
||
assert conn.rolled_back is True
|
||
assert conn.committed is False
|
||
assert pool.acquires == 1
|
||
|
||
|
||
# 查询失败原文进审计表，一次 executemany，不绑 fid。
|
||
def test_save_query_audits_inserts_error_text_apart_from_deal_result():
|
||
conn = _FakeConn()
|
||
pool = _FakePool(conn)
|
||
repo = OracleSelectRepo(pool)
|
||
repo.save_query_audits(
|
||
[
|
||
QueryAudit(
|
||
oms_id=2026100201,
|
||
mapping_fid=42,
|
||
mall_name="1号药城",
|
||
bar_code="06901234567890",
|
||
error_text="ORA-03114 secret-traceback",
|
||
)
|
||
]
|
||
)
|
||
assert pool.acquires == 1
|
||
assert conn.committed is True
|
||
assert conn.rolled_back is False
|
||
assert len(conn.many) == 1
|
||
sql, binds = conn.many[0]
|
||
folded = " ".join(sql.split()).lower()
|
||
assert "insert into rpa_selection_query_audit" in folded
|
||
assert "( oms_id, mapping_fid, mall_name, bar_code, error_text )" in folded
|
||
assert binds[0]["oms_id"] == 2026100201
|
||
assert binds[0]["mapping_fid"] == 42
|
||
assert binds[0]["error_text"] == "ORA-03114 secret-traceback"
|
||
|
||
|
||
# 审计写入失败只回滚审计自己，不向外抛。
|
||
def test_save_query_audits_swallows_write_failure():
|
||
conn = _FakeConn(fail_on_many=True)
|
||
pool = _FakePool(conn)
|
||
repo = OracleSelectRepo(pool)
|
||
repo.save_query_audits(
|
||
[
|
||
QueryAudit(
|
||
oms_id=2026100201,
|
||
mapping_fid=42,
|
||
mall_name="1号药城",
|
||
bar_code="06901234567890",
|
||
error_text="ORA-03114 secret-traceback",
|
||
)
|
||
]
|
||
)
|
||
assert conn.rolled_back is True
|
||
assert conn.committed is False
|
||
|
||
|
||
def _cart_inserts(conn) -> list:
|
||
"""表 2 插入语句。RETURNING 走 execute，不进 executemany。"""
|
||
return [
|
||
item
|
||
for item in conn.executed
|
||
if "insert into rpa_cart_settlement_goods_data" in item[0].lower()
|
||
]
|
||
|
||
|
||
def _fid_bind_count(params: dict) -> int:
|
||
"""数出本段 bind 里的 fid 个数。"""
|
||
return sum(
|
||
1
|
||
for key in params
|
||
if isinstance(key, str) and key.startswith("id") and key[2:].isdigit()
|
||
)
|
||
|
||
|
||
def _reject_oversized_in(params: dict) -> None:
|
||
"""一条 IN 超过 1000 个 fid 时抛错，模拟 ORA-01795。"""
|
||
if _fid_bind_count(params) > 1000:
|
||
raise RuntimeError("ORA-01795")
|
||
|
||
|
||
def test_claim_1001_rows_updates_first_chunk_without_unavailable():
|
||
"""领取 1001 条时第一段 500 能领走，同一事务提交，不报数据库不可用。"""
|
||
rows = [
|
||
(fid, 2026100200 + fid, "0690", 100, 10, "", "药", "0.25g", "厂", "907", 0)
|
||
for fid in range(1, 1002)
|
||
]
|
||
conn = _FakeConn(rows=rows, description=_CLAIM_DESCRIPTION)
|
||
pool = _FakePool(conn)
|
||
repo = OracleSelectRepo(pool)
|
||
claimed = repo.claim_pending()
|
||
assert pool.acquires == 1
|
||
assert len(claimed) == 1001
|
||
assert claimed[0].deal_flag == 3
|
||
updates = [
|
||
item for item in conn.executed if item[0].lstrip().lower().startswith("update")
|
||
]
|
||
assert [_fid_bind_count(params) for _, params in updates] == [500, 500, 1]
|
||
assert updates[0][1]["id0"] == 1
|
||
assert updates[0][1]["deal_user_name"] == "rpa-select"
|
||
assert conn.committed is True
|
||
assert conn.rolled_back is False
|
||
|
||
|
||
def test_claim_rolls_back_when_a_later_chunk_fails():
|
||
"""后一段更新失败时整笔回滚，不提交，对外数据库不可用。"""
|
||
rows = [
|
||
(fid, 2026100200 + fid, "0690", 100, 10, "", "药", "0.25g", "厂", "907", 0)
|
||
for fid in range(1, 502)
|
||
]
|
||
conn = _FakeConn(rows=rows, description=_CLAIM_DESCRIPTION, fail_update_at=2)
|
||
pool = _FakePool(conn)
|
||
repo = OracleSelectRepo(pool)
|
||
with pytest.raises(RepoUnavailable, match="数据库不可用"):
|
||
repo.claim_pending()
|
||
assert conn.rolled_back is True
|
||
assert conn.committed is False
|
||
assert pool.acquires == 1
|