137 lines
4.5 KiB
Python
137 lines
4.5 KiB
Python
# 接口 2 Oracle 取数：SQL 按 oms_id 整组截断；一次 UPDATE 回写。不连真库。
|
||
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
|
||
from app.inbound.repo import RepoUnavailable
|
||
from app.outbound.oracle_repo import _LIST_UNPUSHED_SQL, OracleCartSettlementRepo
|
||
from tests.fake_oracle import FakeDb, ora_error
|
||
|
||
|
||
# 构造一条表 2 未推行。
|
||
def _cart(**overrides) -> dict:
|
||
row = {
|
||
"fid": 1,
|
||
"operate_type": 1,
|
||
"oms_id": 1,
|
||
"platform_type": 48,
|
||
"supplier_name": "示例供应商",
|
||
"bar_code": "06901234567890",
|
||
"purchase_price": 128.5,
|
||
"purchase_qty": 10,
|
||
"purchase_total": 1285,
|
||
"status": 1,
|
||
"fail_reason": None,
|
||
"pay_order_no": None,
|
||
"pay_order_dtl_id": None,
|
||
"org_code": "901",
|
||
"push_status": 0,
|
||
}
|
||
row.update(overrides)
|
||
return row
|
||
|
||
|
||
# 取数 SQL 只捞 push_status=0，按 oms_id 分组，并用 :limit 在库侧截断。
|
||
def test_list_sql_filters_unpushed_and_partitions_by_oms_id():
|
||
sql = " ".join(_LIST_UNPUSHED_SQL.split()).lower()
|
||
assert "rpa_cart_settlement_goods_data" in sql
|
||
assert "push_status = 0" in sql
|
||
assert "partition by" in sql
|
||
assert "oms_id" in sql
|
||
assert ":limit" in sql
|
||
assert "running_start" in sql
|
||
assert "order by" in sql and "oms_id" in sql
|
||
|
||
|
||
# 假连接返回的列名转小写后能对上组装层字段。
|
||
def test_list_unpushed_maps_columns():
|
||
db = FakeDb()
|
||
db.carts[100001] = _cart(
|
||
fid=100001,
|
||
oms_id=20260928000001,
|
||
bar_code="06901234567890",
|
||
)
|
||
repo = OracleCartSettlementRepo(db.connection())
|
||
rows = repo.list_unpushed(limit=500)
|
||
assert rows[0]["fid"] == 100001
|
||
assert rows[0]["operate_type"] == 1
|
||
assert rows[0]["bar_code"] == "06901234567890"
|
||
assert rows[0]["push_status"] == 0
|
||
assert db.closes == 0
|
||
|
||
|
||
# limit=3 时同一 omsId 三行整组进批，多出的另一组留下。
|
||
def test_list_unpushed_keeps_whole_oms_group_when_limit_is_3():
|
||
db = FakeDb()
|
||
db.carts[1] = _cart(fid=1, oms_id=1)
|
||
db.carts[2] = _cart(fid=2, oms_id=1)
|
||
db.carts[3] = _cart(fid=3, oms_id=1)
|
||
db.carts[4] = _cart(fid=4, oms_id=2)
|
||
repo = OracleCartSettlementRepo(db.connection())
|
||
rows = repo.list_unpushed(limit=3)
|
||
assert [row["fid"] for row in rows] == [1, 2, 3]
|
||
sql, params = _last_execute(db)
|
||
assert "partition by" in sql.lower()
|
||
assert params["limit"] == 3
|
||
|
||
|
||
# 库故障对外固定「数据库不可用」，不含 ORA 原文。
|
||
def test_list_unpushed_hides_ora_text():
|
||
db = FakeDb()
|
||
db.raise_execute = ora_error(12541, "ORA-12541: TNS:no listener")
|
||
repo = OracleCartSettlementRepo(db.connection())
|
||
with pytest.raises(RepoUnavailable) as caught:
|
||
repo.list_unpushed(limit=500)
|
||
assert caught.value.message == "数据库不可用"
|
||
assert "ORA-" not in str(caught.value)
|
||
|
||
|
||
# 回写 push_status 只发一条 UPDATE ... WHERE fid IN (...)。
|
||
def test_mark_pushed_updates_in_one_statement():
|
||
db = FakeDb()
|
||
db.carts[1] = _cart(fid=1)
|
||
db.carts[2] = _cart(fid=2)
|
||
repo = OracleCartSettlementRepo(db.connection())
|
||
repo.mark_pushed([1, 2], status=1, reason="ok")
|
||
updates = [
|
||
call for call in db.calls if call[0] == "execute" and "UPDATE" in call[1].upper()
|
||
]
|
||
assert len(updates) == 1
|
||
sql = updates[0][1].upper()
|
||
assert "RPA_CART_SETTLEMENT_GOODS_DATA" in sql
|
||
assert "FID IN" in sql
|
||
assert db.carts[1]["push_status"] == 1
|
||
assert db.carts[2]["push_status"] == 1
|
||
assert db.carts[1]["push_result"] == "ok"
|
||
assert db.commits == 1
|
||
|
||
|
||
# 空 fid 列表不发 SQL、不 commit。
|
||
def test_mark_pushed_empty_fids_does_not_touch_db():
|
||
db = FakeDb()
|
||
repo = OracleCartSettlementRepo(db.connection())
|
||
repo.mark_pushed([], status=1, reason=None)
|
||
assert db.calls == []
|
||
assert db.commits == 0
|
||
|
||
|
||
# 回写失败对外固定「数据库不可用」。
|
||
def test_mark_pushed_hides_ora_text():
|
||
db = FakeDb()
|
||
db.carts[1] = _cart(fid=1)
|
||
db.raise_execute = ora_error(3113, "ORA-03113: end-of-file on communication channel")
|
||
repo = OracleCartSettlementRepo(db.connection())
|
||
with pytest.raises(RepoUnavailable) as caught:
|
||
repo.mark_pushed([1], status=2, reason="fail")
|
||
assert caught.value.message == "数据库不可用"
|
||
assert "ORA-" not in str(caught.value)
|
||
|
||
|
||
# 取出最近一次 execute 的 SQL 和绑定。
|
||
def _last_execute(db: FakeDb) -> tuple[str, dict]:
|
||
execute_calls = [call for call in db.calls if call[0] == "execute"]
|
||
assert execute_calls
|
||
_, sql, params = execute_calls[-1]
|
||
return sql, params or {}
|