137 lines
3.5 KiB
Python
137 lines
3.5 KiB
Python
from repo import OraclePayRepo
|
|
|
|
|
|
def test_load_cart_candidates_filters_yaoex_add_cart_success():
|
|
fetched = [
|
|
(1, 11, 66, "广东壹号药业有限公司", "6921", 10, "901", 0),
|
|
]
|
|
sql_holder: dict[str, str] = {}
|
|
|
|
class FakeCursor:
|
|
description = [
|
|
("FID",),
|
|
("OMS_ID",),
|
|
("PLATFORM_TYPE",),
|
|
("SUPPLIER_NAME",),
|
|
("BAR_CODE",),
|
|
("PURCHASE_QTY",),
|
|
("ORG_CODE",),
|
|
("MULTI_PLATFORM_FLAG",),
|
|
]
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
return False
|
|
|
|
def execute(self, sql, params=None):
|
|
sql_holder["sql"] = sql
|
|
sql_holder["params"] = params or {}
|
|
|
|
def fetchall(self):
|
|
return fetched
|
|
|
|
class FakeConn:
|
|
def cursor(self):
|
|
return FakeCursor()
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
repo = OraclePayRepo(connect=lambda: FakeConn())
|
|
rows = repo.load_cart_candidates()
|
|
assert len(rows) == 1
|
|
assert rows[0].oms_id == 11
|
|
assert rows[0].bar_code == "6921"
|
|
assert "operate_type = 1" in sql_holder["sql"].lower()
|
|
assert sql_holder["params"]["platform_type"] == 66
|
|
|
|
|
|
def test_save_pay_rows_inserts_operate_type_2():
|
|
executed: list[tuple[str, dict]] = []
|
|
|
|
class FakeCursor:
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
return False
|
|
|
|
def execute(self, sql, params=None):
|
|
executed.append((sql, params or {}))
|
|
|
|
def executemany(self, sql, params):
|
|
for item in params:
|
|
executed.append((sql, item))
|
|
|
|
class FakeConn:
|
|
committed = False
|
|
|
|
def cursor(self):
|
|
return FakeCursor()
|
|
|
|
def commit(self):
|
|
self.committed = True
|
|
|
|
def rollback(self):
|
|
raise AssertionError("不应 rollback")
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
conn = FakeConn()
|
|
repo = OraclePayRepo(connect=lambda: conn)
|
|
repo.save_pay_rows(
|
|
[
|
|
{
|
|
"operate_type": 2,
|
|
"oms_id": 11,
|
|
"platform_type": 66,
|
|
"supplier_name": "广东壹号药业有限公司",
|
|
"bar_code": "6921",
|
|
"purchase_price": 9.85,
|
|
"purchase_qty": 10,
|
|
"purchase_total": 98.5,
|
|
"status": None,
|
|
"fail_reason": None,
|
|
"pay_order_no": "ZQD1",
|
|
"pay_order_dtl_id": "881",
|
|
"push_status": 0,
|
|
"org_code": "901",
|
|
"multi_platform_flag": 0,
|
|
}
|
|
]
|
|
)
|
|
assert conn.committed is True
|
|
assert "rpa_cart_settlement_goods_data" in executed[0][0].lower()
|
|
assert executed[0][1]["operate_type"] == 2
|
|
assert executed[0][1]["pay_order_no"] == "ZQD1"
|
|
assert executed[0][1]["pay_order_dtl_id"] == "881"
|
|
|
|
|
|
def test_load_existing_pay_keys():
|
|
class FakeCursor:
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
return False
|
|
|
|
def execute(self, sql, params=None):
|
|
self.sql = sql
|
|
|
|
def fetchall(self):
|
|
return [("ZQD1", "881"), ("ZQD2", None)]
|
|
|
|
class FakeConn:
|
|
def cursor(self):
|
|
return FakeCursor()
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
repo = OraclePayRepo(connect=lambda: FakeConn())
|
|
keys = repo.load_existing_pay_keys()
|
|
assert keys == {("ZQD1", "881")}
|