From afc540005c335a496628c4c6f53f97420df91316 Mon Sep 17 00:00:00 2001 From: Misaka_Company Date: Wed, 13 May 2026 12:43:17 +0800 Subject: [PATCH] feat: support off-shelf location registration --- app/api/v1/location.py | 3 +- app/main.py | 15 ++++++- app/schemas/location.py | 1 + app/services/location_service.py | 29 +++++++++++++ tests/conftest.py | 10 ++++- tests/test_location_api.py | 70 +++++++++++++++++++++++++++++++- 6 files changed, 123 insertions(+), 5 deletions(-) diff --git a/app/api/v1/location.py b/app/api/v1/location.py index 0e5b572..f068e0a 100644 --- a/app/api/v1/location.py +++ b/app/api/v1/location.py @@ -18,7 +18,7 @@ router = APIRouter() "model": ErrorResponse, }, 409: { - "description": "该总排号已存在货位记录", + "description": "重复上架或已下架", "model": ErrorResponse, }, }, @@ -29,4 +29,5 @@ def create_location(req: LocationRequest, db: Session = Depends(get_db)): zongpai_no=record.zongpai_no, location_code=record.location_code, created_at=record.created_at.isoformat(), + previous_location=getattr(record, "previous_location", None), ) diff --git a/app/main.py b/app/main.py index 56330c2..aa41dda 100644 --- a/app/main.py +++ b/app/main.py @@ -18,7 +18,7 @@ except Exception as e: from app.api.v1.location import router as location_router from app.api.v1.box import router as box_router -from app.services.location_service import DuplicateLocationError +from app.services.location_service import AlreadyOffShelfError, DuplicateLocationError from app.services.box_service import ( BoxItemNotFoundError, DuplicateBoxItemError, @@ -47,6 +47,19 @@ async def duplicate_location_handler(request: Request, exc: DuplicateLocationErr ) +@app.exception_handler(AlreadyOffShelfError) +async def already_off_shelf_handler(request: Request, exc: AlreadyOffShelfError): + return JSONResponse( + status_code=409, + content={ + "error_code": "ALREADY_OFF_SHELF", + "message": "该总排号已下架至转运区域，不可重新上架", + "location_code": exc.location_code, + "registered_at": exc.registered_at, + }, + ) + + @app.exception_handler(RequestValidationError) async def validation_error_handler(request: Request, exc: RequestValidationError): for err in exc.errors(): diff --git a/app/schemas/location.py b/app/schemas/location.py index a4af99a..4b9af67 100644 --- a/app/schemas/location.py +++ b/app/schemas/location.py @@ -32,6 +32,7 @@ class LocationResponse(BaseModel): zongpai_no: str location_code: str created_at: str + previous_location: str | None = None class DuplicateLocationDetail(BaseModel): diff --git a/app/services/location_service.py b/app/services/location_service.py index cc69849..bd9f330 100644 --- a/app/services/location_service.py +++ b/app/services/location_service.py @@ -1,3 +1,5 @@ +from datetime import datetime + from sqlalchemy.orm import Session from app.models.finished_goods import FinishedGoodsLocation @@ -12,6 +14,18 @@ class DuplicateLocationError(Exception): self.registered_at = registered_at +class AlreadyOffShelfError(Exception): + """总排号已下架至转运区域。""" + + def __init__(self, location_code: str, registered_at: str): + self.location_code = location_code + self.registered_at = registered_at + + +def _is_transit_location(location_code: str) -> bool: + return location_code.startswith("TRANS-") + + def register_location(db: Session, req: LocationRequest) -> FinishedGoodsLocation: existing = ( db.query(FinishedGoodsLocation) @@ -19,6 +33,21 @@ def register_location(db: Session, req: LocationRequest) -> FinishedGoodsLocatio .first() ) if existing: + if _is_transit_location(existing.location_code): + raise AlreadyOffShelfError( + location_code=existing.location_code, + registered_at=existing.created_at.isoformat(), + ) + + if _is_transit_location(req.location_code): + previous_location = existing.location_code + existing.location_code = req.location_code + existing.created_at = datetime.now() + db.commit() + db.refresh(existing) + existing.previous_location = previous_location + return existing + raise DuplicateLocationError( location_code=existing.location_code, registered_at=existing.created_at.isoformat(), diff --git a/tests/conftest.py b/tests/conftest.py index 873ecd8..1a76cbb 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -22,7 +22,15 @@ def db(): @pytest.fixture(autouse=True) def cleanup_test_data(db: Session): yield - test_zongpai_nos = ["26B999", "26C999", "26BW0999", "26T999"] + test_zongpai_nos = [ + "26B999", + "26C999", + "26BW0999", + "26T999", + "26T998", + "26B998", + "26C998", + ] db.query(FinishedGoodsLocation).filter( FinishedGoodsLocation.zongpai_no.in_(test_zongpai_nos) ).delete(synchronize_session=False) diff --git a/tests/test_location_api.py b/tests/test_location_api.py index 0fb7732..80234d4 100644 --- a/tests/test_location_api.py +++ b/tests/test_location_api.py @@ -1,5 +1,8 @@ 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""" @@ -13,8 +16,8 @@ def test_register_location_success(client: TestClient): assert data["location_code"] == "A01-02-03" -def test_register_location_duplicate(client: TestClient): - """重复上架 — 应返回 409，包含统一错误格式""" +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"}, @@ -30,6 +33,69 @@ def test_register_location_duplicate(client: TestClient): 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(