78 lines
2.0 KiB
Python
78 lines
2.0 KiB
Python
"""接口 2 Oracle 取数：只捞 push_status=0。不连真库。"""
|
|
from __future__ import annotations
|
|
|
|
from app.outbound.oracle_repo import _LIST_UNPUSHED_SQL, OracleCartSettlementRepo
|
|
|
|
|
|
def test_list_sql_filters_unpushed_and_orders_by_oms_id():
|
|
sql = " ".join(_LIST_UNPUSHED_SQL.split())
|
|
assert "rpa_cart_settlement_goods_data" in sql.lower()
|
|
assert "push_status = 0" in sql
|
|
assert "order by oms_id" in sql.lower()
|
|
assert "fid" in sql.lower()
|
|
|
|
|
|
def test_list_unpushed_maps_columns():
|
|
class FakeCursor:
|
|
description = (
|
|
("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",),
|
|
)
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
return False
|
|
|
|
def execute(self, sql, params=None):
|
|
self.sql = sql
|
|
|
|
def fetchall(self):
|
|
return [
|
|
(
|
|
100001,
|
|
1,
|
|
20260928000001,
|
|
48,
|
|
"示例供应商",
|
|
"06901234567890",
|
|
128.5,
|
|
10,
|
|
1285,
|
|
1,
|
|
None,
|
|
None,
|
|
None,
|
|
"901",
|
|
0,
|
|
)
|
|
]
|
|
|
|
class FakeConn:
|
|
def cursor(self):
|
|
return FakeCursor()
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
repo = OracleCartSettlementRepo(connect=lambda: FakeConn())
|
|
rows = repo.list_unpushed()
|
|
assert rows[0]["fid"] == 100001
|
|
assert rows[0]["operate_type"] == 1
|
|
assert rows[0]["bar_code"] == "06901234567890"
|
|
assert rows[0]["push_status"] == 0
|