92 lines
2.4 KiB
Python
92 lines
2.4 KiB
Python
from repo import OraclePayRepo
|
|
|
|
|
|
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": 0,
|
|
"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": "a1b2c3d4e5f6789012345678901234ab",
|
|
"push_status": 0,
|
|
"org_code": None,
|
|
"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]["oms_id"] == 0
|
|
assert executed[0][1]["pay_order_no"] == "ZQD1"
|
|
assert executed[0][1]["pay_order_dtl_id"] == "a1b2c3d4e5f6789012345678901234ab"
|
|
|
|
|
|
def test_load_existing_pay_order_nos():
|
|
class FakeCursor:
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
return False
|
|
|
|
def execute(self, sql, params=None):
|
|
self.sql = sql
|
|
self.params = params
|
|
|
|
def fetchall(self):
|
|
return [("ZQD1",), ("ZQD1",), (None,)]
|
|
|
|
class FakeConn:
|
|
def cursor(self):
|
|
return FakeCursor()
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
repo = OraclePayRepo(connect=lambda: FakeConn())
|
|
keys = repo.load_existing_pay_order_nos()
|
|
assert keys == {"ZQD1"}
|