259 lines
7.6 KiB
Python
259 lines
7.6 KiB
Python
from fastapi.testclient import TestClient
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models.attachment import AttachmentCategory
|
||
|
||
|
||
def _seed(db: Session) -> list[int]:
|
||
cats = [
|
||
AttachmentCategory(name="TEST_CERT", sort_order=1, is_active=True),
|
||
AttachmentCategory(name="TEST_EXTRA", sort_order=2, is_active=True),
|
||
]
|
||
for c in cats:
|
||
db.add(c)
|
||
db.commit()
|
||
for c in cats:
|
||
db.refresh(c)
|
||
return [c.id for c in cats]
|
||
|
||
|
||
def test_attachment_box_basic_and_flip(client: TestClient, db: Session):
|
||
"""扫附件码即到货：未判定 → 自动 has，item 不存在则按 ERP 数量建。"""
|
||
cert_id, _ = _seed(db)
|
||
resp = client.post(
|
||
"/CargoTrace/attachment/box",
|
||
json={
|
||
"zongpai_no": "26BW0011",
|
||
"category_id": cert_id,
|
||
"box_no": 931,
|
||
"quantity": 30,
|
||
},
|
||
)
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["paichan_no"] == "W00009"
|
||
assert body["box_no"] == 931
|
||
assert body["quantity"] == 30
|
||
|
||
status = client.get(
|
||
"/CargoTrace/attachment/status", params={"zongpai_no": "26BW0011"}
|
||
).json()
|
||
assert status["determination"] == "has"
|
||
item = status["items"][0]
|
||
assert item["expected_qty"] == 80 # ERP 数量兜底
|
||
assert item["boxed_qty"] == 30
|
||
|
||
|
||
def test_attachment_box_rejects_over_expected(client: TestClient, db: Session):
|
||
cert_id, _ = _seed(db)
|
||
client.put(
|
||
"/CargoTrace/attachment/config",
|
||
json={
|
||
"zongpai_no": "26BW0011",
|
||
"determination": "has",
|
||
"items": [{"category_id": cert_id, "expected_qty": 80}],
|
||
},
|
||
)
|
||
first = client.post(
|
||
"/CargoTrace/attachment/box",
|
||
json={
|
||
"zongpai_no": "26BW0011",
|
||
"category_id": cert_id,
|
||
"box_no": 932,
|
||
"quantity": 70,
|
||
},
|
||
)
|
||
assert first.status_code == 200
|
||
over = client.post(
|
||
"/CargoTrace/attachment/box",
|
||
json={
|
||
"zongpai_no": "26BW0011",
|
||
"category_id": cert_id,
|
||
"box_no": 933,
|
||
"quantity": 11,
|
||
},
|
||
)
|
||
assert over.status_code == 400
|
||
assert over.json()["error_code"] == "INVALID_QUANTITY"
|
||
|
||
|
||
def test_attachment_box_duplicate_in_same_box(client: TestClient, db: Session):
|
||
cert_id, _ = _seed(db)
|
||
client.post(
|
||
"/CargoTrace/attachment/box",
|
||
json={
|
||
"zongpai_no": "26BW0011",
|
||
"category_id": cert_id,
|
||
"box_no": 934,
|
||
"quantity": 10,
|
||
},
|
||
)
|
||
dup = client.post(
|
||
"/CargoTrace/attachment/box",
|
||
json={
|
||
"zongpai_no": "26BW0011",
|
||
"category_id": cert_id,
|
||
"box_no": 934,
|
||
"quantity": 5,
|
||
},
|
||
)
|
||
assert dup.status_code == 409
|
||
assert dup.json()["error_code"] == "DUPLICATE_BOX_NO"
|
||
|
||
|
||
def test_attachment_box_mixed_with_product(client: TestClient, db: Session):
|
||
"""附件与产品同箱：先装产品，再装附件，同 box_no。"""
|
||
cert_id, _ = _seed(db)
|
||
client.post(
|
||
"/CargoTrace/box",
|
||
json={"zongpai_no": "26BW0011", "box_no": 935, "quantity": 10},
|
||
)
|
||
resp = client.post(
|
||
"/CargoTrace/attachment/box",
|
||
json={
|
||
"zongpai_no": "26BW0011",
|
||
"category_id": cert_id,
|
||
"box_no": 935,
|
||
"quantity": 5,
|
||
},
|
||
)
|
||
assert resp.status_code == 200
|
||
info = client.get(
|
||
"/CargoTrace/box/info", params={"zongpai_no": "26BW0011"}
|
||
).json()
|
||
box = next(b for b in info["existing_boxes"] if b["box_no"] == 935)
|
||
kinds = {i["kind"] for i in box["items"]}
|
||
assert kinds == {"product", "attachment"}
|
||
|
||
|
||
def test_attachment_box_completion(client: TestClient, db: Session):
|
||
cert_id, _ = _seed(db)
|
||
client.put(
|
||
"/CargoTrace/attachment/config",
|
||
json={
|
||
"zongpai_no": "26BW0011",
|
||
"determination": "has",
|
||
"items": [{"category_id": cert_id, "expected_qty": 80}],
|
||
},
|
||
)
|
||
client.post(
|
||
"/CargoTrace/attachment/box",
|
||
json={
|
||
"zongpai_no": "26BW0011",
|
||
"category_id": cert_id,
|
||
"box_no": 936,
|
||
"quantity": 80,
|
||
},
|
||
)
|
||
status = client.get(
|
||
"/CargoTrace/attachment/status", params={"zongpai_no": "26BW0011"}
|
||
).json()
|
||
assert status["all_complete"] is True
|
||
|
||
|
||
def test_attachment_box_update_quantity(client: TestClient, db: Session):
|
||
cert_id, _ = _seed(db)
|
||
created = client.post(
|
||
"/CargoTrace/attachment/box",
|
||
json={
|
||
"zongpai_no": "26BW0011",
|
||
"category_id": cert_id,
|
||
"box_no": 941,
|
||
"quantity": 30,
|
||
},
|
||
).json()
|
||
item_id = created["box_item_id"]
|
||
updated = client.patch(
|
||
f"/CargoTrace/attachment/box/{item_id}",
|
||
json={"box_no": 941, "quantity": 25},
|
||
)
|
||
assert updated.status_code == 200
|
||
assert updated.json()["quantity"] == 25
|
||
|
||
|
||
def test_attachment_box_update_rejects_over_expected(client: TestClient, db: Session):
|
||
cert_id, _ = _seed(db)
|
||
client.put(
|
||
"/CargoTrace/attachment/config",
|
||
json={
|
||
"zongpai_no": "26BW0011",
|
||
"determination": "has",
|
||
"items": [{"category_id": cert_id, "expected_qty": 80}],
|
||
},
|
||
)
|
||
created = client.post(
|
||
"/CargoTrace/attachment/box",
|
||
json={
|
||
"zongpai_no": "26BW0011",
|
||
"category_id": cert_id,
|
||
"box_no": 942,
|
||
"quantity": 70,
|
||
},
|
||
).json()
|
||
over = client.patch(
|
||
f"/CargoTrace/attachment/box/{created['box_item_id']}",
|
||
json={"box_no": 942, "quantity": 81},
|
||
)
|
||
assert over.status_code == 400
|
||
assert over.json()["error_code"] == "INVALID_QUANTITY"
|
||
|
||
|
||
def test_attachment_box_delete_cleans_empty_box(client: TestClient, db: Session):
|
||
cert_id, _ = _seed(db)
|
||
created = client.post(
|
||
"/CargoTrace/attachment/box",
|
||
json={
|
||
"zongpai_no": "26BW0011",
|
||
"category_id": cert_id,
|
||
"box_no": 943,
|
||
"quantity": 10,
|
||
},
|
||
).json()
|
||
item_id = created["box_item_id"]
|
||
deleted = client.delete(f"/CargoTrace/attachment/box/{item_id}")
|
||
assert deleted.status_code == 200
|
||
assert deleted.json() == {"box_item_id": item_id, "deleted": True}
|
||
|
||
# Box was attachment-only; deleting it should also remove the empty box
|
||
# so the same box_no can be recreated for a different总排号.
|
||
recreated = client.post(
|
||
"/CargoTrace/attachment/box",
|
||
json={
|
||
"zongpai_no": "26BW0012",
|
||
"category_id": cert_id,
|
||
"box_no": 943,
|
||
"quantity": 5,
|
||
},
|
||
)
|
||
assert recreated.status_code == 200
|
||
|
||
|
||
def test_box_info_includes_attachment_summary(client: TestClient, db: Session):
|
||
cert_id, extra_id = _seed(db)
|
||
client.put(
|
||
"/CargoTrace/attachment/config",
|
||
json={
|
||
"zongpai_no": "26BW0011",
|
||
"determination": "has",
|
||
"items": [
|
||
{"category_id": cert_id, "expected_qty": 80},
|
||
{"category_id": extra_id, "expected_qty": 5},
|
||
],
|
||
},
|
||
)
|
||
client.post(
|
||
"/CargoTrace/attachment/box",
|
||
json={
|
||
"zongpai_no": "26BW0011",
|
||
"category_id": cert_id,
|
||
"box_no": 951,
|
||
"quantity": 80,
|
||
},
|
||
)
|
||
info = client.get(
|
||
"/CargoTrace/box/info", params={"zongpai_no": "26BW0011"}
|
||
).json()
|
||
assert info["attachment_summary"]["determination"] == "has"
|
||
assert info["attachment_summary"]["all_complete"] is False
|
||
assert info["attachment_summary"]["pending_count"] == 1 # extra not yet boxed
|