127 lines
4.2 KiB
Python
127 lines
4.2 KiB
Python
from fastapi.testclient import TestClient
|
|
|
|
from app.core.database import SessionLocal
|
|
from app.models.finished_goods import FinishedGoodsLocation
|
|
|
|
|
|
def test_register_location_success(client: TestClient):
|
|
"""正常上架 — 应返回 200"""
|
|
resp = client.post(
|
|
"/CargoTrace/location",
|
|
json={"zongpai_no": "26B999", "location_code": "A01-02-03"},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["zongpai_no"] == "26B999"
|
|
assert data["location_code"] == "A01-02-03"
|
|
|
|
|
|
def test_register_location_duplicate_normal_to_normal(client: TestClient):
|
|
"""普通货位重复上架到普通货位 — 应返回 DUPLICATE_LOCATION"""
|
|
client.post(
|
|
"/CargoTrace/location",
|
|
json={"zongpai_no": "26C999", "location_code": "A01-02-03"},
|
|
)
|
|
resp = client.post(
|
|
"/CargoTrace/location",
|
|
json={"zongpai_no": "26C999", "location_code": "A02-01-01"},
|
|
)
|
|
assert resp.status_code == 409
|
|
data = resp.json()
|
|
assert data["error_code"] == "DUPLICATE_LOCATION"
|
|
assert data["location_code"] == "A01-02-03"
|
|
assert "registered_at" in data
|
|
|
|
|
|
def test_register_location_normal_to_transit_updates_record(client: TestClient):
|
|
"""普通货位下架到转运货位 — 应更新记录并返回 previous_location"""
|
|
client.post(
|
|
"/CargoTrace/location",
|
|
json={"zongpai_no": "26T998", "location_code": "A01-02-03"},
|
|
)
|
|
resp = client.post(
|
|
"/CargoTrace/location",
|
|
json={"zongpai_no": "26T998", "location_code": "TRANS-01"},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["zongpai_no"] == "26T998"
|
|
assert data["location_code"] == "TRANS-01"
|
|
assert data["previous_location"] == "A01-02-03"
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
record = (
|
|
db.query(FinishedGoodsLocation)
|
|
.filter(FinishedGoodsLocation.zongpai_no == "26T998")
|
|
.first()
|
|
)
|
|
assert record is not None
|
|
assert record.location_code == "TRANS-01"
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_register_location_transit_to_normal_rejected(client: TestClient):
|
|
"""已下架后再上架到普通货位 — 应返回 ALREADY_OFF_SHELF"""
|
|
client.post(
|
|
"/CargoTrace/location",
|
|
json={"zongpai_no": "26B998", "location_code": "TRANS-01"},
|
|
)
|
|
resp = client.post(
|
|
"/CargoTrace/location",
|
|
json={"zongpai_no": "26B998", "location_code": "A01-02-03"},
|
|
)
|
|
assert resp.status_code == 409
|
|
data = resp.json()
|
|
assert data["error_code"] == "ALREADY_OFF_SHELF"
|
|
assert data["message"] == "该总排号已下架至转运区域，不可重新上架"
|
|
assert data["location_code"] == "TRANS-01"
|
|
assert "registered_at" in data
|
|
|
|
|
|
def test_register_location_transit_to_transit_rejected(client: TestClient):
|
|
"""已下架后再提交转运货位 — 应返回 ALREADY_OFF_SHELF"""
|
|
client.post(
|
|
"/CargoTrace/location",
|
|
json={"zongpai_no": "26C998", "location_code": "TRANS-01"},
|
|
)
|
|
resp = client.post(
|
|
"/CargoTrace/location",
|
|
json={"zongpai_no": "26C998", "location_code": "TRANS-02"},
|
|
)
|
|
assert resp.status_code == 409
|
|
data = resp.json()
|
|
assert data["error_code"] == "ALREADY_OFF_SHELF"
|
|
assert data["location_code"] == "TRANS-01"
|
|
|
|
|
|
def test_register_location_invalid_zongpai(client: TestClient):
|
|
"""无效总排号 — 应返回 400"""
|
|
resp = client.post(
|
|
"/CargoTrace/location",
|
|
json={"zongpai_no": "INVALID", "location_code": "A01-02-03"},
|
|
)
|
|
assert resp.status_code == 400
|
|
assert resp.json()["error_code"] == "INVALID_ZONGPAI"
|
|
|
|
|
|
def test_register_location_invalid_location(client: TestClient):
|
|
"""无效货位号 — 应返回 400"""
|
|
resp = client.post(
|
|
"/CargoTrace/location",
|
|
json={"zongpai_no": "26T999", "location_code": "bad-location"},
|
|
)
|
|
assert resp.status_code == 400
|
|
assert resp.json()["error_code"] == "INVALID_LOCATION"
|
|
|
|
|
|
def test_register_location_trans_location(client: TestClient):
|
|
"""转运特殊货位 — 应返回 200"""
|
|
resp = client.post(
|
|
"/CargoTrace/location",
|
|
json={"zongpai_no": "26BW0999", "location_code": "TRANS-01"},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["location_code"] == "TRANS-01"
|