1924 lines
64 KiB
Python
1924 lines
64 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from PIL import Image, PngImagePlugin
|
|
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 (
|
|
_build_effective_material_lookup,
|
|
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 _write_png_with_metadata(path: Path, *, rgba: tuple[int, int, int, int], date_text: str) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
image = Image.new("RGBA", (8, 8), rgba)
|
|
metadata = PngImagePlugin.PngInfo()
|
|
metadata.add_text("Date", date_text)
|
|
metadata.add_text("Software", "Blender")
|
|
image.save(path, pnginfo=metadata)
|
|
|
|
|
|
def test_effective_material_lookup_keeps_product_assignments_authoritative_and_adds_manifest_aliases():
|
|
cad_file = CadFile(
|
|
id=uuid.uuid4(),
|
|
original_name="bearing.step",
|
|
stored_path="/tmp/bearing.step",
|
|
file_hash=f"hash-{uuid.uuid4().hex}",
|
|
resolved_material_assignments={
|
|
"inner_ring": {
|
|
"source_name": "InnerRing",
|
|
"prim_path": "/Root/Assembly/inner_ring",
|
|
"canonical_material": "HARTOMAT_010101_Steel-Bare",
|
|
},
|
|
"usd_only_part": {
|
|
"source_name": "UsdOnlyPart",
|
|
"prim_path": "/Root/Assembly/usd_only_part",
|
|
"canonical_material": "HARTOMAT_050101_Elastomer-Black",
|
|
},
|
|
},
|
|
)
|
|
|
|
effective = _build_effective_material_lookup(
|
|
cad_file,
|
|
[
|
|
{"part_name": "InnerRing", "material": "Steel raw"},
|
|
],
|
|
)
|
|
|
|
assert effective["InnerRing"] == "Steel raw"
|
|
assert effective["inner_ring"] == "Steel raw"
|
|
assert effective["UsdOnlyPart"] == "HARTOMAT_050101_Elastomer-Black"
|
|
assert effective["usd_only_part"] == "HARTOMAT_050101_Elastomer-Black"
|
|
|
|
|
|
def test_effective_material_lookup_backfills_manifest_part_keys_from_legacy_serialized_names():
|
|
cad_file = CadFile(
|
|
id=uuid.uuid4(),
|
|
original_name="bearing.step",
|
|
stored_path="/tmp/bearing.step",
|
|
file_hash=f"hash-{uuid.uuid4().hex}",
|
|
resolved_material_assignments={
|
|
"rwdr_b_f_802044_tr4_h122bk": {
|
|
"source_name": "RWDR_B_F-802044_TR4_H122BK",
|
|
"prim_path": "/Root/Assembly/rwdr_b_f_802044_tr4_h122bk",
|
|
"canonical_material": "HARTOMAT_010101_Steel-Bare",
|
|
},
|
|
},
|
|
)
|
|
|
|
effective = _build_effective_material_lookup(
|
|
cad_file,
|
|
[
|
|
{"part_name": "RWDR_B_F-802044_TR4_H122B-69186", "material": "Steel--Stahl"},
|
|
],
|
|
)
|
|
|
|
assert effective["RWDR_B_F-802044_TR4_H122B-69186"] == "Steel--Stahl"
|
|
assert effective["RWDR_B_F-802044_TR4_H122BK"] == "Steel--Stahl"
|
|
assert effective["rwdr_b_f_802044_tr4_h122bk"] == "Steel--Stahl"
|
|
|
|
|
|
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(
|
|
"hartomat:canonicalMaterialName\nhartomat:partKey\n",
|
|
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",
|
|
render_config={
|
|
"cache_key": "stephash:0.03:0.05:20.0:materialhash:scriptfingerprint",
|
|
},
|
|
)
|
|
)
|
|
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_queues_refresh_for_legacy_usd_cache_key(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(
|
|
"hartomat:canonicalMaterialName\nhartomat:partKey\n",
|
|
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",
|
|
render_config={
|
|
"cache_key": "stephash:0.03:0.05:20.0:materialhash",
|
|
},
|
|
),
|
|
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] = []
|
|
|
|
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))
|
|
|
|
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)]
|
|
|
|
|
|
def test_prepare_order_line_render_context_accepts_binary_usd_without_literal_hartomat_markers(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_bytes(b"PXR-USDC\x00binary-usd-with-customdata-not-greppable")
|
|
|
|
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",
|
|
render_config={
|
|
"cache_key": "stephash:0.03:0.05:20.0:materialhash:scriptfingerprint",
|
|
},
|
|
)
|
|
)
|
|
sync_session.commit()
|
|
|
|
queued: 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))
|
|
|
|
assert result.is_ready
|
|
assert result.usd_render_path == usd_asset_path
|
|
assert result.glb_reuse_path is None
|
|
assert queued == []
|
|
|
|
|
|
def test_prepare_order_line_render_context_queues_refresh_for_legacy_usd_file_markers(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("legacy-usd-without-hartomat-markers", 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",
|
|
render_config={
|
|
"cache_key": "stephash:0.03:0.05:20.0:materialhash:scriptfingerprint",
|
|
},
|
|
),
|
|
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] = []
|
|
|
|
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))
|
|
|
|
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)]
|
|
|
|
|
|
def test_prepare_order_line_render_context_queues_refresh_for_legacy_usd_material_field(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",
|
|
"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] = []
|
|
|
|
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))
|
|
|
|
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)]
|
|
|
|
|
|
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",
|
|
target_collection="Assembly",
|
|
lighting_only=True,
|
|
shadow_catcher=True,
|
|
camera_orbit=False,
|
|
template_inputs={"studio_variant": "warm"},
|
|
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
|
|
assert invocation.template_inputs == {"studio_variant": "warm"}
|
|
|
|
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["template_inputs"] == {"studio_variant": "warm"}
|
|
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",
|
|
target_collection="Product",
|
|
lighting_only=False,
|
|
shadow_catcher=False,
|
|
camera_orbit=True,
|
|
template_inputs={"studio_variant": "warm"},
|
|
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 turntable_kwargs["template_inputs"] == {"studio_variant": "warm"}
|
|
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"
|
|
assert cinematic_kwargs["template_inputs"] == {"studio_variant": "warm"}
|
|
|
|
|
|
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",
|
|
}
|
|
assert result.target_collection == "ForcedCollection"
|
|
assert result.lighting_only is False
|
|
assert result.shadow_catcher is False
|
|
assert result.camera_orbit is True
|
|
|
|
|
|
def test_resolve_order_line_template_context_applies_template_override_modes(
|
|
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="Overrideable Template",
|
|
category_key="bearings",
|
|
blend_file_path="/templates/overrideable.blend",
|
|
original_filename="overrideable.blend",
|
|
target_collection="TemplateCollection",
|
|
material_replace_enabled=False,
|
|
lighting_only=False,
|
|
shadow_catcher_enabled=False,
|
|
camera_orbit=True,
|
|
is_active=True,
|
|
output_types=[line.output_type],
|
|
)
|
|
sync_session.add(template)
|
|
sync_session.add(
|
|
AssetLibrary(
|
|
id=uuid.uuid4(),
|
|
name="Default Library",
|
|
blend_file_path="/libraries/materials.blend",
|
|
is_active=True,
|
|
)
|
|
)
|
|
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="/libraries/materials.blend",
|
|
target_collection_override="NodeCollection",
|
|
material_replace_mode="enabled",
|
|
lighting_only_mode="enabled",
|
|
shadow_catcher_mode="enabled",
|
|
camera_orbit_mode="disabled",
|
|
)
|
|
|
|
assert result.template is not None
|
|
assert result.use_materials is True
|
|
assert result.material_map == {
|
|
"InnerRing": "resolved:Steel raw",
|
|
"OuterRing": "resolved:Steel raw",
|
|
}
|
|
assert result.target_collection == "NodeCollection"
|
|
assert result.lighting_only is True
|
|
assert result.shadow_catcher is True
|
|
assert result.camera_orbit is False
|
|
|
|
|
|
def test_resolve_order_line_template_context_exposes_template_schema_and_invocation_inputs(
|
|
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="Schema Template",
|
|
category_key="bearings",
|
|
blend_file_path="/templates/schema-template.blend",
|
|
original_filename="schema-template.blend",
|
|
target_collection="Product",
|
|
material_replace_enabled=True,
|
|
lighting_only=False,
|
|
shadow_catcher_enabled=False,
|
|
camera_orbit=True,
|
|
workflow_input_schema=[
|
|
{
|
|
"key": "studio_variant",
|
|
"label": "Studio Variant",
|
|
"type": "select",
|
|
"section": "Template Inputs",
|
|
"default": "default",
|
|
"options": [
|
|
{"value": "default", "label": "Default"},
|
|
{"value": "warm", "label": "Warm"},
|
|
],
|
|
},
|
|
{
|
|
"key": "camera_profile",
|
|
"label": "Camera Profile",
|
|
"type": "text",
|
|
"section": "Template Inputs",
|
|
"default": "macro",
|
|
},
|
|
],
|
|
is_active=True,
|
|
output_types=[line.output_type],
|
|
)
|
|
sync_session.add(template)
|
|
sync_session.add(
|
|
AssetLibrary(
|
|
id=uuid.uuid4(),
|
|
name="Default Library",
|
|
blend_file_path="/libraries/materials.blend",
|
|
is_active=True,
|
|
)
|
|
)
|
|
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))
|
|
template_context = resolve_order_line_template_context(
|
|
sync_session,
|
|
setup,
|
|
template_id_override=str(template.id),
|
|
template_input_overrides={"studio_variant": "warm"},
|
|
)
|
|
invocation = build_order_line_render_invocation(setup, template_context=template_context)
|
|
|
|
assert template_context.workflow_input_schema == template.workflow_input_schema
|
|
assert template_context.template_inputs == {
|
|
"studio_variant": "warm",
|
|
"camera_profile": "macro",
|
|
}
|
|
assert invocation.template_inputs == {
|
|
"studio_variant": "warm",
|
|
"camera_profile": "macro",
|
|
}
|
|
|
|
|
|
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_png_persistence_strips_volatile_metadata_for_primary_and_observer_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)
|
|
|
|
primary_source = upload_dir / "step_files" / "renders" / f"line_{line.id}.png"
|
|
observer_source = upload_dir / "step_files" / "renders" / f"line_{line.id}_shadow.png"
|
|
_write_png_with_metadata(
|
|
primary_source,
|
|
rgba=(12, 34, 56, 255),
|
|
date_text="2026/04/10 17:05:27",
|
|
)
|
|
_write_png_with_metadata(
|
|
observer_source,
|
|
rgba=(12, 34, 56, 255),
|
|
date_text="2026/04/10 17:06:30",
|
|
)
|
|
|
|
primary_result = persist_order_line_output(
|
|
sync_session,
|
|
line,
|
|
success=True,
|
|
output_path=str(primary_source),
|
|
render_log={"renderer": "blender", "engine_used": "cycles"},
|
|
)
|
|
observer_result = persist_order_line_media_asset(
|
|
sync_session,
|
|
line,
|
|
success=True,
|
|
output_path=str(observer_source),
|
|
asset_type=MediaAssetType.still,
|
|
render_log={"renderer": "blender", "engine_used": "cycles"},
|
|
)
|
|
|
|
primary_bytes = Path(primary_result.result_path or "").read_bytes()
|
|
observer_bytes = Path(observer_result.result_path or "").read_bytes()
|
|
|
|
assert primary_bytes == observer_bytes
|
|
assert b"Date" not in primary_bytes
|
|
assert b"Date" not in observer_bytes
|
|
assert Image.open(primary_result.result_path).getpixel((0, 0)) == (12, 34, 56, 255)
|
|
assert Image.open(observer_result.result_path).getpixel((0, 0)) == (12, 34, 56, 255)
|
|
|
|
|
|
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",
|
|
}
|
|
]
|