137 lines
3.6 KiB
Python
137 lines
3.6 KiB
Python
from cart_repo import (
|
|
STAMP_CLAIMING,
|
|
STAMP_DONE,
|
|
STAMP_SELECTION,
|
|
MemoryCartRepo,
|
|
OracleCartRepo,
|
|
)
|
|
|
|
|
|
def _row(**overrides) -> dict:
|
|
data = {
|
|
"fid": 1,
|
|
"oms_id": 2026100201,
|
|
"platform_type": 66,
|
|
"supplier_name": "广东壹号药业有限公司",
|
|
"bar_code": "06901234567890",
|
|
"purchase_qty": 7,
|
|
"status": 1,
|
|
"fail_reason": None,
|
|
"push_status": 0,
|
|
"update_user_name": STAMP_SELECTION,
|
|
"operate_type": 1,
|
|
}
|
|
data.update(overrides)
|
|
return data
|
|
|
|
|
|
def test_claim_pending_stamps_processing_and_skips_done():
|
|
repo = MemoryCartRepo(
|
|
[
|
|
_row(),
|
|
_row(fid=2, oms_id=2026100202, update_user_name=STAMP_DONE),
|
|
_row(fid=3, oms_id=2026100203, status=2, update_user_name=STAMP_SELECTION),
|
|
_row(fid=4, oms_id=2026100204, operate_type=2),
|
|
]
|
|
)
|
|
claimed = repo.claim_pending()
|
|
assert [row.fid for row in claimed] == [1]
|
|
assert claimed[0].update_user_name == STAMP_CLAIMING
|
|
assert repo.rows[1]["update_user_name"] == STAMP_CLAIMING
|
|
assert repo.rows[2]["update_user_name"] == STAMP_DONE
|
|
assert repo.rows[3]["status"] == 2
|
|
|
|
|
|
def test_mark_success_does_not_insert_or_change_status():
|
|
repo = MemoryCartRepo([_row()])
|
|
repo.claim_pending()
|
|
repo.mark_success(1)
|
|
assert list(repo.rows) == [1]
|
|
assert repo.rows[1]["status"] == 1
|
|
assert repo.rows[1]["fail_reason"] is None
|
|
assert repo.rows[1]["update_user_name"] == STAMP_DONE
|
|
assert repo.rows[1]["push_status"] == 0
|
|
|
|
|
|
def test_mark_fail_sets_status_and_reason():
|
|
repo = MemoryCartRepo([_row()])
|
|
repo.claim_pending()
|
|
repo.mark_fail(1, "库存不足")
|
|
assert list(repo.rows) == [1]
|
|
assert repo.rows[1]["status"] == 2
|
|
assert repo.rows[1]["fail_reason"] == "库存不足"
|
|
assert repo.rows[1]["update_user_name"] == STAMP_DONE
|
|
|
|
|
|
def test_oracle_claim_updates_stamp():
|
|
fetched = [
|
|
(
|
|
11,
|
|
2026100201,
|
|
66,
|
|
"广东壹号药业有限公司",
|
|
"0690",
|
|
7,
|
|
1,
|
|
None,
|
|
0,
|
|
STAMP_SELECTION,
|
|
1,
|
|
)
|
|
]
|
|
executed: list[tuple[str, dict]] = []
|
|
|
|
class FakeCursor:
|
|
description = [
|
|
("FID",),
|
|
("OMS_ID",),
|
|
("PLATFORM_TYPE",),
|
|
("SUPPLIER_NAME",),
|
|
("BAR_CODE",),
|
|
("PURCHASE_QTY",),
|
|
("STATUS",),
|
|
("FAIL_REASON",),
|
|
("PUSH_STATUS",),
|
|
("UPDATE_USER_NAME",),
|
|
("OPERATE_TYPE",),
|
|
]
|
|
rowcount = 1
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
return False
|
|
|
|
def execute(self, sql, params=None):
|
|
executed.append((sql, params or {}))
|
|
|
|
def fetchall(self):
|
|
return fetched
|
|
|
|
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 = OracleCartRepo(connect=lambda: conn)
|
|
claimed = repo.claim_pending()
|
|
assert conn.committed is True
|
|
assert len(claimed) == 1
|
|
assert claimed[0].fid == 11
|
|
assert claimed[0].update_user_name == STAMP_CLAIMING
|
|
assert "update_user_name" in executed[1][0].lower()
|
|
assert executed[1][1]["fid"] == 11
|
|
assert executed[1][1]["stamp"] == STAMP_CLAIMING
|