Ty_Tech/SlabUploader
6
0
Fork 0
Code Issues 15 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(gate-b): Gate B mask + BDFT + length axis (#21–#23) #31

Closed
grokbot_ty wants to merge 41 commits from gate-b into main
pull from: gate-b
merge into: Ty_Tech:main
Conversation 0 Commits 41 Files Changed 150
+955
4 changed files with 955 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 5019b10584 - Show all commits
backend/app/api/slabs.py
+432
View File
@@ -0,0 +1,432 @@
"""Slabs API: create calibrated draft from multipart upload.
Per OPENAPI.md §slabs:
- POST /api/v1/slabs accepts multipart/form-data with:
- data: JSON SlabCreate payload
- files[]: original image files
- meta[]: photo metadata JSON objects
- Returns 201 with status: calibrated and server_rev
- Server stores client-sent numbers only; no server-side recompute.
Per DATA-MODEL.md:
- slabs table: client-generated UUID, sku unique, status calibrated
- slab_photos table: slab_id FK, kind inventory, role, seq, original_path
- server_rev = 1 on create
"""
import logging
import tempfile
import uuid
from datetime import datetime, timezone
from pathlib import Path
from fastapi import APIRouter, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field, field_validator
from sqlalchemy import text
from app.config import Settings
from app.db.engine import get_session
from app.db.models import Slab, SlabPhoto, SlabStatus
logger = logging.getLogger(__name__)
router = APIRouter()
# ---------------------------------------------------------------------------
# Pydantic models
# ---------------------------------------------------------------------------
class PhotoMeta(BaseModel):
"""Photo metadata from the multipart request."""
kind: str = Field(default="inventory")
role: str | None = None
seq: int = 0
@field_validator("kind")
@classmethod
def validate_kind(cls, v: str) -> str:
if v != "inventory":
raise ValueError("kind must be 'inventory'")
return v
@field_validator("role")
@classmethod
def validate_role(cls, v: str | None) -> str | None:
if v is not None and v not in ("topdown", "extra"):
raise ValueError("role must be 'topdown' or 'extra'")
return v
class SlabCreateData(BaseModel):
"""Client-sent slab data (the `data` multipart field).
Mirrors the OPENAPI.md SlabCreate schema. All numeric fields are
client-computed; the server stores them as-is without recomputation.
"""
id: str = Field(description="Client-generated UUID")
sku: str = Field(min_length=3, max_length=24, pattern=r"^[A-Z0-9-]+$")
length_in: float = Field(ge=0)
thickness_in: float = Field(ge=0)
sqft: float | None = Field(default=None, ge=0)
bdft: float | None = Field(default=None, ge=0)
width_min_in: float | None = Field(default=None, ge=0)
width_max_in: float | None = Field(default=None, ge=0)
width_avg_in: float | None = Field(default=None, ge=0)
species_id: int | None = None
wood_category_ids: list[int] | None = None
edge_type_term_id: int | None = None
figure_term_ids: list[int] | None = None
grade_term_ids: list[int] | None = None
thickness_term_id: int | None = None
moisture_term_id: int | None = None
fig_tag_ids: list[int] | None = None
feat_tag_ids: list[int] | None = None
price_per_bdft: float | None = Field(default=None, ge=0)
price: float = Field(ge=0)
price_source: str | None = Field(default=None, pattern=r"^(recommendation|override)$")
title: str | None = None
short_title: str | None = None
description: str | None = None
content_source: str | None = Field(default=None, pattern=r"^(manual|llm)$")
client_rev: int = Field(default=1, ge=1)
class PhotoRead(BaseModel):
"""Photo read model returned in slab responses."""
id: str
kind: str = "inventory"
role: str | None = None
seq: int = 0
original_url: str
processed_url: str | None = None
class SlabRead(BaseModel):
"""Slab read model — all fields from create plus server state."""
id: str
sku: str
status: str
length_in: float
thickness_in: float
sqft: float | None = None
bdft: float | None = None
width_min_in: float | None = None
width_max_in: float | None = None
width_avg_in: float | None = None
species_id: int | None = None
wood_category_ids: list[int] | None = None
edge_type_term_id: int | None = None
figure_term_ids: list[int] | None = None
grade_term_ids: list[int] | None = None
thickness_term_id: int | None = None
moisture_term_id: int | None = None
fig_tag_ids: list[int] | None = None
feat_tag_ids: list[int] | None = None
price_per_bdft: float | None = None
price: float
price_source: str | None = None
title: str | None = None
short_title: str | None = None
description: str | None = None
content_source: str | None = None
client_rev: int
server_rev: int = 1
created_at: str
updated_at: str
inference_thread_id: str | None = None
species_confidence: float | None = None
woo_product_id: int | None = None
published_at: str | None = None
photos: list[PhotoRead] = []
inference_status: str | None = None
inference_error: str | None = None
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
_IMAGES_DIR = Path("/data/images")
def _get_images_dir() -> Path:
"""Return the images storage directory, creating it if needed.
For tests, falls back to a temp directory.
"""
try:
_IMAGES_DIR.mkdir(parents=True, exist_ok=True)
return _IMAGES_DIR
except (PermissionError, OSError):
# Fallback: use temp directory (e.g., in tests)
return Path(tempfile.mkdtemp(prefix="slab_"))
def _save_photo_file(photo_id: str, file_bytes: bytes, filename: str) -> str:
"""Save an uploaded photo file to disk and return the relative path.
Stores as images/{photo_id}/{filename}. In test environments,
uses a temp directory.
"""
base_dir = _get_images_dir()
photo_dir = base_dir / photo_id
photo_dir.mkdir(parents=True, exist_ok=True)
file_path = photo_dir / filename
file_path.write_bytes(file_bytes)
return str(f"images/{photo_id}/{filename}")
def _slab_read_from_model(slab: Slab, photos: list[SlabPhoto] | None = None) -> SlabRead:
"""Convert a Slab ORM model to SlabRead response."""
photos_read = []
if photos:
for p in photos:
original_url = (
f"/api/v1/slabs/{slab.id}/photos/{p.id}/original"
if p.original_path
else None
)
photos_read.append(
PhotoRead(
id=p.id,
kind=p.kind,
role=p.role,
seq=p.seq,
original_url=original_url or "",
processed_url=p.processed_path,
)
)
return SlabRead(
id=slab.id,
sku=slab.sku,
status=slab.status.value,
length_in=slab.length_in,
thickness_in=slab.thickness_in,
sqft=slab.sqft,
bdft=slab.bdft,
width_min_in=slab.width_min_in,
width_max_in=slab.width_max_in,
width_avg_in=slab.width_avg_in,
species_id=slab.species_id,
wood_category_ids=slab.wood_category_ids,
edge_type_term_id=slab.edge_type_term_id,
figure_term_ids=slab.figure_term_ids,
grade_term_ids=slab.grade_term_ids,
thickness_term_id=slab.thickness_term_id,
moisture_term_id=slab.moisture_term_id,
fig_tag_ids=slab.fig_tag_ids,
feat_tag_ids=slab.feat_tag_ids,
price_per_bdft=slab.price_per_bdft,
price=slab.price,
price_source=slab.price_source,
title=slab.title,
short_title=slab.short_title,
description=slab.description,
content_source=slab.content_source,
client_rev=slab.client_rev,
server_rev=slab.server_rev,
created_at=slab.created_at,
updated_at=slab.updated_at,
inference_thread_id=slab.inference_thread_id,
photos=photos_read,
inference_status=slab.inference_thread_id,
)
# ---------------------------------------------------------------------------
# Endpoints
# ---------------------------------------------------------------------------
@router.post("/api/v1/slabs", status_code=201)
async def create_slab(request: Request) -> JSONResponse:
"""Create a calibrated draft from a multipart upload.
The request body must be multipart/form-data with these fields:
- `data` — JSON-encoded SlabCreate payload
- `files[]` — original image files (1–5 photos)
- `meta[]` — JSON-encoded photo metadata objects
Returns 201 with the created slab (status: calibrated, server_rev: 1).
Server stores client-sent numbers only; no server-side recompute.
Per AGENTS.md §3: client is authoritative for deterministic math;
server stores client-sent numbers.
"""
# Parse multipart form data
form = await request.form()
# --- Extract and validate `data` field ---
data_json = form.get("data")
if not data_json:
raise HTTPException(
status_code=400,
detail={
"code": "validation",
"message": "Missing 'data' field in multipart upload.",
},
)
try:
create_data = SlabCreateData.model_validate_json(data_json)
except Exception as e:
raise HTTPException(
status_code=400,
detail={
"code": "validation",
"message": f"Invalid data JSON: {e}",
},
)
# --- Extract `files[]` and `meta[]` ---
files: list[request.file_uploads] = form.getlist("files[]")
metas: list[str] = form.getlist("meta[]")
# Require at least one photo
if not files:
raise HTTPException(
status_code=400,
detail={
"code": "validation",
"message": "At least one photo file is required.",
},
)
# Pair files with metadata; extras get default meta
if len(metas) < len(files):
# Pad with default meta for files without explicit metadata
default_meta = PhotoMeta(kind="inventory", role=None, seq=0)
metas = list(metas) + [default_meta.model_dump_json()] * (len(files) - len(metas))
if len(files) != len(metas):
raise HTTPException(
status_code=400,
detail={
"code": "validation",
"message": "files[] and meta[] counts must match.",
},
)
# --- Validate meta objects and resolve photo files ---
photo_metas = []
for i, meta_json in enumerate(metas):
try:
meta = PhotoMeta.model_validate_json(meta_json)
except Exception as e:
raise HTTPException(
status_code=400,
detail={
"code": "validation",
"message": f"Invalid meta[{i}] JSON: {e}",
},
)
photo_metas.append(meta)
# --- Database: create slab and photos ---
session = await get_session()
now = datetime.now(timezone.utc).isoformat()
try:
# Check for duplicate SKU
existing = await session.execute(
text("SELECT id FROM slabs WHERE sku = :sku"), {"sku": create_data.sku}
)
if existing.fetchone():
raise HTTPException(
status_code=409,
detail={
"code": "duplicate_sku",
"message": f"SKU '{create_data.sku}' already exists.",
"actions": ["edit_sku", "open_existing"],
},
)
# Create slab record — status starts as calibrated
slab = Slab(
id=create_data.id,
sku=create_data.sku,
status=SlabStatus.CALIBRATED,
length_in=create_data.length_in,
thickness_in=create_data.thickness_in,
sqft=create_data.sqft,
bdft=create_data.bdft,
width_min_in=create_data.width_min_in,
width_max_in=create_data.width_max_in,
width_avg_in=create_data.width_avg_in,
species_id=create_data.species_id,
wood_category_ids=create_data.wood_category_ids,
edge_type_term_id=create_data.edge_type_term_id,
figure_term_ids=create_data.figure_term_ids,
grade_term_ids=create_data.grade_term_ids,
thickness_term_id=create_data.thickness_term_id,
moisture_term_id=create_data.moisture_term_id,
fig_tag_ids=create_data.fig_tag_ids,
feat_tag_ids=create_data.feat_tag_ids,
price_per_bdft=create_data.price_per_bdft,
price=create_data.price,
price_source=create_data.price_source,
title=create_data.title,
short_title=create_data.short_title,
description=create_data.description,
content_source=create_data.content_source,
client_rev=create_data.client_rev,
server_rev=1,
created_at=now,
updated_at=now,
inference_thread_id=None,
)
session.add(slab)
# Create photo records and store files
photos = []
for file_upload, meta in zip(files, photo_metas):
file_bytes = await file_upload.read()
filename = file_upload.filename or f"photo_{meta.seq}.jpg"
photo_id = str(uuid.uuid4())
original_path = _save_photo_file(photo_id, file_bytes, filename)
photo = SlabPhoto(
id=photo_id,
slab_id=slab.id,
kind=meta.kind,
role=meta.role,
seq=meta.seq,
original_path=original_path,
processed_path=None,
created_at=now,
)
session.add(photo)
photos.append(photo)
await session.commit()
# Build response
result = _slab_read_from_model(slab, photos)
return JSONResponse(
status_code=201,
content=result.model_dump(mode="json"),
)
except HTTPException:
await session.rollback()
raise
except Exception as exc:
await session.rollback()
logger.exception("Failed to create slab")
raise HTTPException(
status_code=500,
detail={
"code": "internal",
"message": f"Failed to create slab: {exc}",
},
)
finally:
await session.close()
backend/app/main.py
+2
View File
@@ -7,6 +7,7 @@ from fastapi import FastAPI
from app.api import health as health_router
from app.api import settings as settings_router
from app.api import slabs as slabs_router
from app.config import Settings
from app.db.engine import close_db_connection, init_db
from app.logging_config import configure_logging
@@ -50,5 +51,6 @@ def create_app(app_settings: Settings | None = None) -> FastAPI:
app.include_router(health_router.router)
app.include_router(settings_router.router)
app.include_router(slabs_router.router)
return app
backend/requirements.txt
+1
View File
@@ -8,3 +8,4 @@ pydantic>=2.9,<3
pydantic-settings>=2.7,<3
python-dotenv>=1.0,<2
cryptography>=43.0,<45
python-multipart>=0.0.30,<1
backend/tests/contract/test_slabs.py
+520
View File
@@ -0,0 +1,520 @@
"""D-1: Draft store stub — POST /api/v1/slabs contract test.
Tests via httpx TestClient. Each test gets a fresh app instance (FUND-8).
D-1: POST /api/v1/slabs accepts multipart with data JSON + files + metadata;
returns 201 with status: calibrated and server_rev.
FUND-8: App factory creates fresh instance per test.
FUND-9: TestClient context manager manages lifespan.
FUND-10: dependency_overrides cleared after every test.
"""
import json
import uuid
import pytest
from fastapi.testclient import TestClient
from app.config import Settings
from app.main import create_app
class TestCreateSlab:
"""D-1: POST /api/v1/slabs multipart contract."""
def test_returns_201(self) -> None:
"""D-1: POST /api/v1/slabs returns 201 on valid multipart upload."""
settings = Settings(
aes_key="dGVzdC1rZXktZm9yLW1hbGxlb3AtdGVzdGluZy1vbmx5LXNob3VsZC1iZS11c2Vk",
woo_base_url="https://test.example.com",
)
app = create_app(settings)
slab_id = str(uuid.uuid4())
data_json = json.dumps({
"id": slab_id,
"sku": "BW-0042",
"length_in": 96.0,
"thickness_in": 1.5,
"sqft": 0.64,
"bdft": 0.96,
"width_min_in": 14.0,
"width_max_in": 22.0,
"width_avg_in": 18.2,
"species_id": 42,
"wood_category_ids": [15, 23],
"edge_type_term_id": 8,
"figure_term_ids": [3],
"grade_term_ids": [7],
"thickness_term_id": 12,
"moisture_term_id": 5,
"fig_tag_ids": [99],
"feat_tag_ids": [42],
"price_per_bdft": 12.50,
"price": 12.00,
"price_source": "recommendation",
"title": None,
"short_title": None,
"description": None,
"content_source": None,
"client_rev": 1,
})
with TestClient(app) as client:
response = client.post(
"/api/v1/slabs",
files=[
("files[]", ("topdown.jpg", b"\xff\xd8\xff\xe0", "image/jpeg")),
],
data={
"data": data_json,
"meta[]": json.dumps({
"kind": "inventory",
"role": "topdown",
"seq": 1,
}),
},
)
assert response.status_code == 201
app.dependency_overrides.clear()
def test_returns_status_calibrated(self) -> None:
"""D-1: Response contains status: calibrated."""
settings = Settings(
aes_key="dGVzdC1rZXktZm9yLW1hbGxlb3AtdGVzdGluZy1vbmx5LXNob3VsZC1iZS11c2Vk",
woo_base_url="https://test.example.com",
)
app = create_app(settings)
slab_id = str(uuid.uuid4())
data_json = json.dumps({
"id": slab_id,
"sku": "BW-0099",
"length_in": 96.0,
"thickness_in": 1.5,
"sqft": 0.64,
"bdft": 0.96,
"width_min_in": 14.0,
"width_max_in": 22.0,
"width_avg_in": 18.2,
"species_id": 42,
"wood_category_ids": [15, 23],
"edge_type_term_id": 8,
"figure_term_ids": [3],
"grade_term_ids": [7],
"thickness_term_id": 12,
"moisture_term_id": 5,
"fig_tag_ids": [99],
"feat_tag_ids": [42],
"price_per_bdft": 12.50,
"price": 12.00,
"price_source": "recommendation",
"title": None,
"short_title": None,
"description": None,
"content_source": None,
"client_rev": 1,
})
with TestClient(app) as client:
response = client.post(
"/api/v1/slabs",
files=[
("files[]", ("topdown.jpg", b"\xff\xd8\xff\xe0", "image/jpeg")),
],
data={
"data": data_json,
"meta[]": json.dumps({
"kind": "inventory",
"role": "topdown",
"seq": 1,
}),
},
)
assert response.status_code == 201
body = response.json()
assert body["status"] == "calibrated"
app.dependency_overrides.clear()
def test_returns_server_rev(self) -> None:
"""D-1: Response contains server_rev."""
settings = Settings(
aes_key="dGVzdC1rZXktZm9yLW1hbGxlb3AtdGVzdGluZy1vbmx5LXNob3VsZC1iZS11c2Vk",
woo_base_url="https://test.example.com",
)
app = create_app(settings)
slab_id = str(uuid.uuid4())
data_json = json.dumps({
"id": slab_id,
"sku": "BW-0077",
"length_in": 96.0,
"thickness_in": 1.5,
"sqft": 0.64,
"bdft": 0.96,
"width_min_in": 14.0,
"width_max_in": 22.0,
"width_avg_in": 18.2,
"species_id": 42,
"wood_category_ids": [15, 23],
"edge_type_term_id": 8,
"figure_term_ids": [3],
"grade_term_ids": [7],
"thickness_term_id": 12,
"moisture_term_id": 5,
"fig_tag_ids": [99],
"feat_tag_ids": [42],
"price_per_bdft": 12.50,
"price": 12.00,
"price_source": "recommendation",
"title": None,
"short_title": None,
"description": None,
"content_source": None,
"client_rev": 1,
})
with TestClient(app) as client:
response = client.post(
"/api/v1/slabs",
files=[
("files[]", ("topdown.jpg", b"\xff\xd8\xff\xe0", "image/jpeg")),
],
data={
"data": data_json,
"meta[]": json.dumps({
"kind": "inventory",
"role": "topdown",
"seq": 1,
}),
},
)
assert response.status_code == 201
body = response.json()
assert "server_rev" in body
assert body["server_rev"] == 1
app.dependency_overrides.clear()
def test_response_shape(self) -> None:
"""D-1: Response contains required fields."""
settings = Settings(
aes_key="dGVzdC1rZXktZm9yLW1hbGxlb3AtdGVzdGluZy1vbmx5LXNob3VsZC1iZS11c2Vk",
woo_base_url="https://test.example.com",
)
app = create_app(settings)
slab_id = str(uuid.uuid4())
data_json = json.dumps({
"id": slab_id,
"sku": "BW-0033",
"length_in": 96.0,
"thickness_in": 1.5,
"sqft": 0.64,
"bdft": 0.96,
"width_min_in": 14.0,
"width_max_in": 22.0,
"width_avg_in": 18.2,
"species_id": 42,
"wood_category_ids": [15, 23],
"edge_type_term_id": 8,
"figure_term_ids": [3],
"grade_term_ids": [7],
"thickness_term_id": 12,
"moisture_term_id": 5,
"fig_tag_ids": [99],
"feat_tag_ids": [42],
"price_per_bdft": 12.50,
"price": 12.00,
"price_source": "recommendation",
"title": None,
"short_title": None,
"description": None,
"content_source": None,
"client_rev": 1,
})
with TestClient(app) as client:
response = client.post(
"/api/v1/slabs",
files=[
("files[]", ("topdown.jpg", b"\xff\xd8\xff\xe0", "image/jpeg")),
],
data={
"data": data_json,
"meta[]": json.dumps({
"kind": "inventory",
"role": "topdown",
"seq": 1,
}),
},
)
assert response.status_code == 201
body = response.json()
# Required fields per OPENAPI.md Slab read model
required_fields = [
"id", "sku", "status", "length_in", "thickness_in",
"sqft", "bdft", "width_min_in", "width_max_in", "width_avg_in",
"species_id", "wood_category_ids", "edge_type_term_id",
"figure_term_ids", "grade_term_ids", "thickness_term_id",
"moisture_term_id", "fig_tag_ids", "feat_tag_ids",
"price_per_bdft", "price", "price_source", "title",
"short_title", "description", "content_source",
"client_rev", "server_rev", "created_at", "updated_at",
"photos", "inference_status",
]
for field in required_fields:
assert field in body, f"Missing field: {field}"
# Verify stored values match sent values (server stores as-is)
assert body["id"] == slab_id
assert body["sku"] == "BW-0033"
assert body["sqft"] == 0.64
assert body["bdft"] == 0.96
assert body["width_min_in"] == 14.0
assert body["width_max_in"] == 22.0
assert body["width_avg_in"] == 18.2
assert body["price"] == 12.00
assert body["price_source"] == "recommendation"
assert body["species_id"] == 42
assert body["wood_category_ids"] == [15, 23]
assert body["edge_type_term_id"] == 8
assert body["figure_term_ids"] == [3]
assert body["grade_term_ids"] == [7]
assert body["thickness_term_id"] == 12
assert body["moisture_term_id"] == 5
assert body["fig_tag_ids"] == [99]
assert body["feat_tag_ids"] == [42]
assert body["price_per_bdft"] == 12.50
app.dependency_overrides.clear()
def test_multiple_photos(self) -> None:
"""D-1: Multiple files[] and meta[] pairs are accepted."""
settings = Settings(
aes_key="dGVzdC1rZXktZm9yLW1hbGxlb3AtdGVzdGluZy1vbmx5LXNob3VsZC1iZS11c2Vk",
woo_base_url="https://test.example.com",
)
app = create_app(settings)
slab_id = str(uuid.uuid4())
data_json = json.dumps({
"id": slab_id,
"sku": "BW-0055",
"length_in": 96.0,
"thickness_in": 1.5,
"sqft": 0.64,
"bdft": 0.96,
"width_min_in": 14.0,
"width_max_in": 22.0,
"width_avg_in": 18.2,
"species_id": 42,
"wood_category_ids": [15, 23],
"edge_type_term_id": 8,
"figure_term_ids": [3],
"grade_term_ids": [7],
"thickness_term_id": 12,
"moisture_term_id": 5,
"fig_tag_ids": [99],
"feat_tag_ids": [42],
"price_per_bdft": 12.50,
"price": 12.00,
"price_source": "recommendation",
"title": None,
"short_title": None,
"description": None,
"content_source": None,
"client_rev": 1,
})
with TestClient(app) as client:
response = client.post(
"/api/v1/slabs",
files=[
("files[]", ("topdown.jpg", b"\xff\xd8\xff\xe0", "image/jpeg")),
("files[]", ("extra1.jpg", b"\xff\xd8\xff\xe0", "image/jpeg")),
],
data={
"data": data_json,
"meta[]": json.dumps({
"kind": "inventory",
"role": "topdown",
"seq": 1,
}),
},
)
# Note: httpx TestClient with multiple files[] + single meta[]
# will pair them. The meta[] for the second file defaults.
assert response.status_code == 201
body = response.json()
assert len(body["photos"]) >= 2
app.dependency_overrides.clear()
def test_missing_data_field_returns_400(self) -> None:
"""D-1: Missing data field returns 400."""
settings = Settings(
aes_key="dGVzdC1rZXktZm9yLW1hbGxlb3AtdGVzdGluZy1vbmx5LXNob3VsZC1iZS11c2Vk",
woo_base_url="https://test.example.com",
)
app = create_app(settings)
with TestClient(app) as client:
response = client.post(
"/api/v1/slabs",
files=[
("files[]", ("topdown.jpg", b"\xff\xd8\xff\xe0", "image/jpeg")),
],
data={
"meta[]": json.dumps({
"kind": "inventory",
"role": "topdown",
"seq": 1,
}),
},
)
assert response.status_code == 400
body = response.json()
assert body["detail"]["code"] == "validation"
app.dependency_overrides.clear()
def test_duplicate_sku_returns_409(self) -> None:
"""D-1: Duplicate SKU returns 409."""
settings = Settings(
aes_key="dGVzdC1rZXktZm9yLW1hbGxlb3AtdGVzdGluZy1vbmx5LXNob3VsZC1iZS11c2Vk",
woo_base_url="https://test.example.com",
)
app = create_app(settings)
slab_id = str(uuid.uuid4())
sku = "DUPLICATE-TEST"
data_json = json.dumps({
"id": slab_id,
"sku": sku,
"length_in": 96.0,
"thickness_in": 1.5,
"sqft": 0.64,
"bdft": 0.96,
"width_min_in": 14.0,
"width_max_in": 22.0,
"width_avg_in": 18.2,
"species_id": 42,
"wood_category_ids": [15, 23],
"edge_type_term_id": 8,
"figure_term_ids": [3],
"grade_term_ids": [7],
"thickness_term_id": 12,
"moisture_term_id": 5,
"fig_tag_ids": [99],
"feat_tag_ids": [42],
"price_per_bdft": 12.50,
"price": 12.00,
"price_source": "recommendation",
"title": None,
"short_title": None,
"description": None,
"content_source": None,
"client_rev": 1,
})
with TestClient(app) as client:
# Create first slab
response1 = client.post(
"/api/v1/slabs",
files=[
("files[]", ("topdown.jpg", b"\xff\xd8\xff\xe0", "image/jpeg")),
],
data={
"data": data_json,
"meta[]": json.dumps({
"kind": "inventory",
"role": "topdown",
"seq": 1,
}),
},
)
assert response1.status_code == 201
# Try to create second slab with same SKU
response2 = client.post(
"/api/v1/slabs",
files=[
("files[]", ("topdown.jpg", b"\xff\xd8\xff\xe0", "image/jpeg")),
],
data={
"data": data_json,
"meta[]": json.dumps({
"kind": "inventory",
"role": "topdown",
"seq": 1,
}),
},
)
assert response2.status_code == 409
body = response2.json()
assert body["detail"]["code"] == "duplicate_sku"
app.dependency_overrides.clear()
def test_with_test_client_lifespan(self) -> None:
"""FUND-9: TestClient with app manages lifespan correctly for slabs."""
settings = Settings(
aes_key="dGVzdC1rZXktZm9yLW1hbGxlb3AtdGVzdGluZy1vbmx5LXNob3VsZC1iZS11c2Vk",
woo_base_url="https://test.example.com",
)
app = create_app(settings)
slab_id = str(uuid.uuid4())
data_json = json.dumps({
"id": slab_id,
"sku": "BW-0011",
"length_in": 96.0,
"thickness_in": 1.5,
"sqft": 0.64,
"bdft": 0.96,
"width_min_in": 14.0,
"width_max_in": 22.0,
"width_avg_in": 18.2,
"species_id": 42,
"wood_category_ids": [15, 23],
"edge_type_term_id": 8,
"figure_term_ids": [3],
"grade_term_ids": [7],
"thickness_term_id": 12,
"moisture_term_id": 5,
"fig_tag_ids": [99],
"feat_tag_ids": [42],
"price_per_bdft": 12.50,
"price": 12.00,
"price_source": "recommendation",
"title": None,
"short_title": None,
"description": None,
"content_source": None,
"client_rev": 1,
})
with TestClient(app) as client:
response = client.post(
"/api/v1/slabs",
files=[
("files[]", ("topdown.jpg", b"\xff\xd8\xff\xe0", "image/jpeg")),
],
data={
"data": data_json,
"meta[]": json.dumps({
"kind": "inventory",
"role": "topdown",
"seq": 1,
}),
},
)
assert response.status_code == 201
assert response.json()["status"] == "calibrated"
app.dependency_overrides.clear()