a70cb55d01
- workflow_builder.py: fix broken stubs, add render_order_line_still_task
(resolves step_path from DB instead of passing order_line_id as step_path)
- domains/rendering/tasks.py: add render_order_line_still_task,
export_gltf_for_order_line_task, export_blend_for_order_line_task,
generate_gltf_geometry_task (trimesh STL→GLB, no Blender needed)
- tasks/step_tasks.py: add generate_gltf_geometry_task for CadFile GLB export
- cad router: POST /{id}/generate-gltf-geometry endpoint (admin/PM)
- worker router: GET /celery-workers + POST /scale (docker compose subprocess)
- Dockerfile: pip install -e "[dev]" to enable pytest
- docker-compose.yml: docker socket + compose file mount on backend
- ThreeDViewer.tsx: mode toggle (geometry/production), wireframe, env presets,
download buttons (GLB + .blend)
- CadPreview.tsx: load gltf_geometry/gltf_production/blend_production assets
from MediaAsset table and pass URLs to ThreeDViewer
- ProductDetail.tsx: "View 3D" button → /cad/:id, "Generate GLB" button
- media router/service: cad_file_id filter on GET /api/media
- WorkerManagement.tsx: new page with worker status, queue depth, scale controls
- App.tsx + Layout.tsx: /workers route + sidebar link (admin/PM)
- tests: test_rendering_service.py, test_orders_service.py (backend)
- tests: WorkerActivity.test.tsx, WorkerManagement.test.tsx (frontend)
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
192 lines
5.7 KiB
Python
192 lines
5.7 KiB
Python
"""Tests for orders domain — order creation, status transitions, and pricing."""
|
|
import uuid
|
|
|
|
import pytest
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def _create_test_product(db):
|
|
from app.domains.products.models import Product
|
|
product = Product(
|
|
id=uuid.uuid4(),
|
|
name=f"Test Product {uuid.uuid4().hex[:6]}",
|
|
category_key="TRB",
|
|
components=[],
|
|
cad_part_materials=[],
|
|
)
|
|
db.add(product)
|
|
await db.commit()
|
|
await db.refresh(product)
|
|
return product
|
|
|
|
|
|
async def _create_test_order(db, user):
|
|
from app.domains.orders.models import Order, OrderStatus
|
|
order = Order(
|
|
id=uuid.uuid4(),
|
|
order_number=f"TEST-{uuid.uuid4().hex[:6].upper()}",
|
|
status=OrderStatus.draft,
|
|
created_by=user.id,
|
|
tenant_id=None,
|
|
)
|
|
db.add(order)
|
|
await db.commit()
|
|
await db.refresh(order)
|
|
return order
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Order creation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_order_draft_status(db, admin_user):
|
|
"""New order starts in draft status."""
|
|
order = await _create_test_order(db, admin_user)
|
|
assert order.id is not None
|
|
assert order.status.value == "draft"
|
|
assert order.order_number.startswith("TEST-")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_order_has_no_lines_initially(db, admin_user):
|
|
"""New order starts with zero order lines."""
|
|
from sqlalchemy import select
|
|
from app.domains.orders.models import Order, OrderLine
|
|
order = await _create_test_order(db, admin_user)
|
|
result = await db.execute(
|
|
select(OrderLine).where(OrderLine.order_id == order.id)
|
|
)
|
|
lines = result.scalars().all()
|
|
assert len(lines) == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Order line creation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_order_line(db, admin_user):
|
|
"""Order line can be added to a draft order."""
|
|
from app.domains.orders.models import OrderLine
|
|
product = await _create_test_product(db)
|
|
order = await _create_test_order(db, admin_user)
|
|
|
|
line = OrderLine(
|
|
id=uuid.uuid4(),
|
|
order_id=order.id,
|
|
product_id=product.id,
|
|
render_status="pending",
|
|
item_status="pending",
|
|
tenant_id=None,
|
|
)
|
|
db.add(line)
|
|
await db.commit()
|
|
await db.refresh(line)
|
|
|
|
assert line.id is not None
|
|
assert line.order_id == order.id
|
|
assert line.render_status == "pending"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Status transitions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_order_status_transition_to_submitted(db, admin_user):
|
|
"""Order status can be changed from draft to submitted."""
|
|
from app.domains.orders.models import Order, OrderStatus
|
|
order = await _create_test_order(db, admin_user)
|
|
order.status = OrderStatus.submitted
|
|
await db.commit()
|
|
await db.refresh(order)
|
|
assert order.status == OrderStatus.submitted
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_order_multiple_lines(db, admin_user):
|
|
"""Multiple lines can be added to the same order."""
|
|
from app.domains.orders.models import OrderLine
|
|
product = await _create_test_product(db)
|
|
order = await _create_test_order(db, admin_user)
|
|
|
|
for _ in range(3):
|
|
line = OrderLine(
|
|
id=uuid.uuid4(),
|
|
order_id=order.id,
|
|
product_id=product.id,
|
|
render_status="pending",
|
|
item_status="pending",
|
|
tenant_id=None,
|
|
)
|
|
db.add(line)
|
|
await db.commit()
|
|
|
|
from sqlalchemy import select
|
|
result = await db.execute(
|
|
select(OrderLine).where(OrderLine.order_id == order.id)
|
|
)
|
|
assert len(result.scalars().all()) == 3
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Render status tracking
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_order_line_render_status_update(db, admin_user):
|
|
"""Order line render_status can be updated to processing/completed."""
|
|
from app.domains.orders.models import OrderLine
|
|
product = await _create_test_product(db)
|
|
order = await _create_test_order(db, admin_user)
|
|
|
|
line = OrderLine(
|
|
id=uuid.uuid4(),
|
|
order_id=order.id,
|
|
product_id=product.id,
|
|
render_status="pending",
|
|
item_status="pending",
|
|
tenant_id=None,
|
|
)
|
|
db.add(line)
|
|
await db.commit()
|
|
|
|
line.render_status = "processing"
|
|
await db.commit()
|
|
await db.refresh(line)
|
|
assert line.render_status == "processing"
|
|
|
|
line.render_status = "completed"
|
|
await db.commit()
|
|
await db.refresh(line)
|
|
assert line.render_status == "completed"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Unit price
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_order_line_unit_price_nullable(db, admin_user):
|
|
"""unit_price defaults to None."""
|
|
from app.domains.orders.models import OrderLine
|
|
product = await _create_test_product(db)
|
|
order = await _create_test_order(db, admin_user)
|
|
|
|
line = OrderLine(
|
|
id=uuid.uuid4(),
|
|
order_id=order.id,
|
|
product_id=product.id,
|
|
render_status="pending",
|
|
item_status="pending",
|
|
tenant_id=None,
|
|
)
|
|
db.add(line)
|
|
await db.commit()
|
|
await db.refresh(line)
|
|
assert line.unit_price is None
|