From 6b722323ec4058a630d5435f7aaedfa37af69d34 Mon Sep 17 00:00:00 2001 From: Misaka_Company Date: Mon, 11 May 2026 13:08:05 +0800 Subject: [PATCH] docs: add implementation plans for shelf registration and 409 fix Co-Authored-By: Claude Opus 4.6 --- docs/plans/2026-05-11-fix-409-response.md | 266 +++++++++ docs/plans/2026-05-11-shelf-registration.md | 626 ++++++++++++++++++++ 2 files changed, 892 insertions(+) create mode 100644 docs/plans/2026-05-11-fix-409-response.md create mode 100644 docs/plans/2026-05-11-shelf-registration.md diff --git a/docs/plans/2026-05-11-fix-409-response.md b/docs/plans/2026-05-11-fix-409-response.md new file mode 100644 index 0000000..89165a5 --- /dev/null +++ b/docs/plans/2026-05-11-fix-409-response.md @@ -0,0 +1,266 @@ +# 修复 409 响应格式 — 第一阶段：FastAPI + +> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task. + +**Goal:** 将 409 重复上架的响应改为统一错误格式 `{"error_code", "message", "location_code", "registered_at"}`。 + +**Architecture:** 用自定义异常 `DuplicateLocationError` 替代 `HTTPException`，通过自定义 exception handler 返回扁平的统一错误结构。同时更新 200 成功响应和 Swagger 文档。 + +**Tech Stack:** FastAPI, SQLAlchemy, pytest + +--- + +## 目标响应格式 + +### 200 成功（不变） + +```json +{ + "zongpai_no": "26B1", + "location_code": "A01-02-03", + "created_at": "2026-05-11T10:30:00" +} +``` + +### 409 重复上架（修改后） + +```json +{ + "error_code": "DUPLICATE_LOCATION", + "message": "该总排号已存在货位记录", + "location_code": "A01-02-03", + "registered_at": "2026-05-11T10:30:00" +} +``` + +### 400 校验失败（修改后） + +```json +{ + "error_code": "INVALID_ZONGPAI", + "message": "INVALID_ZONGPAI" +} +``` + +--- + +### Task 1: 自定义异常 + handler + 测试更新 + +**Files:** +- Modify: `apps/fastAPI/app/services/location_service.py` +- Modify: `apps/fastAPI/app/main.py` +- Modify: `apps/fastAPI/app/api/v1/location.py` +- Modify: `apps/fastAPI/app/schemas/common.py` +- Modify: `apps/fastAPI/tests/test_location_api.py` + +**Step 1: 修改 `app/services/location_service.py` — 用自定义异常替换 HTTPException** + +```python +from sqlalchemy.orm import Session + +from app.models.finished_goods import FinishedGoodsLocation +from app.schemas.location import LocationRequest + + +class DuplicateLocationError(Exception): + """总排号已存在货位记录。""" + + def __init__(self, location_code: str, registered_at: str): + self.location_code = location_code + self.registered_at = registered_at + + +def register_location(db: Session, req: LocationRequest) -> FinishedGoodsLocation: + existing = ( + db.query(FinishedGoodsLocation) + .filter(FinishedGoodsLocation.zongpai_no == req.zongpai_no) + .first() + ) + if existing: + raise DuplicateLocationError( + location_code=existing.location_code, + registered_at=existing.created_at.isoformat(), + ) + + record = FinishedGoodsLocation( + zongpai_no=req.zongpai_no, + location_code=req.location_code, + ) + db.add(record) + db.commit() + db.refresh(record) + return record +``` + +**Step 2: 修改 `app/schemas/common.py` — 更新 ErrorResponse 匹配新格式** + +```python +from pydantic import BaseModel + + +class ErrorResponse(BaseModel): + error_code: str + message: str +``` + +**Step 3: 修改 `app/main.py` — 添加 DuplicateLocationError handler，更新 ValueError handler** + +```python +from fastapi import FastAPI, Request +from fastapi.responses import JSONResponse + +from app.api.v1.location import router as location_router +from app.services.location_service import DuplicateLocationError + +app = FastAPI(title="CargoTrace API", version="0.1.0") + +app.include_router(location_router, prefix="/CargoTrace") + + +@app.exception_handler(DuplicateLocationError) +async def duplicate_location_handler(request: Request, exc: DuplicateLocationError): + return JSONResponse( + status_code=409, + content={ + "error_code": "DUPLICATE_LOCATION", + "message": "该总排号已存在货位记录", + "location_code": exc.location_code, + "registered_at": exc.registered_at, + }, + ) + + +@app.exception_handler(ValueError) +async def value_error_handler(request: Request, exc: ValueError): + error_code = str(exc) + return JSONResponse( + status_code=400, + content={ + "error_code": error_code, + "message": error_code, + }, + ) + + +@app.get("/") +async def root(): + return {"message": "Welcome to CargoTrace API"} +``` + +**Step 4: 修改 `app/api/v1/location.py` — 更新 Swagger 409 文档** + +```python +from fastapi import APIRouter, Depends +from sqlalchemy.orm import Session + +from app.core.database import get_db +from app.schemas.common import ErrorResponse +from app.schemas.location import LocationRequest, LocationResponse +from app.services.location_service import register_location + +router = APIRouter() + + +@router.post( + "/location", + response_model=LocationResponse, + responses={ + 400: { + "description": "总排号或货位号格式不合法", + "model": ErrorResponse, + }, + 409: { + "description": "该总排号已存在货位记录", + "model": ErrorResponse, + }, + }, +) +def create_location(req: LocationRequest, db: Session = Depends(get_db)): + record = register_location(db, req) + return LocationResponse( + zongpai_no=record.zongpai_no, + location_code=record.location_code, + created_at=record.created_at.isoformat(), + ) +``` + +**Step 5: 修改 `tests/test_location_api.py` — 更新断言** + +```python +from fastapi.testclient import TestClient + + +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(client: TestClient): + """重复上架 — 应返回 409，包含统一错误格式""" + 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_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" +``` + +注意变更： +- `INVALID_ZONGPAI` / `INVALID_LOCATION` 现在返回 **400**（不再是 422），由自定义 `ValueError` handler 处理 +- 409 响应断言改为验证 `error_code`、`location_code`、`registered_at` 三个字段 + +**Step 6: 运行测试** + +Run: `cd D:/projects/CargoTrace/apps/fastAPI && .venv/Scripts/python.exe -m pytest tests/ -v` +Expected: 5 tests PASSED + +**Step 7: Commit** + +```bash +git add app/services/location_service.py app/main.py app/api/v1/location.py app/schemas/common.py tests/test_location_api.py +git commit -m "fix: unify error response format with error_code, message and domain fields" +``` diff --git a/docs/plans/2026-05-11-shelf-registration.md b/docs/plans/2026-05-11-shelf-registration.md new file mode 100644 index 0000000..85df51b --- /dev/null +++ b/docs/plans/2026-05-11-shelf-registration.md @@ -0,0 +1,626 @@ +# 上架登记模块 (Shelf Registration) Implementation Plan + +> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task. + +**Goal:** 实现 PRD v1.0 中定义的上架登记后端接口 `POST /CargoTrace/location`，支持总排号与货位号的绑定。 + +**Architecture:** FastAPI + SQLAlchemy (同步模式) 连接 SQL Server。三层结构：API 路由层 → Service 业务层 → Model 数据层。使用 pydantic-settings 管理 .env 配置，Alembic 管理数据库迁移（当前版本表已存在，仅做初始化）。 + +**Tech Stack:** Python 3.12+, FastAPI, SQLAlchemy 2.x (sync), pyodbc, pydantic-settings, python-dotenv + +--- + +## 项目目录结构（最终目标） + +``` +apps/fastAPI/ +├── .env # 数据库连接配置 +├── .gitignore +├── requirements.txt +├── README.md +├── alembic.ini +├── alembic/ +├── app/ +│ ├── __init__.py +│ ├── main.py # FastAPI 应用入口 +│ ├── core/ +│ │ ├── __init__.py +│ │ ├── config.py # pydantic-settings 配置 +│ │ └── database.py # SQLAlchemy engine & session +│ ├── models/ +│ │ ├── __init__.py +│ │ └── finished_goods.py # 3 张表的 ORM 模型 +│ ├── schemas/ +│ │ ├── __init__.py +│ │ ├── location.py # 上架登记请求/响应 schema +│ │ └── common.py # 通用错误响应 schema +│ ├── api/ +│ │ ├── __init__.py +│ │ └── v1/ +│ │ ├── __init__.py +│ │ └── location.py # POST /CargoTrace/location +│ └── services/ +│ ├── __init__.py +│ └── location_service.py # 上架登记业务逻辑 +├── tests/ +│ ├── __init__.py +│ ├── conftest.py # 测试 fixtures (TestClient, mock DB) +│ └── test_location_api.py # 接口测试 +``` + +--- + +### Task 1: 项目基础设施 — 配置与环境 + +**Files:** +- Create: `apps/fastAPI/.env` +- Create: `apps/fastAPI/.gitignore` +- Modify: `apps/fastAPI/requirements.txt` +- Create: `apps/fastAPI/app/core/__init__.py` +- Create: `apps/fastAPI/app/core/config.py` +- Create: `apps/fastAPI/app/core/database.py` + +**Step 1: 更新 requirements.txt** + +``` +fastapi>=0.115.0 +uvicorn[standard]>=0.34.0 +sqlalchemy>=2.0.0 +pyodbc>=5.2.0 +pydantic-settings>=2.0.0 +python-dotenv>=1.0.0 +``` + +**Step 2: 创建 .env 文件** + +```env +# SQL Server +SQL_SERVER_HOST=192.168.110.114 +SQL_SERVER_PORT=1433 +SQL_SERVER_DATABASE=CompanyDB +SQL_SERVER_USERNAME=peng +SQL_SERVER_PASSWORD=Cqbld123456. +SQL_SERVER_DRIVER={ODBC Driver 18 for SQL Server} +SQL_SERVER_TRUST_SERVER_CERTIFICATE=yes +``` + +**Step 3: 创建 .gitignore** + +``` +.venv/ +__pycache__/ +*.pyc +.env +*.egg-info/ +dist/ +build/ +.pytest_cache/ +``` + +**Step 4: 创建 app/core/config.py** + +```python +from pydantic_settings import BaseSettings + + +class Settings(BaseSettings): + SQL_SERVER_HOST: str + SQL_SERVER_PORT: int = 1433 + SQL_SERVER_DATABASE: str + SQL_SERVER_USERNAME: str + SQL_SERVER_PASSWORD: str + SQL_SERVER_DRIVER: str = "{ODBC Driver 18 for SQL Server}" + SQL_SERVER_TRUST_SERVER_CERTIFICATE: str = "yes" + + @property + def database_url(self) -> str: + return ( + f"mssql+pyodbc://{self.SQL_SERVER_USERNAME}:{self.SQL_SERVER_PASSWORD}" + f"@{self.SQL_SERVER_HOST},{self.SQL_SERVER_PORT}/{self.SQL_SERVER_DATABASE}" + f"?driver={self.SQL_SERVER_DRIVER}" + f"&TrustServerCertificate={self.SQL_SERVER_TRUST_SERVER_CERTIFICATE}" + ) + + model_config = {"env_file": ".env", "extra": "ignore"} + + +settings = Settings() +``` + +**Step 5: 创建 app/core/database.py** + +```python +from sqlalchemy import create_engine +from sqlalchemy.orm import DeclarativeBase, sessionmaker + +from app.core.config import settings + +engine = create_engine(settings.database_url) +SessionLocal = sessionmaker(bind=engine) + + +class Base(DeclarativeBase): + pass + + +def get_db(): + db = SessionLocal() + try: + yield db + finally: + db.close() +``` + +**Step 6: 验证数据库连接** + +Run: `cd apps/fastAPI && source .venv/bin/activate && python -c "from app.core.database import engine; engine.connect(); print('DB OK')"` +Expected: `DB OK` + +**Step 7: Commit** + +```bash +git add .env .gitignore requirements.txt app/core/ +git commit -m "feat: add project config and database connection" +``` + +--- + +### Task 2: ORM 模型 — 三张表映射 + +**Files:** +- Create: `apps/fastAPI/app/models/__init__.py` +- Create: `apps/fastAPI/app/models/finished_goods.py` + +**Step 1: 创建 models/finished_goods.py** + +```python +from datetime import datetime + +from sqlalchemy import BigInteger, DateTime, Index, Numeric, String, func +from sqlalchemy.orm import Mapped, mapped_column + +from app.core.database import Base + +SCHEMA = "CargoTrace" + + +class FinishedGoodsLocation(Base): + __tablename__ = "finished_goods_location" + __table_args__ = ( + Index("uk_zongpai_no", "zongpai_no", unique=True), + Index("idx_fgl_location_code", "location_code"), + {"schema": SCHEMA}, + ) + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + zongpai_no: Mapped[str] = mapped_column(String(64), nullable=False) + location_code: Mapped[str] = mapped_column(String(64), nullable=False) + created_at: Mapped[datetime] = mapped_column( + DateTime, nullable=False, server_default=func.getdate() + ) + + +class FinishedGoodsBox(Base): + __tablename__ = "finished_goods_box" + __table_args__ = ( + Index("uk_paichan_box", "paichan_no", "box_no", unique=True), + Index("idx_fgb_paichan_no", "paichan_no"), + {"schema": SCHEMA}, + ) + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + paichan_no: Mapped[str] = mapped_column(String(64), nullable=False) + box_no: Mapped[int] = mapped_column(nullable=False) + created_at: Mapped[datetime] = mapped_column( + DateTime, nullable=False, server_default=func.getdate() + ) + + +class FinishedGoodsBoxItem(Base): + __tablename__ = "finished_goods_box_item" + __table_args__ = ( + Index("idx_fgbi_box_id", "box_id"), + Index("idx_fgbi_zongpai_no", "zongpai_no"), + {"schema": SCHEMA}, + ) + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + box_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + zongpai_no: Mapped[str] = mapped_column(String(64), nullable=False) + quantity: Mapped[float | None] = mapped_column(Numeric(18, 3), nullable=True) + created_at: Mapped[datetime] = mapped_column( + DateTime, nullable=False, server_default=func.getdate() + ) +``` + +**Step 2: 创建 models/__init__.py** + +```python +from app.models.finished_goods import ( + FinishedGoodsBox, + FinishedGoodsBoxItem, + FinishedGoodsLocation, +) + +__all__ = [ + "FinishedGoodsLocation", + "FinishedGoodsBox", + "FinishedGoodsBoxItem", +] +``` + +**Step 3: 验证模型可正常导入** + +Run: `cd apps/fastAPI && python -c "from app.models import FinishedGoodsLocation; print('Models OK')"` +Expected: `Models OK` + +**Step 4: Commit** + +```bash +git add app/models/ +git commit -m "feat: add ORM models for finished goods tables" +``` + +--- + +### Task 3: Pydantic Schemas — 请求/响应模型 + +**Files:** +- Create: `apps/fastAPI/app/schemas/__init__.py` +- Create: `apps/fastAPI/app/schemas/common.py` +- Create: `apps/fastAPI/app/schemas/location.py` + +**Step 1: 创建 schemas/common.py — 通用错误响应** + +```python +from pydantic import BaseModel + + +class ErrorResponse(BaseModel): + error_code: str + message: str + detail: dict | None = None +``` + +**Step 2: 创建 schemas/location.py — 上架登记请求/响应** + +```python +import re + +from pydantic import BaseModel, field_validator + +# 总排号正则: 两位年份 + 类型字母(B/C/T) + 流水号 或 两位年份 + 温度计(BW/CW) + 4位流水号 +ZONGPAI_PATTERN = re.compile(r"^(\d{2}(B|C|T)\d+|\d{2}(BW|CW)\d{4})$") +# 普通货位号: 区域货架-层-格 (全大写字母+数字, 横杠分隔) +LOCATION_NORMAL_PATTERN = re.compile(r"^[A-Z]+\d+-\d+-\d+$") +# 转运特殊货位号: TRANS-序号 +LOCATION_TRANS_PATTERN = re.compile(r"^TRANS-\d+") + + +class LocationRequest(BaseModel): + zongpai_no: str + location_code: str + + @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("location_code") + @classmethod + def validate_location_code(cls, v: str) -> str: + v = v.strip().upper() + if not LOCATION_NORMAL_PATTERN.match(v) and not LOCATION_TRANS_PATTERN.match(v): + raise ValueError("INVALID_LOCATION") + return v + + +class LocationResponse(BaseModel): + zongpai_no: str + location_code: str + created_at: str + + +class DuplicateLocationDetail(BaseModel): + existing_location: str + created_at: str +``` + +**Step 3: 创建 schemas/__init__.py** + +```python +``` + +**Step 4: Commit** + +```bash +git add app/schemas/ +git commit -m "feat: add pydantic schemas for location API" +``` + +--- + +### Task 4: Service 层 — 上架登记业务逻辑 + +**Files:** +- Create: `apps/fastAPI/app/services/__init__.py` +- Create: `apps/fastAPI/app/services/location_service.py` + +**Step 1: 创建 services/location_service.py** + +```python +from fastapi import HTTPException +from sqlalchemy.orm import Session + +from app.models.finished_goods import FinishedGoodsLocation +from app.schemas.location import LocationRequest + + +def register_location(db: Session, req: LocationRequest) -> FinishedGoodsLocation: + existing = ( + db.query(FinishedGoodsLocation) + .filter(FinishedGoodsLocation.zongpai_no == req.zongpai_no) + .first() + ) + if existing: + raise HTTPException( + status_code=409, + detail={ + "error_code": "DUPLICATE_LOCATION", + "message": "该总排号已存在货位记录", + "detail": { + "existing_location": existing.location_code, + "created_at": existing.created_at.isoformat(), + }, + }, + ) + + record = FinishedGoodsLocation( + zongpai_no=req.zongpai_no, + location_code=req.location_code, + ) + db.add(record) + db.commit() + db.refresh(record) + return record +``` + +**Step 2: 创建 services/__init__.py** + +```python +``` + +**Step 3: Commit** + +```bash +git add app/services/ +git commit -m "feat: add location registration service" +``` + +--- + +### Task 5: API 路由层 — POST /CargoTrace/location + +**Files:** +- Create: `apps/fastAPI/app/api/__init__.py` +- Create: `apps/fastAPI/app/api/v1/__init__.py` +- Create: `apps/fastAPI/app/api/v1/location.py` +- Modify: `apps/fastAPI/app/main.py` + +**Step 1: 创建 api/v1/location.py** + +```python +from fastapi import APIRouter, Depends +from sqlalchemy.orm import Session + +from app.core.database import get_db +from app.schemas.location import LocationRequest, LocationResponse +from app.services.location_service import register_location + +router = APIRouter() + + +@router.post("/location", response_model=LocationResponse) +def create_location(req: LocationRequest, db: Session = Depends(get_db)): + record = register_location(db, req) + return LocationResponse( + zongpai_no=record.zongpai_no, + location_code=record.location_code, + created_at=record.created_at.isoformat(), + ) +``` + +**Step 2: 更新 main.py — 挂载路由** + +```python +from fastapi import FastAPI, Request +from fastapi.responses import JSONResponse + +from app.api.v1.location import router as location_router + +app = FastAPI(title="CargoTrace API", version="0.1.0") + +app.include_router(location_router, prefix="/CargoTrace") + + +@app.exception_handler(ValueError) +async def value_error_handler(request: Request, exc: ValueError): + error_code = str(exc) + status_map = { + "INVALID_ZONGPAI": 400, + "INVALID_LOCATION": 400, + } + status = status_map.get(error_code, 400) + return JSONResponse( + status_code=status, + content={ + "error_code": error_code, + "message": str(exc), + }, + ) + + +@app.get("/") +async def root(): + return {"message": "Welcome to CargoTrace API"} +``` + +**Step 3: 启动验证** + +Run: `cd apps/fastAPI && uvicorn app.main:app --reload` +Expected: 服务启动成功，访问 `http://127.0.0.1:8000/docs` 可看到 Swagger UI + +**Step 4: Commit** + +```bash +git add app/api/ app/main.py +git commit -m "feat: add POST /CargoTrace/location endpoint" +``` + +--- + +### Task 6: 测试 — 接口集成测试 + +**Files:** +- Modify: `apps/fastAPI/requirements.txt` (添加测试依赖) +- Create: `apps/fastAPI/tests/__init__.py` +- Create: `apps/fastAPI/tests/conftest.py` +- Create: `apps/fastAPI/tests/test_location_api.py` + +**Step 1: 添加测试依赖到 requirements.txt** + +追加重定内容: + +``` +pytest>=8.0.0 +httpx>=0.28.0 +``` + +**Step 2: 创建 tests/conftest.py — 测试 fixtures，使用真实数据库** + +```python +import pytest +from fastapi.testclient import TestClient +from sqlalchemy.orm import Session + +from app.core.database import SessionLocal +from app.main import app +from app.models.finished_goods import FinishedGoodsLocation + + +@pytest.fixture +def client(): + return TestClient(app) + + +@pytest.fixture +def db(): + session = SessionLocal() + yield session + session.close() + + +@pytest.fixture(autouse=True) +def cleanup_test_data(db: Session): + yield + # 清理测试数据：删除测试用总排号记录 + test_zongpai_nos = ["26B999", "26C999", "26BW0999", "26T999"] + db.query(FinishedGoodsLocation).filter( + FinishedGoodsLocation.zongpai_no.in_(test_zongpai_nos) + ).delete(synchronize_session=False) + db.commit() +``` + +**Step 3: 创建 tests/test_location_api.py** + +```python +from fastapi.testclient import TestClient + + +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(client: TestClient): + """重复上架 — 应返回 409""" + 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["detail"]["error_code"] == "DUPLICATE_LOCATION" + assert "existing_location" in data["detail"]["detail"] + + +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 == 422 # Pydantic validation error + + +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 == 422 + + +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" +``` + +**Step 4: 运行测试** + +Run: `cd apps/fastAPI && pytest tests/ -v` +Expected: 5 tests passed + +**Step 5: Commit** + +```bash +git add requirements.txt tests/ +git commit -m "test: add integration tests for location API" +``` + +--- + +## API 接口汇总 + +| 方法 | 路径 | 描述 | +|------|------|------| +| POST | `/CargoTrace/location` | 上架登记 — 绑定总排号与货位号 | + +### 错误码 + +| HTTP 状态码 | error_code | 含义 | +|-------------|-----------|------| +| 409 | `DUPLICATE_LOCATION` | 该总排号已存在货位记录 | +| 422 | `INVALID_ZONGPAI` | 总排号格式不合法 | +| 422 | `INVALID_LOCATION` | 货位号格式不合法 | +| 500 | `SERVER_ERROR` | 服务器内部错误 |