from __future__ import annotations import os import uuid from pathlib import Path import pytest from sqlalchemy import select, text from sqlalchemy.orm import Session 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, build_order_line_render_invocation, emit_order_line_render_notifications, MaterialResolutionResult, OrderLineRenderSetupResult, persist_order_line_media_asset, persist_order_line_output, resolve_cad_bbox, prepare_order_line_render_context, resolve_order_line_material_map, resolve_order_line_template_context, RenderPositionContext, TemplateResolutionResult, ) from app.domains.tenants.models import Tenant from tests.db_test_utils import sync_test_session as sync_test_session_ctx @pytest.fixture def sync_session(): with sync_test_session_ctx() as session: yield session 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) line.product.cad_file.resolved_material_assignments = { "inner_ring": { "source_name": "InnerRing", "prim_path": "/Root/Assembly/inner_ring", "canonical_material": "HARTOMAT_010101_Steel-Bare", } } 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_queues_refresh_for_legacy_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) line.product.cad_file.resolved_material_assignments = { "inner_ring": { "source_name": "InnerRing", "prim_path": "/Root/Assembly/inner_ring", "canonical_material": "SCHAEFFLER_010101_Steel-Bare", } } 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") glb_asset_path = upload_dir / "step_files" / "bearing_thumbnail.glb" glb_asset_path.parent.mkdir(parents=True, exist_ok=True) glb_asset_path.write_text("GLB", encoding="utf-8") sync_session.add_all( [ 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", ), MediaAsset( id=uuid.uuid4(), cad_file_id=line.product.cad_file_id, product_id=line.product_id, asset_type=MediaAssetType.gltf_geometry, storage_key="step_files/bearing_thumbnail.glb", ), ] ) sync_session.commit() queued: list[str] = [] messages: list[str] = [] class _Task: @staticmethod def delay(cad_file_id: str) -> None: queued.append(cad_file_id) monkeypatch.setattr( "app.tasks.step_tasks.generate_usd_master_task", _Task(), ) 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) expected_glb = tmp_path / "parts" / "bearing_thumbnail.glb" assert result.is_ready assert result.usd_render_path is None assert result.glb_reuse_path == expected_glb assert expected_glb.exists() assert queued == [str(line.product.cad_file_id)] assert any("stale" in message for message in messages) assert any("Queued USD master regeneration" in message for message in messages) assert any("Reusing cached GLB geometry" in message for message in messages) assert line.render_status == "processing" 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_build_order_line_render_invocation_applies_output_and_line_overrides(tmp_path): step_path = tmp_path / "parts" / "bearing.step" step_path.parent.mkdir(parents=True, exist_ok=True) step_path.write_text("STEP", encoding="utf-8") output_type = OutputType( id=uuid.uuid4(), name="Still Preview", renderer="blender", output_format="png", render_settings={"width": 1600, "height": 900}, transparent_bg=False, cycles_device="cpu", ) output_type.invocation_overrides = { "engine": "cycles", "samples": 128, "bg_color": "#202020", "turntable_axis": "world_y", "noise_threshold": "0.05", } cad_file = CadFile( id=uuid.uuid4(), original_name="bearing.step", stored_path=str(step_path), file_hash="hash-1", 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, ) line = OrderLine( id=uuid.uuid4(), order_id=uuid.uuid4(), product_id=product.id, product=product, output_type_id=output_type.id, output_type=output_type, render_overrides={ "height": 800, "samples": 48, "transparent_bg": True, "cycles_device": "cuda", "denoiser": "OPENIMAGEDENOISE", "output_format": "webp", }, ) setup = OrderLineRenderSetupResult( status="ready", order_line=line, cad_file=cad_file, part_colors={"InnerRing": "Steel raw", "OuterRing": "Steel raw"}, ) template = RenderTemplate( id=uuid.uuid4(), name="Studio", blend_file_path="/templates/studio.blend", original_filename="studio.blend", target_collection="Assembly", lighting_only=True, shadow_catcher_enabled=True, camera_orbit=False, ) invocation = build_order_line_render_invocation( setup, template_context=TemplateResolutionResult( template=template, material_library="/libraries/materials.blend", material_map={"InnerRing": "SteelPolished"}, use_materials=True, override_material="Studio White", category_key="bearings", output_type_id=str(output_type.id), ), position_context=RenderPositionContext( rotation_x=12.0, rotation_y=24.0, rotation_z=36.0, focal_length_mm=50.0, sensor_width_mm=36.0, ), ) assert invocation.output_extension == "webp" assert invocation.output_filename.endswith(".webp") assert invocation.width == 1600 assert invocation.height == 800 assert invocation.engine == "cycles" assert invocation.samples == 48 assert invocation.noise_threshold == "0.05" assert invocation.denoiser == "OPENIMAGEDENOISE" assert invocation.transparent_bg is True assert invocation.cycles_device == "cuda" assert invocation.bg_color == "#202020" assert invocation.turntable_axis == "world_z" assert invocation.template_path == "/templates/studio.blend" assert invocation.target_collection == "Assembly" assert invocation.material_library_path == "/libraries/materials.blend" assert invocation.material_map == {"InnerRing": "SteelPolished"} assert invocation.material_override == "Studio White" assert invocation.lighting_only is True assert invocation.shadow_catcher is True assert invocation.camera_orbit is False assert invocation.part_names_ordered == ["InnerRing", "OuterRing"] assert invocation.rotation_x == 12.0 assert invocation.focal_length_mm == 50.0 still_kwargs = invocation.as_still_renderer_kwargs( step_path=str(step_path), output_path=str(tmp_path / "renders" / "bearing.webp"), job_id="job-1", order_line_id="line-1", ) assert still_kwargs["step_path"] == str(step_path) assert still_kwargs["output_path"].endswith("bearing.webp") assert still_kwargs["width"] == 1600 assert still_kwargs["height"] == 800 assert still_kwargs["engine"] == "cycles" assert still_kwargs["samples"] == 48 assert still_kwargs["cycles_device"] == "cuda" assert still_kwargs["material_library_path"] == "/libraries/materials.blend" assert still_kwargs["material_override"] == "Studio White" assert still_kwargs["job_id"] == "job-1" assert still_kwargs["order_line_id"] == "line-1" def test_build_order_line_render_invocation_autoscales_samples_and_prefers_material_context( tmp_path, ): step_path = tmp_path / "parts" / "bearing.step" step_path.parent.mkdir(parents=True, exist_ok=True) step_path.write_text("STEP", encoding="utf-8") output_type = OutputType( id=uuid.uuid4(), name="Still Preview", renderer="blender", output_format="png", render_settings={"width": 1024, "height": 512}, ) output_type.invocation_overrides = {"samples": 128, "engine": "eevee"} cad_file = CadFile( id=uuid.uuid4(), original_name="bearing.step", stored_path=str(step_path), file_hash="hash-2", parsed_objects={"objects": ["InnerRing", "OuterRing"]}, ) product = Product( id=uuid.uuid4(), pim_id="P-1001", name="Bearing B", category_key="bearings", cad_file_id=cad_file.id, cad_file=cad_file, ) line = OrderLine( id=uuid.uuid4(), order_id=uuid.uuid4(), product_id=product.id, product=product, output_type_id=output_type.id, output_type=output_type, ) setup = OrderLineRenderSetupResult( status="ready", order_line=line, cad_file=cad_file, part_colors={"InnerRing": "Steel raw"}, ) template = RenderTemplate( id=uuid.uuid4(), name="Studio", blend_file_path="/templates/studio.blend", original_filename="studio.blend", target_collection="Product", ) invocation = build_order_line_render_invocation( setup, template_context=TemplateResolutionResult( template=template, material_library="/libraries/materials.blend", material_map={"InnerRing": "TemplateSteel"}, use_materials=True, override_material="Template White", category_key="bearings", output_type_id=str(output_type.id), ), material_context=MaterialResolutionResult( material_map={"InnerRing": "ResolvedSteel"}, use_materials=False, override_material="Resolved White", source_material_count=2, resolved_material_count=1, ), ) assert invocation.engine == "eevee" assert invocation.samples == 64 assert invocation.material_map == {"InnerRing": "ResolvedSteel"} assert invocation.material_override == "Resolved White" assert invocation.material_library_path is None turntable_kwargs = invocation.as_turntable_renderer_kwargs( step_path=step_path, output_path=tmp_path / "renders" / "bearing.mp4", smooth_angle=30, default_width=1920, default_height=1920, default_engine="cycles", default_samples=256, ) cinematic_kwargs = invocation.as_cinematic_renderer_kwargs( step_path=step_path, output_path=tmp_path / "renders" / "bearing-cinematic.mp4", smooth_angle=30, default_width=1920, default_height=1080, default_engine="cycles", default_samples=256, ) assert turntable_kwargs["width"] == 1024 assert turntable_kwargs["height"] == 512 assert turntable_kwargs["engine"] == "eevee" assert turntable_kwargs["samples"] == 64 assert turntable_kwargs["material_map"] == {"InnerRing": "ResolvedSteel"} assert turntable_kwargs["material_library_path"] is None assert cinematic_kwargs["width"] == 1024 assert cinematic_kwargs["height"] == 512 assert cinematic_kwargs["engine"] == "eevee" assert cinematic_kwargs["samples"] == 64 assert cinematic_kwargs["material_override"] == "Resolved White" 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" material_library_path = tmp_path / "libraries" / "materials.blend" material_library_path.parent.mkdir(parents=True, exist_ok=True) material_library_path.write_text("BLEND", encoding="utf-8") sync_session.add( AssetLibrary( id=uuid.uuid4(), name="Default Library", blend_file_path=str(material_library_path), 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 == str(material_library_path) 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_template_context_supports_explicit_template_and_library_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) template = RenderTemplate( id=uuid.uuid4(), name="Forced Template", category_key=None, blend_file_path="/templates/forced.blend", original_filename="forced.blend", target_collection="ForcedCollection", material_replace_enabled=True, lighting_only=False, is_active=True, ) 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, template_id_override=str(template.id), material_library_path_override="/custom/library.blend", require_template=True, ) assert result.template is not None assert result.template.id == template.id assert result.material_library == "/custom/library.blend" assert result.use_materials is True assert result.material_map == { "InnerRing": "resolved:Steel raw", "OuterRing": "resolved:Steel raw", } def test_resolve_order_line_template_context_can_disable_material_resolution(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) setup = prepare_order_line_render_context(sync_session, str(line.id)) result = resolve_order_line_template_context( sync_session, setup, disable_materials=True, ) assert result.use_materials is False assert result.material_map is None 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_resolve_order_line_material_map_prefers_authoritative_scene_manifest_assignments( 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) cad_file = line.product.cad_file assert cad_file is not None cad_file.resolved_material_assignments = { "inner_ring": { "source_name": "InnerRing", "prim_path": "/Root/Assembly/inner_ring", "canonical_material": "HARTOMAT_010101_Steel-Bare", }, "outer_ring": { "source_name": "OuterRing", "prim_path": "/Root/Assembly/outer_ring", "canonical_material": "HARTOMAT_020202_Rubber-Black", }, } cad_file.manual_material_overrides = { "outer_ring": "HARTOMAT_020203_Rubber-Black-Gloss", } line.product.cad_part_materials = [ {"part_name": "InnerRing", "material": "Legacy Steel"}, {"part_name": "OuterRing", "material": "Legacy Rubber"}, ] sync_session.commit() result = resolve_order_line_material_map( line, cad_file, line.product.cad_part_materials, material_library="/libraries/materials.blend", template=None, ) assert result.use_materials is True assert result.source_material_count == 4 assert result.material_map == { "InnerRing": "HARTOMAT_010101_Steel-Bare", "inner_ring": "HARTOMAT_010101_Steel-Bare", "OuterRing": "HARTOMAT_020203_Rubber-Black-Gloss", "outer_ring": "HARTOMAT_020203_Rubber-Black-Gloss", } def test_resolve_order_line_material_map_keeps_legacy_source_name_fallback_without_scene_manifest( 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) 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.use_materials is True assert result.source_material_count == 2 assert result.material_map == { "InnerRing": "Steel raw", "OuterRing": "Steel raw", } def test_resolve_order_line_material_map_allows_node_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, material_override="NODE_OVERRIDE", ) assert result.override_material == "NODE_OVERRIDE" assert result.use_materials is True assert result.material_map == { "InnerRing": "NODE_OVERRIDE", "OuterRing": "NODE_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_auto_populate_materials_for_cad_can_rewrite_existing_assignments(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": "Ceramic"}, {"part_name": "OuterRing", "material": "Polymer"}, ] sync_session.commit() result = auto_populate_materials_for_cad( sync_session, str(cad_file.id), include_populated_products=True, ) sync_session.refresh(line.product) assert result.updated_product_ids == [str(line.product.id)] assert line.product.cad_part_materials == [ {"part_name": "InnerRing", "material": "Ceramic"}, {"part_name": "OuterRing", "material": "Polymer"}, ] 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} } def test_persist_order_line_output_saves_success_and_creates_media_asset(sync_session, tmp_path, monkeypatch): from app.config import settings upload_dir = tmp_path / "uploads" monkeypatch.setattr(settings, "upload_dir", str(upload_dir)) line = _seed_order_line_graph(sync_session, tmp_path) output_path = upload_dir / "renders" / str(line.id) / "bearing.png" output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_text("PNGDATA", encoding="utf-8") result = persist_order_line_output( sync_session, line, success=True, output_path=str(output_path), render_log={ "renderer": "blender", "engine": "cycles", "engine_used": "cycles", "samples": 64, "total_duration_s": 1.23, }, ) sync_session.refresh(line) asset = sync_session.execute( select(MediaAsset).where(MediaAsset.order_line_id == line.id) ).scalar_one_or_none() assert result.status == "completed" assert result.result_path == str(output_path) assert result.storage_key == f"renders/{line.id}/bearing.png" assert result.asset_type == MediaAssetType.still assert line.render_status == "completed" assert line.result_path == str(output_path) assert asset is not None assert asset.storage_key == f"renders/{line.id}/bearing.png" assert asset.asset_type == MediaAssetType.still assert asset.file_size_bytes == output_path.stat().st_size assert asset.mime_type == "image/png" assert asset.render_config == { "renderer": "blender", "engine": "cycles", "engine_used": "cycles", "samples": 64, "total_duration_s": 1.23, } def test_persist_order_line_output_reuses_existing_asset(sync_session, tmp_path, monkeypatch): from app.config import settings upload_dir = tmp_path / "uploads" monkeypatch.setattr(settings, "upload_dir", str(upload_dir)) line = _seed_order_line_graph(sync_session, tmp_path) output_path = upload_dir / "renders" / str(line.id) / "bearing.mp4" output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_text("MP4DATA", encoding="utf-8") existing = MediaAsset( id=uuid.uuid4(), order_line_id=line.id, product_id=line.product_id, asset_type=MediaAssetType.turntable, storage_key=f"renders/{line.id}/bearing.mp4", ) sync_session.add(existing) sync_session.commit() result = persist_order_line_output( sync_session, line, success=True, output_path=str(output_path), render_log={"renderer": "blender"}, ) assets = sync_session.execute( select(MediaAsset).where(MediaAsset.storage_key == f"renders/{line.id}/bearing.mp4") ).scalars().all() assert result.asset_id == str(existing.id) assert result.asset_type == MediaAssetType.turntable assert len(assets) == 1 def test_persist_order_line_output_canonicalizes_step_file_outputs(sync_session, tmp_path, monkeypatch): from app.config import settings upload_dir = tmp_path / "uploads" monkeypatch.setattr(settings, "upload_dir", str(upload_dir)) line = _seed_order_line_graph(sync_session, tmp_path) step_render_path = upload_dir / "step_files" / "renders" / f"line_{line.id}.png" step_render_path.parent.mkdir(parents=True, exist_ok=True) step_render_path.write_text("PNGDATA", encoding="utf-8") existing = MediaAsset( id=uuid.uuid4(), order_line_id=line.id, product_id=line.product_id, asset_type=MediaAssetType.still, storage_key=f"renders/{line.id}/bearing.png", ) sync_session.add(existing) sync_session.commit() result = persist_order_line_output( sync_session, line, success=True, output_path=str(step_render_path), render_log={"renderer": "blender", "engine_used": "cycles"}, workflow_run_id=str(uuid.uuid4()), ) sync_session.refresh(line) expected_path = Path(result.result_path or "") asset = sync_session.execute( select(MediaAsset).where(MediaAsset.id == existing.id) ).scalar_one() assert expected_path.exists() assert expected_path.read_text(encoding="utf-8") == "PNGDATA" assert expected_path.parent == upload_dir / "renders" / str(line.id) assert expected_path.name.startswith("Bearing_A_Still-") assert expected_path.suffix == ".png" assert result.result_path == str(expected_path) assert result.storage_key == f"renders/{line.id}/{expected_path.name}" assert line.result_path == str(expected_path) assert result.asset_id == str(existing.id) assert asset.storage_key == f"renders/{line.id}/{expected_path.name}" def test_persist_order_line_output_classifies_blend_outputs_as_blend_assets(sync_session, tmp_path, monkeypatch): from app.config import settings upload_dir = tmp_path / "uploads" monkeypatch.setattr(settings, "upload_dir", str(upload_dir)) line = _seed_order_line_graph(sync_session, tmp_path) rendered = tmp_path / "exports" / "bearing_production.blend" rendered.parent.mkdir(parents=True, exist_ok=True) rendered.write_text("BLENDDATA", encoding="utf-8") result = persist_order_line_output( sync_session, line, success=True, output_path=str(rendered), render_log={"artifact_type": "blend_production"}, workflow_run_id=str(uuid.uuid4()), ) sync_session.refresh(line) expected_path = Path(result.result_path or "") asset = sync_session.execute( select(MediaAsset).where(MediaAsset.id == uuid.UUID(result.asset_id)) ).scalar_one() assert expected_path.exists() assert expected_path.suffix == ".blend" assert result.asset_type == MediaAssetType.blend_production assert asset.asset_type == MediaAssetType.blend_production assert asset.storage_key == f"renders/{line.id}/{expected_path.name}" assert asset.mime_type == "application/x-blender" def test_persist_order_line_output_checks_order_completion(sync_session, tmp_path, monkeypatch): from app.config import settings upload_dir = tmp_path / "uploads" monkeypatch.setattr(settings, "upload_dir", str(upload_dir)) line = _seed_order_line_graph(sync_session, tmp_path) rendered = tmp_path / "rendered.png" rendered.write_text("PNGDATA", encoding="utf-8") calls: list[str] = [] monkeypatch.setattr( "app.domains.orders.service.check_order_completion", lambda order_id: calls.append(order_id) or True, ) persist_order_line_output( sync_session, line, success=True, output_path=str(rendered), render_log={"renderer": "blender"}, ) assert calls == [str(line.order_id)] def test_persist_order_line_output_marks_failure_without_result_path(sync_session, tmp_path): line = _seed_order_line_graph(sync_session, tmp_path) result = persist_order_line_output( sync_session, line, success=False, output_path=str(tmp_path / "renders" / "failed.png"), render_log={"error": "boom"}, ) sync_session.refresh(line) assets = sync_session.execute( select(MediaAsset).where(MediaAsset.order_line_id == line.id) ).scalars().all() assert result.status == "failed" assert result.result_path is None assert result.asset_id is None assert line.render_status == "failed" assert line.result_path is None assert line.render_log == {"error": "boom"} assert assets == [] def test_persist_order_line_media_asset_creates_blend_asset_without_touching_order_line(sync_session, tmp_path, monkeypatch): from app.config import settings upload_dir = tmp_path / "uploads" monkeypatch.setattr(settings, "upload_dir", str(upload_dir)) line = _seed_order_line_graph(sync_session, tmp_path) line.render_status = "completed" line.result_path = str(upload_dir / "renders" / str(line.id) / "bearing.png") sync_session.commit() output_path = upload_dir / "exports" / str(line.id) / "bearing_production.blend" output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_text("BLENDDATA", encoding="utf-8") result = persist_order_line_media_asset( sync_session, line, success=True, output_path=str(output_path), asset_type=MediaAssetType.blend_production, render_log={"artifact_type": "blend_production"}, ) sync_session.refresh(line) asset = sync_session.execute( select(MediaAsset).where(MediaAsset.storage_key == f"exports/{line.id}/bearing_production.blend") ).scalar_one_or_none() assert result.status == "completed" assert result.result_path == str(output_path) assert result.storage_key == f"exports/{line.id}/bearing_production.blend" assert result.asset_type == MediaAssetType.blend_production assert line.render_status == "completed" assert line.result_path == str(upload_dir / "renders" / str(line.id) / "bearing.png") assert asset is not None assert asset.asset_type == MediaAssetType.blend_production assert asset.mime_type == "application/x-blender" assert asset.file_size_bytes == output_path.stat().st_size assert asset.render_config == {"artifact_type": "blend_production"} def test_emit_order_line_render_notifications_emits_websocket_and_activity( sync_session, tmp_path, ): line = _seed_order_line_graph(sync_session, tmp_path) tenant = Tenant(name="Workflow Tenant", slug=f"workflow-{uuid.uuid4().hex[:8]}") sync_session.add(tenant) sync_session.commit() line.product.cad_file.tenant_id = tenant.id line.product.tenant_id = tenant.id line.order.tenant_id = tenant.id sync_session.commit() websocket_events: list[tuple[str, dict]] = [] activity_events: list[dict] = [] def _capture_websocket(tenant_id: str, event: dict) -> None: websocket_events.append((tenant_id, event)) def _capture_activity(**payload) -> None: activity_events.append(payload) monkeypatch = pytest.MonkeyPatch() monkeypatch.setattr( "app.core.websocket.publish_event_sync", _capture_websocket, ) monkeypatch.setattr( "app.services.notification_service.emit_notification_sync", _capture_activity, ) try: emit_order_line_render_notifications( success=True, order_line_id=str(line.id), tenant_id=str(tenant.id), product_name=line.product.name or "product", output_type_name=line.output_type.name, session=sync_session, line=line, ) finally: monkeypatch.undo() assert websocket_events == [ ( str(tenant.id), { "type": "render_complete", "order_line_id": str(line.id), "order_id": str(line.order_id), "status": "completed", }, ) ] assert activity_events == [ { "actor_user_id": None, "target_user_id": str(line.order.created_by), "action": "render.completed", "entity_type": "order", "entity_id": str(line.order_id), "details": { "order_number": line.order.order_number, "product_name": line.product.name, "output_type": line.output_type.name, }, "channel": "activity", } ] def test_emit_order_line_render_notifications_truncates_failure_error_and_skips_websocket_without_tenant( sync_session, tmp_path, ): line = _seed_order_line_graph(sync_session, tmp_path) activity_events: list[dict] = [] websocket_events: list[tuple[str, dict]] = [] def _capture_websocket(tenant_id: str, event: dict) -> None: websocket_events.append((tenant_id, event)) def _capture_activity(**payload) -> None: activity_events.append(payload) monkeypatch = pytest.MonkeyPatch() monkeypatch.setattr( "app.core.websocket.publish_event_sync", _capture_websocket, ) monkeypatch.setattr( "app.services.notification_service.emit_notification_sync", _capture_activity, ) try: emit_order_line_render_notifications( success=False, order_line_id=str(line.id), product_name=line.product.name or "product", output_type_name=line.output_type.name, render_log={"error": "x" * 400}, session=sync_session, line=line, ) finally: monkeypatch.undo() assert websocket_events == [] assert activity_events == [ { "actor_user_id": None, "target_user_id": str(line.order.created_by), "action": "render.failed", "entity_type": "order", "entity_id": str(line.order_id), "details": { "order_number": line.order.order_number, "product_name": line.product.name, "output_type": line.output_type.name, "error": "x" * 300, }, "channel": "activity", } ] def test_emit_order_line_render_notifications_supports_retry_exhausted_activity_payload(): activity_events: list[dict] = [] def _capture_activity(**payload) -> None: activity_events.append(payload) monkeypatch = pytest.MonkeyPatch() monkeypatch.setattr( "app.services.notification_service.emit_notification_sync", _capture_activity, ) try: emit_order_line_render_notifications( success=False, order_line_id="line-1", order_number="ORD-FAIL", order_creator_id="user-1", product_name="unknown", output_type_name="unknown", render_log={"error": "retry exhausted"}, emit_websocket=False, activity_entity_id=None, ) finally: monkeypatch.undo() assert activity_events == [ { "actor_user_id": None, "target_user_id": "user-1", "action": "render.failed", "entity_type": "order", "entity_id": None, "details": { "order_number": "ORD-FAIL", "product_name": "unknown", "output_type": "unknown", "error": "retry exhausted", }, "channel": "activity", } ]