434 lines
14 KiB
Python
434 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine, text
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.database import Base
|
|
from app.domains.auth.models import User, UserRole
|
|
from app.domains.materials.models import AssetLibrary
|
|
from app.domains.media.models import MediaAsset, MediaAssetType
|
|
from app.domains.orders.models import Order, OrderLine, OrderStatus
|
|
from app.domains.products.models import CadFile, Product
|
|
from app.domains.rendering.models import OutputType, RenderTemplate
|
|
from app.domains.rendering.workflow_runtime_services import (
|
|
auto_populate_materials_for_cad,
|
|
resolve_cad_bbox,
|
|
prepare_order_line_render_context,
|
|
resolve_order_line_material_map,
|
|
resolve_order_line_template_context,
|
|
)
|
|
|
|
import app.models # noqa: F401
|
|
|
|
|
|
TEST_DB_URL = os.environ.get(
|
|
"TEST_DATABASE_URL",
|
|
"postgresql+asyncpg://hartomat:hartomat@localhost:5432/hartomat_test",
|
|
).replace("+asyncpg", "")
|
|
|
|
|
|
@pytest.fixture
|
|
def sync_session():
|
|
engine = create_engine(TEST_DB_URL)
|
|
with engine.begin() as conn:
|
|
Base.metadata.create_all(conn)
|
|
|
|
session = Session(engine)
|
|
try:
|
|
yield session
|
|
finally:
|
|
session.close()
|
|
with engine.begin() as conn:
|
|
conn.execute(text("DROP SCHEMA public CASCADE"))
|
|
conn.execute(text("CREATE SCHEMA public"))
|
|
engine.dispose()
|
|
|
|
|
|
def _seed_order_line_graph(session: Session, tmp_path: Path) -> OrderLine:
|
|
step_path = tmp_path / "parts" / "bearing.step"
|
|
step_path.parent.mkdir(parents=True, exist_ok=True)
|
|
step_path.write_text("STEP", encoding="utf-8")
|
|
|
|
user = User(
|
|
id=uuid.uuid4(),
|
|
email=f"workflow-{uuid.uuid4().hex[:8]}@test.local",
|
|
password_hash="hash",
|
|
full_name="Workflow Tester",
|
|
role=UserRole.admin,
|
|
is_active=True,
|
|
)
|
|
cad_file = CadFile(
|
|
id=uuid.uuid4(),
|
|
original_name="bearing.step",
|
|
stored_path=str(step_path),
|
|
file_hash=f"hash-{uuid.uuid4().hex}",
|
|
parsed_objects={"objects": ["InnerRing", "OuterRing"]},
|
|
)
|
|
product = Product(
|
|
id=uuid.uuid4(),
|
|
pim_id="P-1000",
|
|
name="Bearing A",
|
|
category_key="bearings",
|
|
cad_file_id=cad_file.id,
|
|
cad_file=cad_file,
|
|
cad_part_materials=[
|
|
{"part_name": "InnerRing", "material": "Steel raw"},
|
|
{"part_name": "OuterRing", "material": "Steel raw"},
|
|
],
|
|
)
|
|
output_type = OutputType(
|
|
id=uuid.uuid4(),
|
|
name=f"Still-{uuid.uuid4().hex[:6]}",
|
|
renderer="blender",
|
|
output_format="png",
|
|
render_settings={"width": 1600, "height": 900},
|
|
material_override=None,
|
|
)
|
|
order = Order(
|
|
id=uuid.uuid4(),
|
|
order_number=f"ORD-{uuid.uuid4().hex[:8]}",
|
|
status=OrderStatus.processing,
|
|
created_by=user.id,
|
|
)
|
|
line = OrderLine(
|
|
id=uuid.uuid4(),
|
|
order_id=order.id,
|
|
product_id=product.id,
|
|
product=product,
|
|
output_type_id=output_type.id,
|
|
output_type=output_type,
|
|
render_status="pending",
|
|
)
|
|
|
|
session.add_all([user, cad_file, product, output_type, order, line])
|
|
session.commit()
|
|
return line
|
|
|
|
|
|
def test_prepare_order_line_render_context_marks_line_processing_and_prefers_usd(sync_session, tmp_path, monkeypatch):
|
|
from app.config import settings
|
|
|
|
monkeypatch.setattr(settings, "upload_dir", str(tmp_path / "uploads"))
|
|
upload_dir = Path(settings.upload_dir)
|
|
upload_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
line = _seed_order_line_graph(sync_session, tmp_path)
|
|
usd_asset_path = upload_dir / "usd" / "bearing.usd"
|
|
usd_asset_path.parent.mkdir(parents=True, exist_ok=True)
|
|
usd_asset_path.write_text("USD", encoding="utf-8")
|
|
|
|
sync_session.add(
|
|
MediaAsset(
|
|
id=uuid.uuid4(),
|
|
cad_file_id=line.product.cad_file_id,
|
|
product_id=line.product_id,
|
|
asset_type=MediaAssetType.usd_master,
|
|
storage_key="usd/bearing.usd",
|
|
)
|
|
)
|
|
sync_session.commit()
|
|
|
|
messages: list[str] = []
|
|
|
|
result = prepare_order_line_render_context(
|
|
sync_session,
|
|
str(line.id),
|
|
emit=lambda order_line_id, message, level=None: messages.append(message),
|
|
)
|
|
|
|
sync_session.refresh(line)
|
|
|
|
assert result.is_ready
|
|
assert result.usd_render_path == usd_asset_path
|
|
assert result.glb_reuse_path is None
|
|
assert result.part_colors == {
|
|
"InnerRing": "Steel raw",
|
|
"OuterRing": "Steel raw",
|
|
}
|
|
assert line.render_status == "processing"
|
|
assert line.render_backend_used == "celery"
|
|
assert line.render_started_at is not None
|
|
assert any("Using USD master for render" in message for message in messages)
|
|
|
|
|
|
def test_prepare_order_line_render_context_skips_closed_orders(sync_session, tmp_path, monkeypatch):
|
|
from app.config import settings
|
|
|
|
monkeypatch.setattr(settings, "upload_dir", str(tmp_path / "uploads"))
|
|
line = _seed_order_line_graph(sync_session, tmp_path)
|
|
line.order.status = OrderStatus.completed
|
|
sync_session.commit()
|
|
|
|
result = prepare_order_line_render_context(sync_session, str(line.id))
|
|
sync_session.refresh(line)
|
|
|
|
assert result.status == "skip"
|
|
assert result.reason == "order_closed"
|
|
assert line.render_status == "cancelled"
|
|
|
|
|
|
def test_resolve_order_line_template_context_uses_exact_template_and_override(sync_session, tmp_path, monkeypatch):
|
|
from app.config import settings
|
|
|
|
monkeypatch.setattr(settings, "upload_dir", str(tmp_path / "uploads"))
|
|
line = _seed_order_line_graph(sync_session, tmp_path)
|
|
line.material_override = "HARTOMAT_OVERRIDE"
|
|
|
|
sync_session.add(
|
|
AssetLibrary(
|
|
id=uuid.uuid4(),
|
|
name="Default Library",
|
|
blend_file_path="/libraries/materials.blend",
|
|
is_active=True,
|
|
)
|
|
)
|
|
template = RenderTemplate(
|
|
id=uuid.uuid4(),
|
|
name="Bearing Studio",
|
|
category_key="bearings",
|
|
blend_file_path="/templates/bearing.blend",
|
|
original_filename="bearing.blend",
|
|
target_collection="Product",
|
|
material_replace_enabled=False,
|
|
lighting_only=True,
|
|
is_active=True,
|
|
output_types=[line.output_type],
|
|
)
|
|
sync_session.add(template)
|
|
sync_session.commit()
|
|
|
|
monkeypatch.setattr(
|
|
"app.domains.rendering.workflow_runtime_services.resolve_material_map",
|
|
lambda raw_map: {key: f"resolved:{value}" for key, value in raw_map.items()},
|
|
)
|
|
|
|
setup = prepare_order_line_render_context(sync_session, str(line.id))
|
|
result = resolve_order_line_template_context(sync_session, setup)
|
|
|
|
assert result.template is not None
|
|
assert result.template.name == "Bearing Studio"
|
|
assert result.material_library == "/libraries/materials.blend"
|
|
assert result.override_material == "HARTOMAT_OVERRIDE"
|
|
assert result.use_materials is True
|
|
assert result.material_map == {
|
|
"InnerRing": "HARTOMAT_OVERRIDE",
|
|
"OuterRing": "HARTOMAT_OVERRIDE",
|
|
}
|
|
|
|
|
|
def test_resolve_order_line_material_map_disables_materials_when_template_blocks_replacement(
|
|
sync_session,
|
|
tmp_path,
|
|
monkeypatch,
|
|
):
|
|
from app.config import settings
|
|
|
|
monkeypatch.setattr(settings, "upload_dir", str(tmp_path / "uploads"))
|
|
line = _seed_order_line_graph(sync_session, tmp_path)
|
|
template = RenderTemplate(
|
|
id=uuid.uuid4(),
|
|
name="Lighting Only",
|
|
category_key="bearings",
|
|
blend_file_path="/templates/lighting-only.blend",
|
|
original_filename="lighting-only.blend",
|
|
target_collection="Product",
|
|
material_replace_enabled=False,
|
|
lighting_only=True,
|
|
is_active=True,
|
|
)
|
|
|
|
result = resolve_order_line_material_map(
|
|
line,
|
|
line.product.cad_file,
|
|
line.product.cad_part_materials,
|
|
material_library="/libraries/materials.blend",
|
|
template=template,
|
|
)
|
|
|
|
assert result.use_materials is False
|
|
assert result.material_map is None
|
|
assert result.override_material is None
|
|
|
|
|
|
def test_resolve_order_line_material_map_prefers_line_override_over_output_override(
|
|
sync_session,
|
|
tmp_path,
|
|
monkeypatch,
|
|
):
|
|
from app.config import settings
|
|
|
|
monkeypatch.setattr(settings, "upload_dir", str(tmp_path / "uploads"))
|
|
line = _seed_order_line_graph(sync_session, tmp_path)
|
|
line.material_override = "LINE_OVERRIDE"
|
|
line.output_type.material_override = "OUTPUT_OVERRIDE"
|
|
sync_session.commit()
|
|
|
|
result = resolve_order_line_material_map(
|
|
line,
|
|
line.product.cad_file,
|
|
line.product.cad_part_materials,
|
|
material_library="/libraries/materials.blend",
|
|
template=None,
|
|
)
|
|
|
|
assert result.override_material == "LINE_OVERRIDE"
|
|
assert result.use_materials is True
|
|
assert result.material_map == {
|
|
"InnerRing": "LINE_OVERRIDE",
|
|
"OuterRing": "LINE_OVERRIDE",
|
|
}
|
|
|
|
|
|
def test_auto_populate_materials_for_cad_updates_only_blank_products_and_queues_once(sync_session, tmp_path):
|
|
line = _seed_order_line_graph(sync_session, tmp_path)
|
|
cad_file = line.product.cad_file
|
|
assert cad_file is not None
|
|
|
|
line.product.components = [
|
|
{"part_name": "InnerRing", "material": "Steel"},
|
|
{"part_name": "OuterRing", "material": "Rubber"},
|
|
]
|
|
line.product.cad_part_materials = []
|
|
|
|
second_product = Product(
|
|
id=uuid.uuid4(),
|
|
pim_id="P-2000",
|
|
name="Bearing B",
|
|
category_key="bearings",
|
|
cad_file_id=cad_file.id,
|
|
cad_file=cad_file,
|
|
components=[
|
|
{"part_name": "InnerRing", "material": "Brass"},
|
|
{"part_name": "OuterRing", "material": "Copper"},
|
|
],
|
|
cad_part_materials=[{"part_name": "InnerRing", "material": "Existing"}],
|
|
)
|
|
sync_session.add(second_product)
|
|
sync_session.commit()
|
|
|
|
queued: list[tuple[str, dict[str, str]]] = []
|
|
|
|
result = auto_populate_materials_for_cad(
|
|
sync_session,
|
|
str(cad_file.id),
|
|
enqueue_thumbnail=lambda current_cad_file_id, part_colors: queued.append(
|
|
(current_cad_file_id, part_colors)
|
|
),
|
|
)
|
|
|
|
sync_session.refresh(line.product)
|
|
sync_session.refresh(second_product)
|
|
|
|
assert result.updated_product_ids == [str(line.product.id)]
|
|
assert result.queued_thumbnail_regeneration is True
|
|
assert result.part_colors == {"InnerRing": "Steel", "OuterRing": "Rubber"}
|
|
assert queued == [(str(cad_file.id), {"InnerRing": "Steel", "OuterRing": "Rubber"})]
|
|
assert line.product.cad_part_materials == [
|
|
{"part_name": "InnerRing", "material": "Steel"},
|
|
{"part_name": "OuterRing", "material": "Rubber"},
|
|
]
|
|
assert second_product.cad_part_materials == [{"part_name": "InnerRing", "material": "Existing"}]
|
|
|
|
|
|
def test_auto_populate_materials_for_cad_skips_when_materials_already_present(sync_session, tmp_path):
|
|
line = _seed_order_line_graph(sync_session, tmp_path)
|
|
cad_file = line.product.cad_file
|
|
assert cad_file is not None
|
|
|
|
line.product.components = [
|
|
{"part_name": "InnerRing", "material": "Steel"},
|
|
{"part_name": "OuterRing", "material": "Rubber"},
|
|
]
|
|
sync_session.commit()
|
|
|
|
queued: list[tuple[str, dict[str, str]]] = []
|
|
|
|
result = auto_populate_materials_for_cad(
|
|
sync_session,
|
|
str(cad_file.id),
|
|
enqueue_thumbnail=lambda current_cad_file_id, part_colors: queued.append(
|
|
(current_cad_file_id, part_colors)
|
|
),
|
|
)
|
|
|
|
sync_session.refresh(line.product)
|
|
|
|
assert result.updated_product_ids == []
|
|
assert result.queued_thumbnail_regeneration is False
|
|
assert result.part_colors is None
|
|
assert queued == []
|
|
assert line.product.cad_part_materials == [
|
|
{"part_name": "InnerRing", "material": "Steel raw"},
|
|
{"part_name": "OuterRing", "material": "Steel raw"},
|
|
]
|
|
|
|
|
|
def test_resolve_cad_bbox_prefers_glb_over_step(monkeypatch):
|
|
monkeypatch.setattr(
|
|
"app.domains.rendering.workflow_runtime_services.extract_bbox_from_glb",
|
|
lambda path: {
|
|
"dimensions_mm": {"x": 10.0, "y": 20.0, "z": 30.0},
|
|
"bbox_center_mm": {"x": 1.0, "y": 2.0, "z": 3.0},
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
"app.domains.rendering.workflow_runtime_services.extract_bbox_from_step_cadquery",
|
|
lambda path: {
|
|
"dimensions_mm": {"x": 100.0, "y": 200.0, "z": 300.0},
|
|
"bbox_center_mm": {"x": 10.0, "y": 20.0, "z": 30.0},
|
|
},
|
|
)
|
|
|
|
result = resolve_cad_bbox("/tmp/model.step", glb_path="/tmp/model_thumbnail.glb")
|
|
|
|
assert result.source_kind == "glb"
|
|
assert result.bbox_data == {
|
|
"dimensions_mm": {"x": 10.0, "y": 20.0, "z": 30.0},
|
|
"bbox_center_mm": {"x": 1.0, "y": 2.0, "z": 3.0},
|
|
}
|
|
|
|
|
|
def test_resolve_cad_bbox_falls_back_to_step(monkeypatch):
|
|
monkeypatch.setattr(
|
|
"app.domains.rendering.workflow_runtime_services.extract_bbox_from_glb",
|
|
lambda path: None,
|
|
)
|
|
monkeypatch.setattr(
|
|
"app.domains.rendering.workflow_runtime_services.extract_bbox_from_step_cadquery",
|
|
lambda path: {
|
|
"dimensions_mm": {"x": 100.0, "y": 200.0, "z": 300.0},
|
|
"bbox_center_mm": {"x": 10.0, "y": 20.0, "z": 30.0},
|
|
},
|
|
)
|
|
|
|
result = resolve_cad_bbox("/tmp/model.step", glb_path="/tmp/model_thumbnail.glb")
|
|
|
|
assert result.source_kind == "step"
|
|
assert result.bbox_data == {
|
|
"dimensions_mm": {"x": 100.0, "y": 200.0, "z": 300.0},
|
|
"bbox_center_mm": {"x": 10.0, "y": 20.0, "z": 30.0},
|
|
}
|
|
|
|
|
|
def test_extract_metadata_bbox_wrappers_delegate_to_runtime_services(monkeypatch):
|
|
from app.domains.pipeline.tasks.extract_metadata import _bbox_from_glb, _bbox_from_step_cadquery
|
|
|
|
monkeypatch.setattr(
|
|
"app.domains.rendering.workflow_runtime_services.extract_bbox_from_glb",
|
|
lambda path: {"dimensions_mm": {"x": 1.0, "y": 2.0, "z": 3.0}},
|
|
)
|
|
monkeypatch.setattr(
|
|
"app.domains.rendering.workflow_runtime_services.extract_bbox_from_step_cadquery",
|
|
lambda path: {"dimensions_mm": {"x": 4.0, "y": 5.0, "z": 6.0}},
|
|
)
|
|
|
|
assert _bbox_from_glb("/tmp/a.glb") == {"dimensions_mm": {"x": 1.0, "y": 2.0, "z": 3.0}}
|
|
assert _bbox_from_step_cadquery("/tmp/a.step") == {
|
|
"dimensions_mm": {"x": 4.0, "y": 5.0, "z": 6.0}
|
|
}
|