diff --git a/app/api/v1/box.py b/app/api/v1/box.py new file mode 100644 index 0000000..0855104 --- /dev/null +++ b/app/api/v1/box.py @@ -0,0 +1,23 @@ +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.services.box_service import get_box_info, save_box_record + +router = APIRouter(tags=["boxing"]) + + +@router.get("/box/info", response_model=BoxInfoResponse) +def box_info( + zongpai_no: str = Query(..., description="总排号"), + db: Session = Depends(get_db), +): + """查询装箱信息：根据总排号查询排产号、数量和已有箱号明细。""" + return get_box_info(db, zongpai_no) + + +@router.post("/box", response_model=BoxSaveResponse) +def create_box(req: BoxSaveRequest, db: Session = Depends(get_db)): + """保存装箱记录：将总排号记录到指定箱号中。""" + return save_box_record(db, req) diff --git a/app/main.py b/app/main.py index 7e48053..bccd4dd 100644 --- a/app/main.py +++ b/app/main.py @@ -3,11 +3,14 @@ from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse 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.box_service import ZongpaiNotFoundError, DuplicateBoxItemError, InvalidZongpaiError app = FastAPI(title="CargoTrace API", version="0.1.0") app.include_router(location_router, prefix="/CargoTrace") +app.include_router(box_router, prefix="/CargoTrace") @app.exception_handler(DuplicateLocationError) @@ -27,7 +30,7 @@ async def duplicate_location_handler(request: Request, exc: DuplicateLocationErr async def validation_error_handler(request: Request, exc: RequestValidationError): for err in exc.errors(): msg = err.get("msg", "") - if "INVALID_ZONGPAI" in msg or "INVALID_LOCATION" in msg: + if "INVALID_ZONGPAI" in msg or "INVALID_LOCATION" in msg or "INVALID_BOX_NO" in msg or "INVALID_QUANTITY" in msg: error_code = msg.replace("Value error, ", "") return JSONResponse( status_code=400, @@ -54,6 +57,41 @@ async def value_error_handler(request: Request, exc: ValueError): ) +@app.exception_handler(InvalidZongpaiError) +async def invalid_zongpai_handler(request: Request, exc: InvalidZongpaiError): + return JSONResponse( + status_code=400, + content={ + "error_code": "INVALID_ZONGPAI", + "message": "总排号格式不合法", + }, + ) + + +@app.exception_handler(ZongpaiNotFoundError) +async def zongpai_not_found_handler(request: Request, exc: ZongpaiNotFoundError): + return JSONResponse( + status_code=404, + content={ + "error_code": "ZONGPAI_NOT_FOUND", + "message": "未找到该总排号对应的排产号信息", + }, + ) + + +@app.exception_handler(DuplicateBoxItemError) +async def duplicate_box_handler(request: Request, exc: DuplicateBoxItemError): + return JSONResponse( + status_code=409, + content={ + "error_code": "DUPLICATE_BOX_NO", + "message": f"排产号 {exc.paichan_no} 下箱号 {exc.box_no} 已存在", + "paichan_no": exc.paichan_no, + "box_no": exc.box_no, + }, + ) + + @app.get("/") async def root(): return {"message": "Welcome to CargoTrace API"} diff --git a/app/schemas/box.py b/app/schemas/box.py new file mode 100644 index 0000000..8cf0b48 --- /dev/null +++ b/app/schemas/box.py @@ -0,0 +1,65 @@ +import re + +from pydantic import BaseModel, field_validator + +ZONGPAI_PATTERN = re.compile(r"^(\d{2}(B|C|T)\d+|\d{2}(BW|CW)\d{4})$") + + +class BoxItemDetail(BaseModel): + """箱号下某个总排号的明细""" + zongpai_no: str + quantity: int + + +class BoxDetail(BaseModel): + """一个箱号的完整信息""" + box_no: int + items: list[BoxItemDetail] + + +class BoxInfoResponse(BaseModel): + """GET /box/info 响应""" + zongpai_no: str + paichan_no: str + quantity: int + existing_boxes: list[BoxDetail] + max_box_no: int + suggested_box_no: int + + +class BoxSaveRequest(BaseModel): + """POST /box 请求""" + zongpai_no: str + box_no: int + quantity: int + + @field_validator("zongpai_no") + @classmethod + def validate_zongpai_no(cls, v: str) -> str: + v = v.strip().upper() + if not ZONGPAI_PATTERN.match(v): + raise ValueError("INVALID_ZONGPAI") + return v + + @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 BoxSaveResponse(BaseModel): + """POST /box 响应""" + paichan_no: str + box_no: int + zongpai_no: str + quantity: int + created_at: str diff --git a/app/services/box_service.py b/app/services/box_service.py new file mode 100644 index 0000000..d9d5f21 --- /dev/null +++ b/app/services/box_service.py @@ -0,0 +1,135 @@ +import re + +from sqlalchemy import text +from sqlalchemy.orm import Session + +from app.models.finished_goods import FinishedGoodsBox, FinishedGoodsBoxItem +from app.schemas.box import BoxSaveRequest + +ZONGPAI_PATTERN = re.compile(r"^(\d{2}(B|C|T)\d+|\d{2}(BW|CW)\d{4})$") + + +class InvalidZongpaiError(Exception): + pass + + +class ZongpaiNotFoundError(Exception): + pass + + +class DuplicateBoxItemError(Exception): + def __init__(self, paichan_no: str, box_no: int): + self.paichan_no = paichan_no + self.box_no = box_no + + +def query_erp_info(db: Session, zongpai_no: str) -> dict: + """从 ERP 视图查询总排号对应的排产号和数量。""" + sql = text( + "SELECT TOP 1 [总排号], [排产号], [数量] " + "FROM [ERPAuto].[vw_productionContractData] " + "WHERE [总排号] = :zongpai_no" + ) + row = db.execute(sql, {"zongpai_no": zongpai_no}).fetchone() + if not row: + raise ZongpaiNotFoundError() + return { + "zongpai_no": row[0], + "paichan_no": row[1], + "quantity": int(row[2]), + } + + +def get_box_info(db: Session, zongpai_no: str) -> dict: + """获取装箱信息：排产号、数量、已有箱号明细。""" + zongpai_no = zongpai_no.strip().upper() + if not ZONGPAI_PATTERN.match(zongpai_no): + raise InvalidZongpaiError() + erp = query_erp_info(db, zongpai_no) + paichan_no = erp["paichan_no"] + quantity = erp["quantity"] + + boxes = ( + db.query(FinishedGoodsBox) + .filter(FinishedGoodsBox.paichan_no == paichan_no) + .order_by(FinishedGoodsBox.box_no) + .all() + ) + + existing_boxes = [] + max_box_no = 0 + for box in boxes: + items = ( + db.query(FinishedGoodsBoxItem) + .filter(FinishedGoodsBoxItem.box_id == box.id) + .all() + ) + existing_boxes.append({ + "box_no": box.box_no, + "items": [ + {"zongpai_no": item.zongpai_no, "quantity": int(item.quantity or 0)} + for item in items + ], + }) + if box.box_no > max_box_no: + max_box_no = box.box_no + + return { + "zongpai_no": zongpai_no, + "paichan_no": paichan_no, + "quantity": quantity, + "existing_boxes": existing_boxes, + "max_box_no": max_box_no, + "suggested_box_no": max_box_no + 1, + } + + +def save_box_record(db: Session, req: BoxSaveRequest) -> dict: + """保存装箱记录。箱已存在时仅追加明细（多码一箱场景）。""" + erp = query_erp_info(db, req.zongpai_no) + paichan_no = erp["paichan_no"] + + box = ( + db.query(FinishedGoodsBox) + .filter( + FinishedGoodsBox.paichan_no == paichan_no, + FinishedGoodsBox.box_no == req.box_no, + ) + .first() + ) + + if box: + existing_item = ( + db.query(FinishedGoodsBoxItem) + .filter( + FinishedGoodsBoxItem.box_id == box.id, + FinishedGoodsBoxItem.zongpai_no == req.zongpai_no, + ) + .first() + ) + if existing_item: + raise DuplicateBoxItemError(paichan_no, req.box_no) + else: + box = FinishedGoodsBox( + paichan_no=paichan_no, + box_no=req.box_no, + ) + db.add(box) + db.flush() + + item = FinishedGoodsBoxItem( + box_id=box.id, + zongpai_no=req.zongpai_no, + quantity=req.quantity, + ) + db.add(item) + db.commit() + db.refresh(item) + + return { + "paichan_no": paichan_no, + "box_no": req.box_no, + "zongpai_no": req.zongpai_no, + "quantity": req.quantity, + "created_at": item.created_at.isoformat(), + } diff --git a/docs/plans/2026-05-11-boxing-module.md b/docs/plans/2026-05-11-boxing-module.md new file mode 100644 index 0000000..4108307 --- /dev/null +++ b/docs/plans/2026-05-11-boxing-module.md @@ -0,0 +1,507 @@ +# 装箱编号模块 (Boxing) Implementation Plan — FastAPI + +> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task. + +**Goal:** 实现 PRD v1.3 中定义的装箱编号模块后端接口：`GET /CargoTrace/box/info` 和 `POST /CargoTrace/box`，支持总排号查询排产号信息、已有箱号查询、以及装箱记录保存。 + +**Architecture:** 在现有三层架构（API → Service → Model）基础上扩展。ERP 视图数据通过 SQLAlchemy raw SQL 查询（跨 schema 访问 `[ERPAuto].[vw_productionContractData]`），CargoTrace 业务表通过 ORM 操作。 + +**Tech Stack:** Python 3.12+, FastAPI, SQLAlchemy 2.x (sync), pyodbc, pydantic + +--- + +## 数据流 + +``` +Flutter App + │ + ├── GET /CargoTrace/box/info?zongpai_no=xxx + │ └── 查询 ERP 视图 → 排产号 + 数量 + │ └── 查询 CargoTrace 表 → 已有箱号 + 明细 + │ └── 返回: paichan_no, quantity, existing_boxes, max_box_no, suggested_box_no + │ + └── POST /CargoTrace/box + └── 查询 ERP 视图 → 排产号 + └── 查找或创建 CargoTrace.finished_goods_box + └── 创建 CargoTrace.finished_goods_box_item + └── 返回: paichan_no, box_no, zongpai_no, quantity, created_at +``` + +## 关键设计决策 + +**多码一箱的 POST 逻辑：** +- 若 (paichan_no, box_no) 不存在 → 创建 box 记录 + 创建 box_item +- 若 (paichan_no, box_no) 已存在且该 zongpai_no 不在此箱中 → 仅创建 box_item（多码一箱场景） +- 若 (paichan_no, box_no) 已存在且该 zongpai_no 已在此箱中 → 返回 409 DUPLICATE_BOX_NO + +--- + +### Task 1: Schema 层 — 装箱请求/响应模型 + +**Files:** +- Create: `app/schemas/box.py` + +**Step 1: 创建 box.py** + +```python +import re + +from pydantic import BaseModel, field_validator + +ZONGPAI_PATTERN = re.compile(r"^(\d{2}(B|C|T)\d+|\d{2}(BW|CW)\d{4})$") + + +class BoxItemDetail(BaseModel): + """箱号下某个总排号的明细""" + zongpai_no: str + quantity: int + + +class BoxDetail(BaseModel): + """一个箱号的完整信息""" + box_no: int + items: list[BoxItemDetail] + + +class BoxInfoResponse(BaseModel): + """GET /box/info 响应""" + zongpai_no: str + paichan_no: str + quantity: int + existing_boxes: list[BoxDetail] + max_box_no: int + suggested_box_no: int + + +class BoxSaveRequest(BaseModel): + """POST /box 请求""" + zongpai_no: str + box_no: int + quantity: int + + @field_validator("zongpai_no") + @classmethod + def validate_zongpai_no(cls, v: str) -> str: + v = v.strip().upper() + if not ZONGPAI_PATTERN.match(v): + raise ValueError("INVALID_ZONGPAI") + return v + + @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 BoxSaveResponse(BaseModel): + """POST /box 响应""" + paichan_no: str + box_no: int + zongpai_no: str + quantity: int + created_at: str +``` + +**Step 2: 验证导入** + +Run: `cd apps/fastAPI && .venv/Scripts/python -c "from app.schemas.box import BoxInfoResponse, BoxSaveRequest; print('Schemas OK')"` + +**Step 3: Commit** + +```bash +git add app/schemas/box.py +git commit -m "feat: add box schemas for boxing module API" +``` + +--- + +### Task 2: Service 层 — 装箱业务逻辑 + +**Files:** +- Create: `app/services/box_service.py` + +**Step 1: 创建 box_service.py** + +```python +from sqlalchemy import text +from sqlalchemy.orm import Session + +from app.models.finished_goods import FinishedGoodsBox, FinishedGoodsBoxItem +from app.schemas.box import BoxSaveRequest + + +class ZongpaiNotFoundError(Exception): + pass + + +class DuplicateBoxItemError(Exception): + def __init__(self, paichan_no: str, box_no: int): + self.paichan_no = paichan_no + self.box_no = box_no + + +def query_erp_info(db: Session, zongpai_no: str) -> dict: + """从 ERP 视图查询总排号对应的排产号和数量。""" + sql = text( + "SELECT TOP 1 [总排号], [排产号], [数量] " + "FROM [ERPAuto].[vw_productionContractData] " + "WHERE [总排号] = :zongpai_no" + ) + row = db.execute(sql, {"zongpai_no": zongpai_no}).fetchone() + if not row: + raise ZongpaiNotFoundError() + return { + "zongpai_no": row[0], + "paichan_no": row[1], + "quantity": int(row[2]), + } + + +def get_box_info(db: Session, zongpai_no: str) -> dict: + """获取装箱信息：排产号、数量、已有箱号明细。""" + erp = query_erp_info(db, zongpai_no) + paichan_no = erp["paichan_no"] + quantity = erp["quantity"] + + # 查询该排产号下的所有箱（按 box_no 升序） + boxes = ( + db.query(FinishedGoodsBox) + .filter(FinishedGoodsBox.paichan_no == paichan_no) + .order_by(FinishedGoodsBox.box_no) + .all() + ) + + existing_boxes = [] + max_box_no = 0 + for box in boxes: + items = ( + db.query(FinishedGoodsBoxItem) + .filter(FinishedGoodsBoxItem.box_id == box.id) + .all() + ) + existing_boxes.append({ + "box_no": box.box_no, + "items": [ + {"zongpai_no": item.zongpai_no, "quantity": int(item.quantity or 0)} + for item in items + ], + }) + if box.box_no > max_box_no: + max_box_no = box.box_no + + return { + "zongpai_no": zongpai_no, + "paichan_no": paichan_no, + "quantity": quantity, + "existing_boxes": existing_boxes, + "max_box_no": max_box_no, + "suggested_box_no": max_box_no + 1, + } + + +def save_box_record(db: Session, req: BoxSaveRequest) -> dict: + """保存装箱记录。处理多码一箱场景：箱已存在时仅追加明细。""" + # 1. 查询 ERP 获取排产号 + erp = query_erp_info(db, req.zongpai_no) + paichan_no = erp["paichan_no"] + + # 2. 查找已有箱记录 + box = ( + db.query(FinishedGoodsBox) + .filter( + FinishedGoodsBox.paichan_no == paichan_no, + FinishedGoodsBox.box_no == req.box_no, + ) + .first() + ) + + if box: + # 箱已存在 → 检查该总排号是否已在此箱中 + existing_item = ( + db.query(FinishedGoodsBoxItem) + .filter( + FinishedGoodsBoxItem.box_id == box.id, + FinishedGoodsBoxItem.zongpai_no == req.zongpai_no, + ) + .first() + ) + if existing_item: + raise DuplicateBoxItemError(paichan_no, req.box_no) + else: + # 创建新箱 + box = FinishedGoodsBox( + paichan_no=paichan_no, + box_no=req.box_no, + ) + db.add(box) + db.flush() # 获取 box.id + + # 3. 创建箱明细 + item = FinishedGoodsBoxItem( + box_id=box.id, + zongpai_no=req.zongpai_no, + quantity=req.quantity, + ) + db.add(item) + db.commit() + db.refresh(item) + + return { + "paichan_no": paichan_no, + "box_no": req.box_no, + "zongpai_no": req.zongpai_no, + "quantity": req.quantity, + "created_at": item.created_at.isoformat(), + } +``` + +**Step 2: 验证导入** + +Run: `cd apps/fastAPI && .venv/Scripts/python -c "from app.services.box_service import get_box_info, save_box_record; print('Service OK')"` + +**Step 3: Commit** + +```bash +git add app/services/box_service.py +git commit -m "feat: add box service with ERP query and save logic" +``` + +--- + +### Task 3: API 路由层 — GET /box/info 和 POST /box + +**Files:** +- Create: `app/api/v1/box.py` +- Modify: `app/main.py` + +**Step 1: 创建 api/v1/box.py** + +```python +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.services.box_service import get_box_info, save_box_record + +router = APIRouter(tags=["boxing"]) + + +@router.get("/box/info", response_model=BoxInfoResponse) +def box_info( + zongpai_no: str = Query(..., description="总排号"), + db: Session = Depends(get_db), +): + """查询装箱信息：根据总排号查询排产号、数量和已有箱号明细。""" + return get_box_info(db, zongpai_no) + + +@router.post("/box", response_model=BoxSaveResponse) +def create_box(req: BoxSaveRequest, db: Session = Depends(get_db)): + """保存装箱记录：将总排号记录到指定箱号中。""" + return save_box_record(db, req) +``` + +**Step 2: 更新 main.py — 注册路由 + 异常处理** + +在现有 `main.py` 中添加： + +```python +# 新增导入（文件顶部） +from app.api.v1.box import router as box_router +from app.services.box_service import ZongpaiNotFoundError, DuplicateBoxItemError + +# 注册路由（在 location_router 之后） +app.include_router(box_router, prefix="/CargoTrace") + +# 新增异常处理（在现有 handler 之后） +@app.exception_handler(ZongpaiNotFoundError) +async def zongpai_not_found_handler(request: Request, exc: ZongpaiNotFoundError): + return JSONResponse( + status_code=404, + content={ + "error_code": "ZONGPAI_NOT_FOUND", + "message": "未找到该总排号对应的排产号信息", + }, + ) + + +@app.exception_handler(DuplicateBoxItemError) +async def duplicate_box_handler(request: Request, exc: DuplicateBoxItemError): + return JSONResponse( + status_code=409, + content={ + "error_code": "DUPLICATE_BOX_NO", + "message": f"排产号 {exc.paichan_no} 下箱号 {exc.box_no} 已存在", + "paichan_no": exc.paichan_no, + "box_no": exc.box_no, + }, + ) +``` + +**Step 3: 启动验证** + +Run: `cd apps/fastAPI && .venv/Scripts/python -m uvicorn app.main:app --reload` + +访问 `http://127.0.0.1:8000/docs`，确认 Swagger UI 中出现 `GET /CargoTrace/box/info` 和 `POST /CargoTrace/box` 两个端点。 + +**Step 4: 快速功能验证** + +使用 Swagger UI 或 curl 测试： +- GET `/CargoTrace/box/info?zongpai_no=26BW0011` → 应返回 W00009 的箱号信息 +- POST `/CargoTrace/box` body: `{"zongpai_no": "26BW0011", "box_no": 99, "quantity": 5}` → 应返回 200 + +**Step 5: Commit** + +```bash +git add app/api/v1/box.py app/main.py +git commit -m "feat: add GET /box/info and POST /box endpoints for boxing module" +``` + +--- + +### Task 4: 测试 — 接口集成测试 + +**Files:** +- Modify: `tests/conftest.py` +- Create: `tests/test_box_api.py` + +**Step 1: 更新 conftest.py — 添加装箱测试数据清理** + +在现有 `cleanup_test_data` fixture 中追加装箱表的测试数据清理： + +```python +# 在 conftest.py 的 cleanup_test_data 中追加 +from app.models.finished_goods import FinishedGoodsBox, FinishedGoodsBoxItem + +# 清理装箱测试数据 +test_box_items = db.query(FinishedGoodsBoxItem).filter( + FinishedGoodsBoxItem.zongpai_no.in_(["26B997", "26B998", "26B999"]) +).all() +for item in test_box_items: + db.delete(item) + +db.query(FinishedGoodsBox).filter( + FinishedGoodsBox.paichan_no.in_(["TEST_PAICHAN_999"]) +).delete(synchronize_session=False) +db.commit() +``` + +> 注意：装箱测试使用 ERP 中真实存在的总排号 `26BW0011`（排产号 W00009）做查询测试，保存测试则使用无效的测试总排号（会被 404 拦截），避免污染真实数据。如需完整保存测试，使用临时箱号（如 999），测试后立即清理。 + +**Step 2: 创建 test_box_api.py** + +```python +from fastapi.testclient import TestClient + + +# --- GET /box/info 测试 --- + + +def test_box_info_success(client: TestClient): + """查询真实存在的总排号 → 200""" + resp = client.get("/CargoTrace/box/info", params={"zongpai_no": "26BW0011"}) + assert resp.status_code == 200 + data = resp.json() + assert data["zongpai_no"] == "26BW0011" + assert data["paichan_no"] == "W00009" + assert data["quantity"] == 80 + assert "existing_boxes" in data + assert "max_box_no" in data + assert data["suggested_box_no"] == data["max_box_no"] + 1 + + +def test_box_info_invalid_zongpai(client: TestClient): + """无效总排号格式 → 400""" + resp = client.get("/CargoTrace/box/info", params={"zongpai_no": "INVALID"}) + assert resp.status_code == 400 + assert resp.json()["error_code"] == "INVALID_ZONGPAI" + + +def test_box_info_not_found(client: TestClient): + """格式合法但 ERP 中不存在 → 404""" + resp = client.get("/CargoTrace/box/info", params={"zongpai_no": "26B99999"}) + assert resp.status_code == 404 + assert resp.json()["error_code"] == "ZONGPAI_NOT_FOUND" + + +# --- POST /box 测试 --- + + +def test_box_save_invalid_zongpai(client: TestClient): + """无效总排号 → 400""" + resp = client.post( + "/CargoTrace/box", + json={"zongpai_no": "INVALID", "box_no": 1, "quantity": 10}, + ) + assert resp.status_code == 400 + assert resp.json()["error_code"] == "INVALID_ZONGPAI" + + +def test_box_save_invalid_box_no(client: TestClient): + """箱号 <= 0 → 400""" + resp = client.post( + "/CargoTrace/box", + json={"zongpai_no": "26BW0011", "box_no": 0, "quantity": 10}, + ) + assert resp.status_code == 400 + + +def test_box_save_invalid_quantity(client: TestClient): + """数量 <= 0 → 400""" + resp = client.post( + "/CargoTrace/box", + json={"zongpai_no": "26BW0011", "box_no": 1, "quantity": 0}, + ) + assert resp.status_code == 400 + + +def test_box_save_zongpai_not_found(client: TestClient): + """ERP 中不存在的总排号 → 404""" + resp = client.post( + "/CargoTrace/box", + json={"zongpai_no": "26B99999", "box_no": 1, "quantity": 10}, + ) + assert resp.status_code == 404 + assert resp.json()["error_code"] == "ZONGPAI_NOT_FOUND" +``` + +**Step 3: 运行测试** + +Run: `cd apps/fastAPI && .venv/Scripts/python -m pytest tests/test_box_api.py -v` + +**Step 4: Commit** + +```bash +git add tests/ +git commit -m "test: add integration tests for boxing API" +``` + +--- + +## API 接口汇总 + +| 方法 | 路径 | 描述 | +|------|------|------| +| GET | `/CargoTrace/box/info?zongpai_no=xxx` | 查询总排号对应的装箱信息 | +| POST | `/CargoTrace/box` | 保存装箱记录 | + +### 错误码 + +| HTTP 状态码 | error_code | 含义 | +|-------------|-----------|------| +| 400 | `INVALID_ZONGPAI` | 总排号格式不合法 | +| 400 | `INVALID_BOX_NO` | 箱号不合法（≤0） | +| 400 | `INVALID_QUANTITY` | 数量不合法（≤0） | +| 404 | `ZONGPAI_NOT_FOUND` | 总排号在 ERP 中未找到对应排产号 | +| 409 | `DUPLICATE_BOX_NO` | 该排产号下该箱号中已存在该总排号 | +| 500 | `SERVER_ERROR` | 服务器内部错误 | diff --git a/tests/test_box_api.py b/tests/test_box_api.py new file mode 100644 index 0000000..d11899e --- /dev/null +++ b/tests/test_box_api.py @@ -0,0 +1,78 @@ +from fastapi.testclient import TestClient + + +# --- GET /box/info 测试 --- + + +def test_box_info_success(client: TestClient): + """查询真实存在的总排号 → 200""" + resp = client.get( + "/CargoTrace/box/info", params={"zongpai_no": "26BW0011"} + ) + assert resp.status_code == 200 + data = resp.json() + assert data["zongpai_no"] == "26BW0011" + assert data["paichan_no"] == "W00009" + assert data["quantity"] == 80 + assert "existing_boxes" in data + assert "max_box_no" in data + assert data["suggested_box_no"] == data["max_box_no"] + 1 + + +def test_box_info_invalid_zongpai(client: TestClient): + """无效总排号格式 → 400""" + resp = client.get( + "/CargoTrace/box/info", params={"zongpai_no": "INVALID"} + ) + assert resp.status_code == 400 + assert resp.json()["error_code"] == "INVALID_ZONGPAI" + + +def test_box_info_not_found(client: TestClient): + """格式合法但 ERP 中不存在 → 404""" + resp = client.get( + "/CargoTrace/box/info", params={"zongpai_no": "26B99999"} + ) + assert resp.status_code == 404 + assert resp.json()["error_code"] == "ZONGPAI_NOT_FOUND" + + +# --- POST /box 测试 --- + + +def test_box_save_invalid_zongpai(client: TestClient): + """无效总排号 → 400""" + resp = client.post( + "/CargoTrace/box", + json={"zongpai_no": "INVALID", "box_no": 1, "quantity": 10}, + ) + assert resp.status_code == 400 + assert resp.json()["error_code"] == "INVALID_ZONGPAI" + + +def test_box_save_invalid_box_no(client: TestClient): + """箱号 <= 0 → 400""" + resp = client.post( + "/CargoTrace/box", + json={"zongpai_no": "26BW0011", "box_no": 0, "quantity": 10}, + ) + assert resp.status_code == 400 + + +def test_box_save_invalid_quantity(client: TestClient): + """数量 <= 0 → 400""" + resp = client.post( + "/CargoTrace/box", + json={"zongpai_no": "26BW0011", "box_no": 1, "quantity": 0}, + ) + assert resp.status_code == 400 + + +def test_box_save_zongpai_not_found(client: TestClient): + """ERP 中不存在的总排号 → 404""" + resp = client.post( + "/CargoTrace/box", + json={"zongpai_no": "26B99999", "box_no": 1, "quantity": 10}, + ) + assert resp.status_code == 404 + assert resp.json()["error_code"] == "ZONGPAI_NOT_FOUND"