feat(box): simplify boxing mode API

app/api/v1/box.py

@@ -2,8 +2,20 @@ from fastapi import APIRouter, Depends, Query
 from sqlalchemy.orm import Session
 
 from app.core.database import get_db
-from app.schemas.box import BoxInfoResponse, BoxSaveRequest, BoxSaveResponse
+from app.schemas.box import (
-from app.services.box_service import get_box_info, save_box_record
+    BoxDeleteResponse,
+    BoxInfoResponse,
+    BoxSaveRequest,
+    BoxSaveResponse,
+    BoxUpdateRequest,
+    BoxUpdateResponse,
+)
+from app.services.box_service import (
+    delete_box_item,
+    get_box_info,
+    save_box_record,
+    update_box_item,
+)
 
 router = APIRouter(tags=["boxing"])
 
@@ -21,3 +33,15 @@ def box_info(
 def create_box(req: BoxSaveRequest, db: Session = Depends(get_db)):
     """保存装箱记录：将总排号记录到指定箱号中。"""
     return save_box_record(db, req)
+
+
+@router.patch("/box/{box_item_id}", response_model=BoxUpdateResponse)
+def update_box(box_item_id: int, req: BoxUpdateRequest, db: Session = Depends(get_db)):
+    """更新已分配装箱明细。"""
+    return update_box_item(db, box_item_id, req)
+
+
+@router.delete("/box/{box_item_id}", response_model=BoxDeleteResponse)
+def delete_box(box_item_id: int, db: Session = Depends(get_db)):
+    """删除误录的装箱明细。"""
+    return delete_box_item(db, box_item_id)

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),
     )

app/main.py

@@ -18,8 +18,15 @@ 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 ZongpaiNotFoundError, DuplicateBoxItemError, InvalidZongpaiError, DuplicateZongpaiBindError
+from app.services.box_service import (
+    BoxItemNotFoundError,
+    DuplicateBoxItemError,
+    InvalidBoxItemError,
+    InvalidQuantityError,
+    InvalidZongpaiError,
+    ZongpaiNotFoundError,
+)
 
 app = FastAPI(title="CargoTrace API", version="0.1.0")
 
@@ -40,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():
@@ -106,16 +126,35 @@ async def duplicate_box_handler(request: Request, exc: DuplicateBoxItemError):
     )
 
 
-@app.exception_handler(DuplicateZongpaiBindError)
+@app.exception_handler(BoxItemNotFoundError)
-async def duplicate_zongpai_bind_handler(request: Request, exc: DuplicateZongpaiBindError):
+async def box_item_not_found_handler(request: Request, exc: BoxItemNotFoundError):
     return JSONResponse(
-        status_code=409,
+        status_code=404,
         content={
-            "error_code": "DUPLICATE_ZONGPAI_BIND",
+            "error_code": "BOX_ITEM_NOT_FOUND",
-            "message": f"总排号 {exc.zongpai_no} 已绑定箱号 {exc.box_no}，请勿重复装箱",
+            "message": "指定装箱明细不存在",
-            "paichan_no": exc.paichan_no,
+        },
-            "box_no": exc.box_no,
+    )
-            "zongpai_no": exc.zongpai_no,
+
+
+@app.exception_handler(InvalidQuantityError)
+async def invalid_quantity_handler(request: Request, exc: InvalidQuantityError):
+    return JSONResponse(
+        status_code=400,
+        content={
+            "error_code": "INVALID_QUANTITY",
+            "message": "装箱数量超出可装数量上限",
+        },
+    )
+
+
+@app.exception_handler(InvalidBoxItemError)
+async def invalid_box_item_handler(request: Request, exc: InvalidBoxItemError):
+    return JSONResponse(
+        status_code=400,
+        content={
+            "error_code": "INVALID_BOX_ITEM",
+            "message": "装箱明细记录不合法或不允许修改",
         },
     )
 

app/schemas/box.py

@@ -1,15 +1,24 @@
 import re
 
-from pydantic import BaseModel, field_validator
+from pydantic import BaseModel, Field, field_validator
 
 ZONGPAI_PATTERN = re.compile(r"^(\d{2}(B|C|T)\d+|\d{2}(BW|CW)\d{4})$")
 
 
 class BoxItemDetail(BaseModel):
     """箱号下某个总排号的明细"""
+    box_item_id: int | None = None
     zongpai_no: str
     work_order_no: str | None = None
     quantity: int
+    total_quantity: int | None = None
+
+
+class CurrentZongpaiBox(BaseModel):
+    """当前总排号已经分配的箱号明细"""
+    box_item_id: int
+    box_no: int
+    quantity: int
 
 
 class BoxDetail(BaseModel):
@@ -24,6 +33,7 @@ class BoxInfoResponse(BaseModel):
     paichan_no: str
     work_order_no: str | None = None
     quantity: int
+    current_zongpai_boxes: list[CurrentZongpaiBox] = Field(default_factory=list)
     existing_boxes: list[BoxDetail]
     max_box_no: int
     suggested_box_no: int
@@ -34,7 +44,6 @@ class BoxSaveRequest(BaseModel):
     zongpai_no: str
     box_no: int
     quantity: int
-    box_mode: str | None = None  # one-to-one | one-to-many | many-to-one
 
     @field_validator("zongpai_no")
     @classmethod
@@ -61,8 +70,45 @@ class BoxSaveRequest(BaseModel):
 
 class BoxSaveResponse(BaseModel):
     """POST /box 响应"""
+    box_item_id: int
     paichan_no: str
     box_no: int
     zongpai_no: str
     quantity: int
     created_at: str
+
+
+class BoxUpdateRequest(BaseModel):
+    """PATCH /box/{box_item_id} 请求"""
+    box_no: int
+    quantity: int
+
+    @field_validator("box_no")
+    @classmethod
+    def validate_box_no(cls, v: int) -> int:
+        if v <= 0:
+            raise ValueError("INVALID_BOX_NO")
+        return v
+
+    @field_validator("quantity")
+    @classmethod
+    def validate_quantity(cls, v: int) -> int:
+        if v <= 0:
+            raise ValueError("INVALID_QUANTITY")
+        return v
+
+
+class BoxUpdateResponse(BaseModel):
+    """PATCH /box/{box_item_id} 响应"""
+    box_item_id: int
+    paichan_no: str
+    box_no: int
+    zongpai_no: str
+    quantity: int
+    updated_at: str
+
+
+class BoxDeleteResponse(BaseModel):
+    """DELETE /box/{box_item_id} 响应"""
+    box_item_id: int
+    deleted: bool

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):

app/services/box_service.py

@@ -4,7 +4,7 @@ from sqlalchemy import bindparam, text
 from sqlalchemy.orm import Session
 
 from app.models.finished_goods import FinishedGoodsBox, FinishedGoodsBoxItem
-from app.schemas.box import BoxSaveRequest
+from app.schemas.box import BoxSaveRequest, BoxUpdateRequest
 
 ZONGPAI_PATTERN = re.compile(r"^(\d{2}(B|C|T)\d+|\d{2}(BW|CW)\d{4})$")
 
@@ -23,11 +23,16 @@ class DuplicateBoxItemError(Exception):
         self.box_no = box_no
 
 
-class DuplicateZongpaiBindError(Exception):
+class BoxItemNotFoundError(Exception):
-    def __init__(self, paichan_no: str, box_no: int, zongpai_no: str):
+    pass
-        self.paichan_no = paichan_no
+
-        self.box_no = box_no
+
-        self.zongpai_no = zongpai_no
+class InvalidBoxItemError(Exception):
+    pass
+
+
+class InvalidQuantityError(Exception):
+    pass
 
 
 def query_erp_info(db: Session, zongpai_no: str) -> dict:
@@ -60,21 +65,21 @@ def _erp_info_sql(db: Session) -> str:
     )
 
 
-def query_erp_work_order_map(db: Session, zongpai_nos: list[str]) -> dict[str, str | None]:
+def query_erp_item_info_map(db: Session, zongpai_nos: list[str]) -> dict[str, dict]:
-    """批量查询总排号对应的工令号。"""
+    """批量查询总排号对应的工令号和 ERP 总数量。"""
     if not zongpai_nos:
         return {}
 
     dialect_name = db.bind.dialect.name if db.bind is not None else ""
     if dialect_name == "postgresql":
         sql = text(
-            'SELECT "总排号", "工令号" '
+            'SELECT "总排号", "工令号", "数量" '
             'FROM "ERPAuto"."vw_productionContractData" '
             'WHERE "总排号" IN :zongpai_nos'
         )
     else:
         sql = text(
-            "SELECT [总排号], [工令号] "
+            "SELECT [总排号], [工令号], [数量] "
             "FROM [ERPAuto].[vw_productionContractData] "
             "WHERE [总排号] IN :zongpai_nos"
         )
@@ -82,7 +87,13 @@ def query_erp_work_order_map(db: Session, zongpai_nos: list[str]) -> dict[str, s
         sql.bindparams(bindparam("zongpai_nos", expanding=True)),
         {"zongpai_nos": sorted(set(zongpai_nos))},
     ).fetchall()
-    return {row[0]: row[1] for row in rows}
+    return {
+        row[0]: {
+            "work_order_no": row[1],
+            "total_quantity": int(row[2]) if row[2] is not None else None,
+        }
+        for row in rows
+    }
 
 
 def get_box_info(db: Session, zongpai_no: str) -> dict:
@@ -112,19 +123,33 @@ def get_box_info(db: Session, zongpai_no: str) -> dict:
         box_items[box.id] = items
         item_zongpai_nos.extend(item.zongpai_no for item in items)
 
-    work_order_map = query_erp_work_order_map(db, item_zongpai_nos)
+    erp_item_info_map = query_erp_item_info_map(db, item_zongpai_nos)
 
     existing_boxes = []
+    current_zongpai_boxes = []
     max_box_no = 0
     for box in boxes:
         items = box_items[box.id]
+        for item in items:
+            if item.zongpai_no == zongpai_no:
+                current_zongpai_boxes.append({
+                    "box_item_id": item.id,
+                    "box_no": box.box_no,
+                    "quantity": int(item.quantity or 0),
+                })
         existing_boxes.append({
             "box_no": box.box_no,
             "items": [
                 {
+                    "box_item_id": item.id,
                     "zongpai_no": item.zongpai_no,
-                    "work_order_no": work_order_map.get(item.zongpai_no),
+                    "work_order_no": erp_item_info_map.get(
+                        item.zongpai_no, {}
+                    ).get("work_order_no"),
                     "quantity": int(item.quantity or 0),
+                    "total_quantity": erp_item_info_map.get(
+                        item.zongpai_no, {}
+                    ).get("total_quantity"),
                 }
                 for item in items
             ],
@@ -137,30 +162,55 @@ def get_box_info(db: Session, zongpai_no: str) -> dict:
         "paichan_no": paichan_no,
         "work_order_no": erp["work_order_no"],
         "quantity": quantity,
+        "current_zongpai_boxes": current_zongpai_boxes,
         "existing_boxes": existing_boxes,
         "max_box_no": max_box_no,
         "suggested_box_no": max_box_no + 1,
     }
 
 
+def _packed_quantity_for_zongpai(
+    db: Session,
+    paichan_no: str,
+    zongpai_no: str,
+    exclude_box_item_id: int | None = None,
+) -> int:
+    query = (
+        db.query(FinishedGoodsBoxItem)
+        .join(FinishedGoodsBox, FinishedGoodsBoxItem.box_id == FinishedGoodsBox.id)
+        .filter(
+            FinishedGoodsBox.paichan_no == paichan_no,
+            FinishedGoodsBoxItem.zongpai_no == zongpai_no,
+        )
+    )
+    if exclude_box_item_id is not None:
+        query = query.filter(FinishedGoodsBoxItem.id != exclude_box_item_id)
+    return sum(int(item.quantity or 0) for item in query.all())
+
+
+def _ensure_quantity_within_erp_total(
+    db: Session,
+    paichan_no: str,
+    zongpai_no: str,
+    erp_quantity: int,
+    submitted_quantity: int,
+    exclude_box_item_id: int | None = None,
+) -> None:
+    packed_quantity = _packed_quantity_for_zongpai(
+        db,
+        paichan_no,
+        zongpai_no,
+        exclude_box_item_id=exclude_box_item_id,
+    )
+    if packed_quantity + submitted_quantity > erp_quantity:
+        raise InvalidQuantityError()
+
+
 def save_box_record(db: Session, req: BoxSaveRequest) -> dict:
-    """保存装箱记录。箱已存在时仅追加明细（多码一箱场景）。"""
+    """保存装箱记录。同箱可凑箱，但同箱同总排不可重复。"""
     erp = query_erp_info(db, req.zongpai_no)
     paichan_no = erp["paichan_no"]
 
-    if req.box_mode == "one-to-one":
-        existing_box_no = (
-            db.query(FinishedGoodsBox.box_no)
-            .join(FinishedGoodsBoxItem, FinishedGoodsBox.id == FinishedGoodsBoxItem.box_id)
-            .filter(
-                FinishedGoodsBox.paichan_no == paichan_no,
-                FinishedGoodsBoxItem.zongpai_no == req.zongpai_no,
-            )
-            .scalar()
-        )
-        if existing_box_no is not None:
-            raise DuplicateZongpaiBindError(paichan_no, existing_box_no, req.zongpai_no)
-
     box = (
         db.query(FinishedGoodsBox)
         .filter(
@@ -181,7 +231,16 @@ def save_box_record(db: Session, req: BoxSaveRequest) -> dict:
         )
         if existing_item:
             raise DuplicateBoxItemError(paichan_no, req.box_no)
-    else:
+
+    _ensure_quantity_within_erp_total(
+        db,
+        paichan_no,
+        req.zongpai_no,
+        erp["quantity"],
+        req.quantity,
+    )
+
+    if box is None:
         box = FinishedGoodsBox(
             paichan_no=paichan_no,
             box_no=req.box_no,
@@ -199,9 +258,128 @@ def save_box_record(db: Session, req: BoxSaveRequest) -> dict:
     db.refresh(item)
 
     return {
+        "box_item_id": item.id,
         "paichan_no": paichan_no,
         "box_no": req.box_no,
         "zongpai_no": req.zongpai_no,
         "quantity": req.quantity,
         "created_at": item.created_at.isoformat(),
     }
+
+
+def update_box_item(db: Session, box_item_id: int, req: BoxUpdateRequest) -> dict:
+    """更新已分配明细的箱号和数量。"""
+    item = (
+        db.query(FinishedGoodsBoxItem)
+        .filter(FinishedGoodsBoxItem.id == box_item_id)
+        .first()
+    )
+    if item is None:
+        raise BoxItemNotFoundError()
+
+    current_box = (
+        db.query(FinishedGoodsBox)
+        .filter(FinishedGoodsBox.id == item.box_id)
+        .first()
+    )
+    if current_box is None:
+        raise InvalidBoxItemError()
+
+    paichan_no = current_box.paichan_no
+    target_box = (
+        db.query(FinishedGoodsBox)
+        .filter(
+            FinishedGoodsBox.paichan_no == paichan_no,
+            FinishedGoodsBox.box_no == req.box_no,
+        )
+        .first()
+    )
+
+    if target_box is None:
+        target_box = FinishedGoodsBox(paichan_no=paichan_no, box_no=req.box_no)
+        db.add(target_box)
+        db.flush()
+    else:
+        duplicate_item = (
+            db.query(FinishedGoodsBoxItem)
+            .filter(
+                FinishedGoodsBoxItem.box_id == target_box.id,
+                FinishedGoodsBoxItem.zongpai_no == item.zongpai_no,
+                FinishedGoodsBoxItem.id != item.id,
+            )
+            .first()
+        )
+        if duplicate_item is not None:
+            raise DuplicateBoxItemError(paichan_no, req.box_no)
+
+    erp = query_erp_info(db, item.zongpai_no)
+    _ensure_quantity_within_erp_total(
+        db,
+        paichan_no,
+        item.zongpai_no,
+        erp["quantity"],
+        req.quantity,
+        exclude_box_item_id=box_item_id,
+    )
+
+    old_box_id = item.box_id
+    item.box_id = target_box.id
+    item.quantity = req.quantity
+
+    if old_box_id != target_box.id:
+        old_box_has_items = (
+            db.query(FinishedGoodsBoxItem.id)
+            .filter(
+                FinishedGoodsBoxItem.box_id == old_box_id,
+                FinishedGoodsBoxItem.id != item.id,
+            )
+            .first()
+        )
+        if old_box_has_items is None:
+            old_box = (
+                db.query(FinishedGoodsBox)
+                .filter(FinishedGoodsBox.id == old_box_id)
+                .first()
+            )
+            if old_box is not None:
+                db.delete(old_box)
+
+    db.commit()
+    db.refresh(item)
+
+    return {
+        "box_item_id": item.id,
+        "paichan_no": paichan_no,
+        "box_no": req.box_no,
+        "zongpai_no": item.zongpai_no,
+        "quantity": int(item.quantity or 0),
+        "updated_at": item.created_at.isoformat(),
+    }
+
+
+def delete_box_item(db: Session, box_item_id: int) -> dict:
+    """删除误录的装箱明细；若箱号下无其他明细，同步删除空箱号。"""
+    item = (
+        db.query(FinishedGoodsBoxItem)
+        .filter(FinishedGoodsBoxItem.id == box_item_id)
+        .first()
+    )
+    if item is None:
+        raise BoxItemNotFoundError()
+
+    box_id = item.box_id
+    db.delete(item)
+    db.flush()
+
+    remaining = (
+        db.query(FinishedGoodsBoxItem.id)
+        .filter(FinishedGoodsBoxItem.box_id == box_id)
+        .first()
+    )
+    if remaining is None:
+        box = db.query(FinishedGoodsBox).filter(FinishedGoodsBox.id == box_id).first()
+        if box is not None:
+            db.delete(box)
+
+    db.commit()
+    return {"box_item_id": box_item_id, "deleted": True}

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(),

tests/conftest.py

@@ -4,7 +4,11 @@ from sqlalchemy.orm import Session
 
 from app.core.database import SessionLocal
 from app.main import app
-from app.models.finished_goods import FinishedGoodsLocation
+from app.models.finished_goods import (
+    FinishedGoodsBox,
+    FinishedGoodsBoxItem,
+    FinishedGoodsLocation,
+)
 
 
 @pytest.fixture
@@ -22,8 +26,32 @@ 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)
+    test_boxes = (
+        db.query(FinishedGoodsBox)
+        .filter(
+            FinishedGoodsBox.paichan_no == "W00009",
+            FinishedGoodsBox.box_no >= 900,
+        )
+        .all()
+    )
+    test_box_ids = [box.id for box in test_boxes]
+    if test_box_ids:
+        db.query(FinishedGoodsBoxItem).filter(
+            FinishedGoodsBoxItem.box_id.in_(test_box_ids)
+        ).delete(synchronize_session=False)
+        db.query(FinishedGoodsBox).filter(
+            FinishedGoodsBox.id.in_(test_box_ids)
+        ).delete(synchronize_session=False)
     db.commit()

tests/test_box_api.py

@@ -16,7 +16,11 @@ def test_box_info_success(client: TestClient):
     assert "work_order_no" in data
     assert data["work_order_no"] is None or isinstance(data["work_order_no"], str)
     assert data["quantity"] == 80
+    assert "current_zongpai_boxes" in data
     assert "existing_boxes" in data
+    for box in data["existing_boxes"]:
+        for item in box["items"]:
+            assert "box_item_id" in item
     assert "max_box_no" in data
     assert data["suggested_box_no"] == data["max_box_no"] + 1
 
@@ -78,3 +82,182 @@ def test_box_save_zongpai_not_found(client: TestClient):
     )
     assert resp.status_code == 404
     assert resp.json()["error_code"] == "ZONGPAI_NOT_FOUND"
+
+
+def test_box_save_multi_code_appends_different_zongpais(client: TestClient):
+    """多码凑箱：同一排产号同一箱号可追加不同总排号。"""
+    box_no = 901
+    first = client.post(
+        "/CargoTrace/box",
+        json={
+            "zongpai_no": "26BW0011",
+            "box_no": box_no,
+            "quantity": 10,
+        },
+    )
+    assert first.status_code == 200
+    assert first.json()["box_item_id"] is not None
+
+    second = client.post(
+        "/CargoTrace/box",
+        json={
+            "zongpai_no": "26BW0012",
+            "box_no": box_no,
+            "quantity": 8,
+        },
+    )
+    assert second.status_code == 200
+    assert second.json()["box_no"] == box_no
+    assert second.json()["zongpai_no"] == "26BW0012"
+
+    info = client.get("/CargoTrace/box/info", params={"zongpai_no": "26BW0011"})
+    assert info.status_code == 200
+    target_box = next(
+        box for box in info.json()["existing_boxes"] if box["box_no"] == box_no
+    )
+    assert {item["zongpai_no"] for item in target_box["items"]} == {
+        "26BW0011",
+        "26BW0012",
+    }
+    for item in target_box["items"]:
+        assert "work_order_no" in item
+        assert isinstance(item["total_quantity"], int)
+
+
+def test_box_save_multi_code_duplicate_same_zongpai(client: TestClient):
+    """多码凑箱：同箱同总排重复录入返回 DUPLICATE_BOX_NO。"""
+    box_no = 902
+    first = client.post(
+        "/CargoTrace/box",
+        json={
+            "zongpai_no": "26BW0011",
+            "box_no": box_no,
+            "quantity": 10,
+        },
+    )
+    assert first.status_code == 200
+
+    duplicate = client.post(
+        "/CargoTrace/box",
+        json={
+            "zongpai_no": "26BW0011",
+            "box_no": box_no,
+            "quantity": 5,
+        },
+    )
+    assert duplicate.status_code == 409
+    assert duplicate.json()["error_code"] == "DUPLICATE_BOX_NO"
+
+
+def test_box_save_split_same_zongpai_up_to_total_quantity(client: TestClient):
+    """单码装箱：同一总排号可分箱提交，累计等于 ERP 数量时成功。"""
+    first = client.post(
+        "/CargoTrace/box",
+        json={"zongpai_no": "26BW0011", "box_no": 905, "quantity": 30},
+    )
+    assert first.status_code == 200
+
+    second = client.post(
+        "/CargoTrace/box",
+        json={"zongpai_no": "26BW0011", "box_no": 906, "quantity": 50},
+    )
+    assert second.status_code == 200
+
+    info = client.get("/CargoTrace/box/info", params={"zongpai_no": "26BW0011"})
+    assert info.status_code == 200
+    current_boxes = [
+        item
+        for item in info.json()["current_zongpai_boxes"]
+        if item["box_no"] in (905, 906)
+    ]
+    assert sum(item["quantity"] for item in current_boxes) == 80
+
+
+def test_box_save_rejects_quantity_over_remaining(client: TestClient):
+    """保存后累计数量超过 ERP 总数量时返回 INVALID_QUANTITY。"""
+    created = client.post(
+        "/CargoTrace/box",
+        json={"zongpai_no": "26BW0011", "box_no": 907, "quantity": 70},
+    )
+    assert created.status_code == 200
+
+    over = client.post(
+        "/CargoTrace/box",
+        json={"zongpai_no": "26BW0011", "box_no": 908, "quantity": 11},
+    )
+    assert over.status_code == 400
+    assert over.json()["error_code"] == "INVALID_QUANTITY"
+
+
+def test_box_update_same_box_no_does_not_conflict(client: TestClient):
+    """PATCH 原箱号未变时，应排除当前 box_item_id，避免误判重复。"""
+    box_no = 903
+    created = client.post(
+        "/CargoTrace/box",
+        json={
+            "zongpai_no": "26BW0011",
+            "box_no": box_no,
+            "quantity": 10,
+        },
+    )
+    assert created.status_code == 200
+    box_item_id = created.json()["box_item_id"]
+
+    updated = client.patch(
+        f"/CargoTrace/box/{box_item_id}",
+        json={"box_no": box_no, "quantity": 7},
+    )
+    assert updated.status_code == 200
+    assert updated.json()["quantity"] == 7
+
+
+def test_box_update_rejects_quantity_over_remaining(client: TestClient):
+    """PATCH 调整数量导致累计超过 ERP 总数量时返回 INVALID_QUANTITY。"""
+    first = client.post(
+        "/CargoTrace/box",
+        json={"zongpai_no": "26BW0011", "box_no": 909, "quantity": 70},
+    )
+    assert first.status_code == 200
+
+    second = client.post(
+        "/CargoTrace/box",
+        json={"zongpai_no": "26BW0011", "box_no": 910, "quantity": 5},
+    )
+    assert second.status_code == 200
+    box_item_id = second.json()["box_item_id"]
+
+    over = client.patch(
+        f"/CargoTrace/box/{box_item_id}",
+        json={"box_no": 910, "quantity": 11},
+    )
+    assert over.status_code == 400
+    assert over.json()["error_code"] == "INVALID_QUANTITY"
+
+
+def test_box_delete_removes_empty_box(client: TestClient):
+    """DELETE 删除最后一条明细后，同步清理空箱号。"""
+    box_no = 904
+    created = client.post(
+        "/CargoTrace/box",
+        json={
+            "zongpai_no": "26BW0011",
+            "box_no": box_no,
+            "quantity": 10,
+        },
+    )
+    assert created.status_code == 200
+    box_item_id = created.json()["box_item_id"]
+
+    deleted = client.delete(f"/CargoTrace/box/{box_item_id}")
+    assert deleted.status_code == 200
+    assert deleted.json() == {"box_item_id": box_item_id, "deleted": True}
+
+    recreated = client.post(
+        "/CargoTrace/box",
+        json={
+            "zongpai_no": "26BW0012",
+            "box_no": box_no,
+            "quantity": 8,
+        },
+    )
+    assert recreated.status_code == 200

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):
+def test_register_location_duplicate_normal_to_normal(client: TestClient):
-    """重复上架 — 应返回 409，包含统一错误格式"""
+    """普通货位重复上架到普通货位 — 应返回 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(