boundaryoriginal_boundarymeshmesh_build_reportpatchesbuild(resolution=25, skip_boundary=False)
+ build(resolution=25, skip_boundary=False, **interior_kwargs)
Build the implicit function describing the domain and optionally extract boundary/mesh artifacts.
resolution (int, default=25): Grid resolution for boundary extraction.skip_boundary (bool, default=False): If true, only assemble fast-evaluation structures and skip boundary and interior mesh generation.**interior_kwargs: Options forwarded to get_interior(), including the bounded surface-recovery controls below.resolution (int): Grid resolution for marching cubes/squaresget_largest (bool, default=True): Reserved (current implementation always returns the largest connected component)get_largest (bool, default=True): Keep only the largest connected component during initial boundary extraction. Set to false to retain all components.get_interior(verbose=False, **kwargs)
Generate tetrahedral (3D) or triangular (2D) mesh of the interior.
+Generate a tetrahedral (3D) or triangular (2D) interior mesh. In 3D, geometry-related TetGen failures use a validated, component-preserving recovery pipeline.
verbose (bool): Print mesh generation progress**kwargs: Parameters passed to TetGen/Triangleswitches strings are rejected in 3D because Domain enforces linear order=1 elements and nobisect=True. Use named options instead.repair_on_failure (bool, default=True): Retry a geometry rejection after component-preserving MeshFix repair.repair_max_distance_ratio (float, default=0.01): Maximum symmetric repair displacement and bounds change, measured against the source bounding-box diagonal.remesh_on_failure (bool, default=True): Enable validated PyACVD as the final fallback after the original and MeshFix candidates.remesh_subdivisions, remesh_clusters, remesh_clean_tolerance: Configure the optional PyACVD stage.mesh (pv.UnstructuredGrid): Interior mesh+ The original surface is tried first with the caller's TetGen options. A valid + first attempt returns immediately. After a geometry rejection, MeshFix runs with + component joining and component removal disabled. Candidates must remain closed, + manifold, finite, triangular, component-preserving, and within the configured + displacement bound. PyACVD output is checked by the same policy and is repaired + before TetGen when necessary; unsafe candidates are recorded and rejected. +
++ Closed, watertight, and manifold are topological properties; they do not prove + that a surface has no self-intersecting facets. TetGen may therefore reject a + surface that passes those checks. The structured report records intersection + diagnostics from a separate verbose pass when the initial failure is opaque. +
+
+ When recovery succeeds, boundary, boundary sampling arrays, and the
+ extracted volume-mesh surface all describe the selected recovery geometry.
+ original_boundary remains unchanged. Inspect
+ mesh_build_report.selected_strategy and
+ mesh_build_report.attempts for provenance and diagnostics.
+
tetrahedralize(surface, *tet_args, return_result=False, **tet_kwargs)
+
+ Lower-level public surface meshing helper. Unlike Domain.get_interior(),
+ this compatibility API forwards named TetGen options or a raw
+ switches string and can return linear or quadratic tetrahedra.
+
(grid, nodes, elements) tuple.return_result=True: TetrahedralizationResult with grid, nodes, elements, the exact selected surface, and a structured report.
+ The helper tries the original surface, bounded MeshFix repair, and validated
+ PyACVD recovery in order. Set repair_on_failure=False or
+ remesh_on_failure=False to disable a stage. Raw switches are unchanged
+ for real meshing attempts; an isolated diagnostic-only rerun removes quiet mode
+ and adds TetGen diagnostic verbosity when needed.
+
If point-picking does not work, ensure the domain was created/solved/built before loading.
++ Loading a 3D mesh runs the domain create, solve, and build stages. The imported surface is sent to TetGen first. + If TetGen detects a geometry problem, the GUI automatically tries a component-preserving MeshFix repair and then + a validated PyACVD fallback. A recovered surface is accepted only when it is finite, triangular, closed, + manifold, preserves connected components, and stays within the configured geometric-fidelity bound. +
++ A surface can be closed, watertight, and manifold while still containing self-intersecting facets. Those + topology checks alone therefore do not guarantee that TetGen can construct a volume mesh. When this occurs, + the expanded diagnostics identify representative intersecting facets or segments when TetGen reports them. +
+
+ A successful recovery is reported in the status bar as a build completed after surface repair; it does not open
+ a blocking success dialog. The working domain boundary is updated to the exact surface that generated the volume
+ mesh, while the imported surface remains available as original_boundary.
+
+ Python callers can disable individual recovery stages or tighten the default 1% displacement limit with
+ repair_on_failure, remesh_on_failure, and
+ repair_max_distance_ratio. The GUI retains the structured build report for troubleshooting.
+
If importing the GUI fails with _ARRAY_API not found or numpy.core.multiarray failed to import, recreate the environment with a NumPy version supported by your Python version. On Python 3.9–3.12, use python -m pip install --force-reinstall "numpy<2" svv. Python 3.13 requires NumPy 2.1 or newer; use python -m pip install --force-reinstall "numpy>=2.1" svv.
["'])(?:[a-z]:[\\/]|\\\\)[^"']+(?P=quote)''' +) +_QUOTED_UNIX_PATH = re.compile( + r'''(?P["'])/(?!/)[^"']+(?P=quote)''' +) +_WINDOWS_EXTENDED_PATH = re.compile( + r'''(?i)(? str: + """Replace absolute local filesystem paths in diagnostic text.""" + + def replace_quoted(match): + quote = match.group("quote") + return "{}{}".format(quote, quote) + + sanitized = _QUOTED_WINDOWS_PATH.sub(replace_quoted, str(value)) + sanitized = _QUOTED_UNIX_PATH.sub(replace_quoted, sanitized) + sanitized = _WINDOWS_EXTENDED_PATH.sub(" ", sanitized) + sanitized = _WINDOWS_UNC_PATH.sub(" ", sanitized) + sanitized = _WINDOWS_DRIVE_PATH.sub(" ", sanitized) + return _UNIX_PATH.sub(" ", sanitized) + + +def extract_build_report(exception=None, domain=None) -> Optional[TetrahedralizationReport]: + """Find a structured report on an exception or completed Domain.""" + + report = getattr(exception, "report", None) + if report is not None: + return report + attempt = getattr(exception, "attempt", None) + if attempt is not None: + return TetrahedralizationReport( + source=attempt.surface, + attempts=[attempt], + selected_strategy=None, + selected_surface=None, + versions={}, + ) + return getattr(domain, "mesh_build_report", None) + + +def _report_without_duplicate_tracebacks(report): + seen_tracebacks = set() + attempts = [] + for attempt in report.attempts: + diagnostic = attempt.diagnostics + if diagnostic is None: + attempts.append(attempt) + continue + replacements = {} + for stream_name in ("stdout", "stderr"): + stream = getattr(diagnostic, stream_name) + if "traceback (most recent call last)" not in stream.lower(): + continue + signature = stream.strip() + if signature in seen_tracebacks: + replacements[stream_name] = "[duplicate Python traceback omitted]" + else: + seen_tracebacks.add(signature) + if replacements: + diagnostic = replace(diagnostic, **replacements) + attempt = replace(attempt, diagnostics=diagnostic) + attempts.append(attempt) + return TetrahedralizationReport( + source=report.source, + attempts=attempts, + selected_strategy=report.selected_strategy, + selected_surface=report.selected_surface, + versions=dict(report.versions), + ) + + +def build_domain_feedback( + *, + report: Optional[TetrahedralizationReport] = None, + exception=None, + success: bool, +) -> DomainBuildFeedback: + """Format a Domain build result for status and optional warning display.""" + + if report is None: + report = extract_build_report(exception=exception) + + if success: + recovered = bool( + report is not None + and report.selected_strategy + and report.selected_strategy != "original" + ) + if recovered: + strategy = report.selected_strategy + status = "Domain built successfully after surface repair ({})".format( + strategy + ) + informative = report.user_summary() + else: + status = "Domain built successfully" + informative = ( + report.user_summary() + if report is not None + else "The interior mesh was built successfully." + ) + details = report.detailed_text() if report is not None else "" + return DomainBuildFeedback( + text="The domain and its interior mesh were built successfully.", + informative_text=sanitize_local_paths(informative), + detailed_text=sanitize_local_paths(details), + status=status, + recovered=recovered, + ) + + if report is not None: + display_report = _report_without_duplicate_tracebacks(report) + informative = display_report.user_summary() + details = display_report.detailed_text() + else: + informative = ( + "Volume meshing failed. Review the technical details, verify the " + "meshing installation, and inspect the source surface." + ) + details = "{}: {}".format( + type(exception).__name__ if exception is not None else "Error", + exception if exception is not None else "No additional details were recorded.", + ) + return DomainBuildFeedback( + text="The domain loaded, but its interior mesh could not be built.", + informative_text=sanitize_local_paths(informative), + detailed_text=sanitize_local_paths(details), + status="Domain loaded without an interior mesh", + recovered=False, + ) + + +def apply_feedback_to_message_box(message_box, feedback: DomainBuildFeedback) -> None: + """Populate a QMessageBox-compatible object with structured feedback.""" + + box_type = type(message_box) + message_box.setIcon(box_type.Warning) + message_box.setText(feedback.text) + message_box.setInformativeText(feedback.informative_text) + message_box.setDetailedText(feedback.detailed_text) + message_box.setStandardButtons(box_type.Ok) + # Some native dialog backends initialize the platform window while setting + # the icon and clear an earlier title in the process. + message_box.setWindowTitle("Domain Build Warning") + + +def _sanitize_payload(value: Any) -> Any: + if isinstance(value, dict): + return {str(key): _sanitize_payload(item) for key, item in value.items()} + if isinstance(value, (list, tuple)): + return [_sanitize_payload(item) for item in value] + if isinstance(value, str): + return sanitize_local_paths(value) + if isinstance(value, (str, int, float, bool)) or value is None: + return value + return repr(value) + + +def report_for_telemetry(report: Optional[TetrahedralizationReport]) -> Optional[Dict[str, Any]]: + """Return a path-sanitized, JSON-safe report without mesh arrays.""" + + if report is None: + return None + return _sanitize_payload(report.to_dict()) diff --git a/svv/visualize/gui/main_window.py b/svv/visualize/gui/main_window.py index d9990f5..e411389 100644 --- a/svv/visualize/gui/main_window.py +++ b/svv/visualize/gui/main_window.py @@ -30,9 +30,15 @@ from svv.visualize.gui.point_selector import PointSelectorWidget from svv.visualize.gui.parameter_panel import ParameterPanel from svv.visualize.gui.theme import CADTheme, CADIcons +from svv.visualize.gui.domain_build_feedback import ( + apply_feedback_to_message_box, + build_domain_feedback, + extract_build_report, + report_for_telemetry, +) import svv.tree.tree as _svv_tree_mod import svv.forest.forest as _svv_forest_mod -from svv.telemetry import capture_exception, capture_message +from svv.telemetry import capture_exception, capture_message, telemetry_enabled from svv.visualize.spline_export import export_spline_files @@ -2673,7 +2679,8 @@ def show_about(self): ) def _record_telemetry(self, exc=None, message: Optional[str] = None, level: str = "error", - traceback_str: Optional[str] = None, **tags): + traceback_str: Optional[str] = None, + telemetry_context: Optional[dict] = None, **tags): """ Send errors or warnings to telemetry without interrupting the GUI. @@ -2687,10 +2694,14 @@ def _record_telemetry(self, exc=None, message: Optional[str] = None, level: str Sentry level ("error", "warning", "info"). traceback_str : str, optional Full traceback string to include as extra context. + telemetry_context : dict, optional + Sanitized structured context to attach to the event. **tags Additional tags to attach to the event. """ try: + if not telemetry_enabled(): + return if exc is not None: try: import sentry_sdk # type: ignore[import] @@ -2700,6 +2711,8 @@ def _record_telemetry(self, exc=None, message: Optional[str] = None, level: str scope.set_tag(key, value) if traceback_str: scope.set_extra("full_traceback", traceback_str) + if telemetry_context: + scope.set_context("domain_build", telemetry_context) sentry_sdk.capture_exception(exc) # Flush to ensure the event is sent before the popup blocks sentry_sdk.flush(timeout=2.0) @@ -2707,6 +2720,19 @@ def _record_telemetry(self, exc=None, message: Optional[str] = None, level: str capture_exception(exc) return if message: + if telemetry_context: + try: + import sentry_sdk # type: ignore[import] + + with sentry_sdk.push_scope() as scope: + for key, value in tags.items(): + scope.set_tag(key, value) + scope.set_context("domain_build", telemetry_context) + sentry_sdk.capture_message(message, level=level) + sentry_sdk.flush(timeout=2.0) + return + except Exception: + pass capture_message(message, level=level, **tags) except Exception: # Telemetry should never break the UI @@ -3059,6 +3085,8 @@ def _stage_report(progress=None, label=None, indeterminate=None): build_failed = False build_error_msg = None + build_exception = None + build_report = extract_build_report(domain=domain) if mesh_already_loaded: # Mesh was included in .dmn file - skip rebuild @@ -3072,24 +3100,49 @@ def _stage_report(progress=None, label=None, indeterminate=None): domain.build(resolution=build_resolution, progress_callback=build_progress) else: domain.build(progress_callback=build_progress) - report_progress(90, "Domain built successfully") + build_report = extract_build_report(domain=domain) + feedback = build_domain_feedback( + report=build_report, + success=True, + ) + report_progress(90, feedback.status) + if feedback.recovered: + self._record_telemetry( + message=feedback.status, + level="info", + action="load_domain_build_recovered", + telemetry_context={ + "tetrahedralization": report_for_telemetry(build_report) + }, + ) except Exception as exc: # If build fails (e.g., TetGen errors) continue with loaded # fast-eval structures only so tree/forest generation still # works, but record the failure in telemetry for diagnosis. build_failed = True - build_error_msg = str(exc) - try: - import traceback - tb = traceback.format_exc() - self._record_telemetry(exc, action="load_domain_build", traceback_str=tb) - except Exception: - pass + build_exception = exc + build_report = extract_build_report(exception=exc, domain=domain) + feedback = build_domain_feedback( + exception=exc, + report=build_report, + success=False, + ) + build_error_msg = feedback.informative_text + self._record_telemetry( + message=feedback.informative_text, + level="warning", + action="load_domain_build", + telemetry_context={ + "tetrahedralization": report_for_telemetry(build_report) + }, + ) report_progress(90, "Domain build failed (continuing without mesh)") # Store build status on domain for later reference domain._build_failed = build_failed domain._build_error = build_error_msg + domain._build_exception = build_exception + domain._build_report = build_report if cancel_event.is_set(): return None elif suffix in {".vtp", ".vtu", ".stl"}: @@ -3120,27 +3173,54 @@ def _stage_report(progress=None, label=None, indeterminate=None): report_progress(65, "Building domain (tetrahedralization + boundary extraction)...") build_failed = False build_error_msg = None + build_exception = None + build_report = None try: build_progress = make_stage_reporter(65, 90) if build_resolution is not None: domain.build(resolution=build_resolution, progress_callback=build_progress) else: domain.build(progress_callback=build_progress) - report_progress(90, "Domain built successfully") + build_report = extract_build_report(domain=domain) + feedback = build_domain_feedback( + report=build_report, + success=True, + ) + report_progress(90, feedback.status) + if feedback.recovered: + self._record_telemetry( + message=feedback.status, + level="info", + action="load_mesh_build_recovered", + telemetry_context={ + "tetrahedralization": report_for_telemetry(build_report) + }, + ) except Exception as exc: build_failed = True - build_error_msg = str(exc) - try: - import traceback - tb = traceback.format_exc() - self._record_telemetry(exc, action="load_mesh_build", traceback_str=tb) - except Exception: - pass + build_exception = exc + build_report = extract_build_report(exception=exc, domain=domain) + feedback = build_domain_feedback( + exception=exc, + report=build_report, + success=False, + ) + build_error_msg = feedback.informative_text + self._record_telemetry( + message=feedback.informative_text, + level="warning", + action="load_mesh_build", + telemetry_context={ + "tetrahedralization": report_for_telemetry(build_report) + }, + ) report_progress(90, "Domain build failed (continuing without mesh)") # Store build status on domain for later reference domain._build_failed = build_failed domain._build_error = build_error_msg + domain._build_exception = build_exception + domain._build_report = build_report else: raise ValueError(f"Unsupported file type: {suffix}") @@ -3322,19 +3402,24 @@ def _finish_load_future(self): # Warn user if domain build failed (mesh not available) if getattr(result, '_build_failed', False): - error_msg = getattr(result, '_build_error', 'Unknown error') - self._record_telemetry( - message=f"Domain build failed: {error_msg}", - level="warning", - action="domain_build_failed_warning", + build_exception = getattr(result, '_build_exception', None) + build_report = getattr(result, '_build_report', None) + feedback = build_domain_feedback( + exception=build_exception, + report=build_report, + success=False, ) - QMessageBox.warning( - self, - "Domain Build Warning", - f"The domain was loaded but mesh building failed:\n\n{error_msg}\n\n" - f"Some features (like auto-selecting boundary start points) may not work.\n" - f"You can still use the domain by manually selecting start points." + self.update_status(feedback.status) + message_box = QMessageBox(self) + apply_feedback_to_message_box(message_box, feedback) + message_box.exec() + else: + build_report = getattr(result, '_build_report', None) + feedback = build_domain_feedback( + report=build_report, + success=True, ) + self.update_status(feedback.status) # Backwards compatibility alias diff --git a/test/conftest.py b/test/conftest.py index 8ca9cac..6dc484b 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -1,5 +1,45 @@ +import hashlib +import os import sys from pathlib import Path +import pyvista as pv +import pytest + # Add the parent directory to PYTHONPATH sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + + +@pytest.fixture +def closed_self_intersecting_surface(): + """Small closed manifold whose folded cap intersects the lower surface.""" + + surface = pv.Sphere(theta_resolution=20, phi_resolution=20) + surface.points[surface.points[:, 2] > 0.2, 2] -= 0.8 + regions = surface.connectivity().cell_data["RegionId"] + + assert surface.is_all_triangles + assert int(regions.max()) + 1 == 1 + assert surface.is_manifold + assert surface.n_open_edges == 0 + return surface.copy(deep=True) + + +@pytest.fixture +def issue_102_stl_path(): + """Return the verified issue attachment path when explicitly provided.""" + + configured = os.environ.get("SVV_ISSUE_102_STL") + if not configured: + pytest.skip("Set SVV_ISSUE_102_STL to run the issue #102 attachment regression") + + path = Path(configured).expanduser().resolve(strict=True) + digest = hashlib.sha256() + with path.open("rb") as stream: + for block in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(block) + + assert digest.hexdigest() == ( + "ec3d4e23757659604c939e7d2f418587bfedc2a067479e4964f0ab40ee637275" + ) + return path diff --git a/test/test_domain_tetrahedralization_recovery.py b/test/test_domain_tetrahedralization_recovery.py new file mode 100644 index 0000000..1c7a309 --- /dev/null +++ b/test/test_domain_tetrahedralization_recovery.py @@ -0,0 +1,364 @@ +import importlib +from types import SimpleNamespace + +import numpy as np +import pyvista as pv +import pytest + +from svv.domain.domain import Domain +from svv.domain.routines.tetrahedralize import _symmetric_surface_distance +from svv.tree.tree import Tree + + +domain_mod = importlib.import_module("svv.domain.domain") + + +def _domain_with_boundary(surface): + domain = Domain(surface) + domain.boundary = surface.copy(deep=True) + domain.set_random_seed(102) + domain.set_random_generator() + return domain + + +def test_domain_installs_recovered_surface_and_sampling_state( + closed_self_intersecting_surface, +): + source = closed_self_intersecting_surface + source_points = np.asarray(source.points).copy() + source_faces = np.asarray(source.faces).copy() + domain = _domain_with_boundary(source) + + mesh = domain.get_interior(repair_max_distance_ratio=0.4) + + assert domain.mesh_build_report.selected_strategy in {"original", "meshfix"} + if domain.mesh_build_report.attempts[0].status == "failed": + assert domain.mesh_build_report.selected_strategy == "meshfix" + assert [ + attempt.strategy for attempt in domain.mesh_build_report.attempts[:2] + ] == ["original", "meshfix"] + else: + assert domain.mesh_build_report.attempts[0].strategy == "original" + assert domain.mesh_build_report.attempts[0].status == "succeeded" + assert np.array_equal(domain.original_boundary.points, source_points) + assert np.array_equal(domain.original_boundary.faces, source_faces) + assert np.array_equal(source.points, source_points) + assert np.array_equal(source.faces, source_faces) + + assert domain.boundary.is_manifold + assert domain.boundary.n_open_edges == 0 + assert domain.boundary.n_points == domain.mesh_build_report.selected_surface.n_points + assert domain.boundary.n_cells == domain.mesh_build_report.selected_surface.n_cells + assert np.array_equal(domain.boundary_nodes, domain.boundary.points) + assert np.array_equal( + domain.boundary_vertices, + domain.boundary.faces.reshape(-1, 4)[:, 1:], + ) + boundary_weights = np.asarray(domain.boundary.cell_data["Normalized_Area"]) + assert np.isfinite(boundary_weights).all() + assert (boundary_weights >= 0).all() + assert boundary_weights.sum() == pytest.approx(1.0) + + assert mesh is domain.mesh + assert mesh.n_cells > 0 + assert set(np.unique(mesh.celltypes)).issubset( + {int(pv.CellType.TETRA), int(pv.CellType.QUADRATIC_TETRA)} + ) + assert np.isfinite(domain.mesh_nodes).all() + assert domain.mesh_vertices.min() >= 0 + assert domain.mesh_vertices.max() < domain.mesh_nodes.shape[0] + volume_weights = np.asarray(mesh.cell_data["Normalized_Volume"]) + assert np.isfinite(volume_weights).all() + assert (volume_weights >= 0).all() + assert volume_weights.sum() == pytest.approx(1.0) + assert domain.cumulative_probability[-1] == pytest.approx(1.0) + assert domain.mesh_tree is not None + assert domain.mesh_tree_2 is not None + + extracted = mesh.extract_surface().triangulate() + diagonal = np.linalg.norm( + np.asarray(domain.boundary.bounds)[1::2] + - np.asarray(domain.boundary.bounds)[::2] + ) + assert _symmetric_surface_distance(domain.boundary, extracted) <= 0.01 * diagonal + + boundary_point = domain.get_boundary_points(1) + assert boundary_point.shape == (1, 3) + assert np.isfinite(boundary_point).all() + + domain.evaluate_fast = lambda points, **kwargs: -0.5 * np.ones( + (np.asarray(points).shape[0], 1), + dtype=float, + ) + tree = Tree() + tree.set_domain(domain) + tree.set_root(max_attempts=5, attempts=20) + assert tree.data.shape[0] == 1 + assert np.isfinite(np.asarray(tree.data)[0, :6]).all() + + +def test_domain_clean_surface_keeps_original_strategy_and_geometry(): + source = pv.Cube().triangulate() + domain = _domain_with_boundary(source) + + domain.get_interior() + + assert domain.mesh_build_report.selected_strategy == "original" + assert domain.boundary.n_points == source.n_points + assert domain.boundary.n_cells == source.n_cells + assert _symmetric_surface_distance(domain.boundary, source) == pytest.approx(0.0) + + +def test_domain_failure_does_not_install_partial_volume_state(monkeypatch): + source = pv.Cube().triangulate() + domain = _domain_with_boundary(source) + previous_boundary = domain.boundary.copy(deep=True) + previous_mesh = object() + previous_tree = object() + previous_tree_2 = object() + previous_report = object() + previous_nodes = np.array([[9.0, 8.0, 7.0]]) + previous_vertices = np.array([[0, 0, 0, 0]]) + domain.mesh = previous_mesh + domain.mesh_tree = previous_tree + domain.mesh_tree_2 = previous_tree_2 + domain.mesh_build_report = previous_report + domain.mesh_nodes = previous_nodes + domain.mesh_vertices = previous_vertices + + invalid_result = SimpleNamespace( + grid=pv.UnstructuredGrid(), + nodes=np.empty((0, 3)), + elements=np.empty((0, 4), dtype=np.int64), + surface=pv.Sphere(), + report=object(), + ) + monkeypatch.setattr(domain_mod, "tetrahedralize", lambda *args, **kwargs: invalid_result) + + with pytest.raises(ValueError, match="empty|non-empty"): + domain.get_interior() + + assert domain.mesh is previous_mesh + assert domain.mesh_tree is previous_tree + assert domain.mesh_tree_2 is previous_tree_2 + assert domain.mesh_build_report is previous_report + assert domain.mesh_nodes is previous_nodes + assert domain.mesh_vertices is previous_vertices + assert np.array_equal(domain.boundary.points, previous_boundary.points) + assert np.array_equal(domain.boundary.faces, previous_boundary.faces) + + +def test_domain_rejects_worker_arrays_that_do_not_match_grid(monkeypatch): + points = np.array( + [[0.0, 0.0, 0.0], [1.0, 0.0, 0.0], [0.0, 1.0, 0.0], [0.0, 0.0, 1.0]] + ) + cells = np.array([4, 0, 1, 2, 3], dtype=np.int64) + grid = pv.UnstructuredGrid( + cells, + np.array([pv.CellType.TETRA], dtype=np.uint8), + points, + ) + surface = grid.extract_surface().triangulate() + domain = _domain_with_boundary(surface) + invalid_result = SimpleNamespace( + grid=grid, + nodes=np.vstack((points, [[2.0, 2.0, 2.0]])), + elements=np.array([[0, 1, 2, 3]], dtype=np.int64), + surface=surface, + report=object(), + ) + monkeypatch.setattr(domain_mod, "tetrahedralize", lambda *args, **kwargs: invalid_result) + + with pytest.raises(ValueError, match="node count"): + domain.get_interior() + + assert domain.mesh is None + assert domain.mesh_tree is None + assert domain.mesh_tree_2 is None + assert domain.mesh_build_report is None + + +def test_domain_rejects_quadratic_tetrahedra_from_first_order_worker(monkeypatch): + corner_points = np.array( + [[0.0, 0.0, 0.0], [1.0, 0.0, 0.0], [0.0, 1.0, 0.0], [0.0, 0.0, 1.0]] + ) + edge_points = np.array( + [ + [0.5, 0.0, 0.0], + [0.5, 0.5, 0.0], + [0.0, 0.5, 0.0], + [0.0, 0.0, 0.5], + [0.5, 0.0, 0.5], + [0.0, 0.5, 0.5], + ] + ) + points = np.vstack((corner_points, edge_points)) + elements = np.arange(10, dtype=np.int64).reshape(1, 10) + grid = pv.UnstructuredGrid( + np.concatenate(([10], elements[0])), + np.array([pv.CellType.QUADRATIC_TETRA], dtype=np.uint8), + points, + ) + surface = grid.extract_surface().triangulate() + domain = _domain_with_boundary(surface) + result = SimpleNamespace( + grid=grid, + nodes=points, + elements=elements, + surface=surface, + report=object(), + ) + monkeypatch.setattr(domain_mod, "tetrahedralize", lambda *args, **kwargs: result) + + with pytest.raises(ValueError, match="first-order.*M, 4"): + domain.get_interior() + + assert domain.mesh is None + assert domain.mesh_build_report is None + + +def test_domain_rejects_raw_tetgen_switches_before_worker(monkeypatch): + domain = _domain_with_boundary(pv.Cube().triangulate()) + + def unexpected_worker(*args, **kwargs): + raise AssertionError("Raw switches must not reach the Domain TetGen worker") + + monkeypatch.setattr(domain_mod, "tetrahedralize", unexpected_worker) + + with pytest.raises(ValueError, match="switches.*order=1.*nobisect"): + domain.get_interior(switches="pq1.2") + + +def test_boundary_rebuild_failure_cannot_leave_stale_volume_state(monkeypatch): + domain = _domain_with_boundary(pv.Cube().triangulate()) + stale = object() + domain.mesh = stale + domain.mesh_nodes = stale + domain.mesh_vertices = stale + domain.mesh_tree = stale + domain.mesh_tree_2 = stale + domain.all_mesh_cells = stale + domain.cumulative_probability = stale + domain.characteristic_length = 2.0 + domain.area = 3.0 + domain.volume = 4.0 + domain.convexity = 0.5 + domain.mesh_build_report = stale + domain.random_points = stale + + replacement = pv.Sphere(theta_resolution=8, phi_resolution=8).triangulate() + domain.original_boundary = replacement + replacement_grid = object() + monkeypatch.setattr( + domain_mod, + "contour", + lambda *args, **kwargs: (replacement.copy(deep=True), replacement_grid), + ) + + boundary, grid = domain.get_boundary(25) + + assert grid is replacement_grid + assert boundary.n_points == replacement.n_points + assert domain.mesh is None + assert domain.mesh_nodes is None + assert domain.mesh_vertices is None + assert domain.mesh_tree is None + assert domain.mesh_tree_2 is None + assert domain.all_mesh_cells is None + assert domain.cumulative_probability is None + assert domain.characteristic_length is None + assert domain.area is None + assert domain.volume is None + assert domain.convexity is None + assert domain.mesh_build_report is None + assert domain.random_points is None + + monkeypatch.setattr( + domain_mod, + "tetrahedralize", + lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError("worker failed")), + ) + with pytest.raises(RuntimeError, match="worker failed"): + domain.get_interior() + + assert domain.mesh is None + assert domain.mesh_build_report is None + assert domain.boundary.n_points == replacement.n_points + + +def test_cached_build_rechecks_mesh_after_boundary_replacement(monkeypatch): + domain = _domain_with_boundary(pv.Cube().triangulate()) + domain.patches = [] + domain.PTS = np.zeros((1, 1, 1, 1, 1, 3)) + domain.function_tree = object() + domain.random_generator = object() + domain.boundary = None + domain.mesh = object() + domain.mesh_tree = object() + domain.mesh_tree_2 = object() + calls = [] + replacement = pv.Sphere(theta_resolution=8, phi_resolution=8).triangulate() + + def rebuild_boundary(resolution): + domain._set_boundary_mesh(replacement) + domain.grid = object() + return domain.boundary, domain.grid + + def rebuild_interior(**kwargs): + calls.append("interior") + domain.mesh = object() + domain.mesh_tree = object() + domain.mesh_tree_2 = object() + return domain.mesh + + monkeypatch.setattr(domain, "get_boundary", rebuild_boundary) + monkeypatch.setattr(domain, "get_interior", rebuild_interior) + + domain.build(resolution=25) + + assert calls == ["interior"] + assert domain.mesh is not None + assert domain.mesh_tree is not None + + +def test_domain_rejects_zero_volume_selected_surface(monkeypatch): + points = np.array( + [[0.0, 0.0, 0.0], [1.0, 0.0, 0.0], [0.0, 1.0, 0.0], [0.0, 0.0, 1.0]] + ) + elements = np.array([[0, 1, 2, 3]], dtype=np.int64) + grid = pv.UnstructuredGrid( + np.array([4, 0, 1, 2, 3], dtype=np.int64), + np.array([pv.CellType.TETRA], dtype=np.uint8), + points, + ) + domain = _domain_with_boundary(pv.Cube().triangulate()) + result = SimpleNamespace( + grid=grid, + nodes=points, + elements=elements, + surface=pv.Plane().triangulate(), + report=object(), + ) + monkeypatch.setattr(domain_mod, "tetrahedralize", lambda *args, **kwargs: result) + monkeypatch.setattr(domain_mod, "validate_recovery_surface", lambda *args, **kwargs: None) + + with pytest.raises(ValueError, match="selected surface volume.*positive.*finite"): + domain.get_interior() + + assert domain.mesh is None + assert domain.mesh_build_report is None + + +def test_boundary_install_rejects_zero_measure_without_partial_state(): + domain = Domain(np.zeros((1, 3))) + previous = pv.Cube().triangulate() + domain.boundary = previous + degenerate = pv.PolyData( + np.array([[0.0, 0.0, 0.0], [1.0, 0.0, 0.0], [2.0, 0.0, 0.0]]), + np.array([3, 0, 1, 2]), + ) + + with pytest.raises(ValueError, match="positive finite"): + domain._set_boundary_mesh(degenerate) + + assert domain.boundary is previous diff --git a/test/test_gui_domain_build_feedback.py b/test/test_gui_domain_build_feedback.py new file mode 100644 index 0000000..678c398 --- /dev/null +++ b/test/test_gui_domain_build_feedback.py @@ -0,0 +1,434 @@ +import importlib +import json +from queue import Queue +import sys +import threading +from types import SimpleNamespace + +import pyvista as pv +import pytest +from PySide6.QtWidgets import QApplication, QMessageBox + +from svv.domain.routines.mesh_diagnostics import ( + TetGenAttemptReport, + TetrahedralizationError, + TetrahedralizationReport, + summarize_surface, + summarize_tetgen_output, +) +from svv.visualize.gui.domain_build_feedback import ( + apply_feedback_to_message_box, + build_domain_feedback, + report_for_telemetry, + sanitize_local_paths, +) + + +@pytest.fixture(scope="module") +def qt_app(): + return QApplication.instance() or QApplication([]) + + +def _attempt( + *, + strategy="original", + status="failed", + recoverable=True, + message="TetGen rejected the surface.", + diagnostics=None, +): + return TetGenAttemptReport( + strategy=strategy, + status=status, + surface=summarize_surface(pv.Cube().triangulate()), + duration_seconds=1.25, + recoverable=recoverable, + tetgen_args=(), + tetgen_kwargs={"order": 1, "nobisect": True}, + diagnostics=diagnostics, + message=message, + ) + + +def _report(attempts, selected_strategy=None): + surface = summarize_surface(pv.Cube().triangulate()) + return TetrahedralizationReport( + source=surface, + attempts=list(attempts), + selected_strategy=selected_strategy, + selected_surface=surface if selected_strategy else None, + versions={"tetgen": "0.6.4", "pyvista": "0.46.5"}, + ) + + +def test_recovered_success_feedback_names_automatic_strategy(): + report = _report( + [ + _attempt(), + _attempt(strategy="meshfix", status="succeeded", recoverable=False), + ], + selected_strategy="meshfix", + ) + + feedback = build_domain_feedback(report=report, success=True) + + assert feedback.recovered is True + assert "successfully after surface repair" in feedback.status.lower() + assert "meshfix" in feedback.status.lower() + assert "automatic surface recovery" in feedback.informative_text.lower() + + +def test_intersection_failure_names_cause_and_action(): + diagnostics = summarize_tetgen_output( + "Warning: Two facets exactly intersect.\n" + " 1st facet triangle: [1,2,3] tag(-1).\n" + " 2nd facet triangle: [4,5,6] tag(-1).\n", + "free(): invalid next size (normal)\n", + -6, + ) + report = _report([_attempt(diagnostics=diagnostics)]) + + feedback = build_domain_feedback( + exception=TetrahedralizationError(report), + report=report, + success=False, + ) + + assert "intersecting surface facets" in feedback.informative_text.lower() + assert "repair the source surface" in feedback.informative_text.lower() + assert "facet triangle: [1,2,3]" in feedback.detailed_text + + +def test_open_nonmanifold_fallback_names_rejected_geometry(): + report = _report( + [ + _attempt(), + _attempt( + strategy="pyacvd", + status="rejected", + message="Recovery surface is non-manifold and has 16 open edges.", + diagnostics=None, + ), + ] + ) + + feedback = build_domain_feedback(report=report, success=False) + + assert "open or non-manifold" in feedback.informative_text.lower() + assert "repair the source surface" in feedback.informative_text.lower() + + +def test_infrastructure_failure_does_not_blame_source_geometry(): + diagnostics = summarize_tetgen_output( + "", + "ModuleNotFoundError: No module named 'tetgen'\n", + 1, + ) + report = _report( + [ + _attempt( + status="infrastructure-error", + recoverable=False, + message="TetGen worker dependency failed.", + diagnostics=diagnostics, + ) + ] + ) + + feedback = build_domain_feedback(report=report, success=False) + + assert "worker or its environment failed" in feedback.informative_text.lower() + assert "verify the installation" in feedback.informative_text.lower() + assert "intersecting" not in feedback.informative_text.lower() + + +def test_details_are_bounded_path_sanitized_and_deduplicate_tracebacks(): + traceback_text = ( + "Traceback (most recent call last):\n" + " File \"/home/person/private/project/worker.py\", line 9, in \n" + "RuntimeError: Failed to tetrahedralize\n" + ) + diagnostics = summarize_tetgen_output("x" * 40000, traceback_text, 1) + report = _report( + [ + _attempt(strategy="original", diagnostics=diagnostics), + _attempt(strategy="meshfix", diagnostics=diagnostics), + ] + ) + + feedback = build_domain_feedback(report=report, success=False) + + assert "Attempt 1: original" in feedback.detailed_text + assert "Attempt 2: meshfix" in feedback.detailed_text + assert len(feedback.detailed_text) < 70000 + assert feedback.detailed_text.count("Traceback (most recent call last)") == 1 + assert "/home/person/private" not in feedback.detailed_text + assert " " in feedback.detailed_text + assert "array([" not in feedback.detailed_text + + +def test_message_box_receives_short_informative_and_detailed_fields(qt_app): + report = _report([_attempt()]) + feedback = build_domain_feedback(report=report, success=False) + box = QMessageBox() + + apply_feedback_to_message_box(box, feedback) + + # Headless macOS Qt backends may not retain an unshown native window title. + # The portable message-box test below verifies the title setter and ordering. + assert box.text() == "The domain loaded, but its interior mesh could not be built." + assert box.informativeText() == feedback.informative_text + assert box.detailedText() == feedback.detailed_text + assert box.standardButtons() == QMessageBox.Ok + + +def test_message_box_title_survives_native_icon_initialization(): + class NativeResettingMessageBox: + Warning = "warning" + Ok = "ok" + + def __init__(self): + self.title = "" + + def setWindowTitle(self, title): + self.title = title + + def setIcon(self, icon): + assert icon == self.Warning + self.title = "" + + def setText(self, text): + self.text = text + + def setInformativeText(self, text): + self.informative_text = text + + def setDetailedText(self, text): + self.detailed_text = text + + def setStandardButtons(self, buttons): + self.standard_buttons = buttons + + report = _report([_attempt()]) + feedback = build_domain_feedback(report=report, success=False) + box = NativeResettingMessageBox() + + apply_feedback_to_message_box(box, feedback) + + assert box.title == "Domain Build Warning" + + +def test_telemetry_report_is_json_safe_and_removes_local_paths(): + diagnostics = summarize_tetgen_output( + "", + "Failure while reading /home/person/private/surface.stl\n", + 1, + ) + report = _report([_attempt(diagnostics=diagnostics)]) + + payload = report_for_telemetry(report) + + json.dumps(payload) + serialized = json.dumps(payload) + assert "/home/person/private" not in serialized + assert " " in serialized + assert "points" not in payload + assert "faces" not in payload + + +@pytest.mark.parametrize( + "local_path", + [ + "/private.stl", + r"C:\private.stl", + "C:/private.stl", + r"\\server\share\private.stl", + r"\\?\C:\private.stl", + r"\\?\UNC\server\share\private.stl", + ], +) +def test_path_sanitizer_removes_root_drive_and_unc_paths(local_path): + sanitized = sanitize_local_paths("Failure while reading {}".format(local_path)) + + assert sanitized == "Failure while reading " + + +@pytest.mark.parametrize( + "local_path", + [ + "/home/person/My Files/private.stl", + r"C:\Users\Person Name\private.stl", + r"\\server\share\Private Files\private.stl", + ], +) +def test_path_sanitizer_removes_quoted_paths_containing_spaces(local_path): + sanitized = sanitize_local_paths('Failure while reading "{}"'.format(local_path)) + + assert sanitized == 'Failure while reading " "' + + +def test_structured_telemetry_honors_disabled_gate(monkeypatch): + main_window_mod = importlib.import_module("svv.visualize.gui.main_window") + sentry_calls = [] + + class FakeScope: + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + def set_tag(self, *args): + sentry_calls.append(("tag", args)) + + def set_extra(self, *args): + sentry_calls.append(("extra", args)) + + def set_context(self, *args): + sentry_calls.append(("context", args)) + + fake_sentry = SimpleNamespace( + push_scope=lambda: FakeScope(), + capture_exception=lambda *args, **kwargs: sentry_calls.append( + ("exception", args, kwargs) + ), + capture_message=lambda *args, **kwargs: sentry_calls.append( + ("message", args, kwargs) + ), + flush=lambda *args, **kwargs: sentry_calls.append(("flush", args, kwargs)), + ) + monkeypatch.setitem(sys.modules, "sentry_sdk", fake_sentry) + monkeypatch.setattr(main_window_mod, "telemetry_enabled", lambda: False, raising=False) + monkeypatch.setattr( + main_window_mod, + "capture_exception", + lambda *args, **kwargs: sentry_calls.append(("wrapped-exception", args, kwargs)), + ) + monkeypatch.setattr( + main_window_mod, + "capture_message", + lambda *args, **kwargs: sentry_calls.append(("wrapped-message", args, kwargs)), + ) + + main_window_mod.VascularizeGUI._record_telemetry( + object(), + RuntimeError("disabled"), + telemetry_context={"tetrahedralization": {"selected_strategy": "meshfix"}}, + ) + main_window_mod.VascularizeGUI._record_telemetry( + object(), + message="disabled", + telemetry_context={"tetrahedralization": {"selected_strategy": "meshfix"}}, + ) + + assert sentry_calls == [] + + +class _LoaderHarness: + def __init__(self): + self.telemetry = [] + + def _record_telemetry(self, *args, **kwargs): + self.telemetry.append((args, kwargs)) + + +def _progress_labels(progress_queue): + labels = [] + while not progress_queue.empty(): + item = progress_queue.get_nowait() + if isinstance(item, dict) and item.get("label"): + labels.append(item["label"]) + return labels + + +def test_mesh_background_load_preserves_success_report(monkeypatch): + main_window_mod = importlib.import_module("svv.visualize.gui.main_window") + domain_module = importlib.import_module("svv.domain.domain") + report = _report( + [_attempt(strategy="meshfix", status="succeeded", recoverable=False)], + selected_strategy="meshfix", + ) + + class FakeDomain: + def __init__(self, mesh): + self.mesh = None + self.mesh_tree = None + self.boundary = None + self.patches = [] + self.mesh_build_report = None + + def create(self, progress_callback=None): + return None + + def solve(self, progress_callback=None): + return None + + def build(self, resolution=25, progress_callback=None): + self.mesh_build_report = report + self.boundary = object() + + monkeypatch.setattr(domain_module, "Domain", FakeDomain) + monkeypatch.setattr(pv, "read", lambda path: object()) + harness = _LoaderHarness() + progress = Queue() + + result = main_window_mod.VascularizeGUI._load_domain_file( + harness, + "/tmp/surface.stl", + threading.Event(), + progress, + 25, + ) + + assert result._build_failed is False + assert result._build_error is None + assert result._build_exception is None + assert result._build_report is report + assert any("successfully after surface repair" in label.lower() for label in _progress_labels(progress)) + assert harness.telemetry[0][1]["telemetry_context"]["tetrahedralization"]["selected_strategy"] == "meshfix" + + +def test_dmn_background_load_preserves_structured_failure(monkeypatch): + main_window_mod = importlib.import_module("svv.visualize.gui.main_window") + domain_module = importlib.import_module("svv.domain.domain") + diagnostics = summarize_tetgen_output( + "Warning: A segment and a facet intersect.\n", + "free(): invalid next size (normal)\n", + -6, + ) + report = _report([_attempt(diagnostics=diagnostics)]) + build_exception = TetrahedralizationError(report) + + class FakeDomain: + def __init__(self): + self.mesh = None + self.mesh_tree = None + self.boundary = None + self.patches = [] + self.mesh_build_report = None + + @classmethod + def load(cls, path): + return cls() + + def build(self, resolution=25, progress_callback=None): + raise build_exception + + monkeypatch.setattr(domain_module, "Domain", FakeDomain) + harness = _LoaderHarness() + + result = main_window_mod.VascularizeGUI._load_domain_file( + harness, + "/tmp/domain.dmn", + threading.Event(), + Queue(), + 25, + ) + + assert result._build_failed is True + assert "intersecting surface facets" in result._build_error.lower() + assert result._build_exception is build_exception + assert result._build_report is report + telemetry = harness.telemetry[0][1] + assert telemetry["action"] == "load_domain_build" + assert telemetry["telemetry_context"]["tetrahedralization"]["attempts"][0]["strategy"] == "original" diff --git a/test/test_issue_102_attachment.py b/test/test_issue_102_attachment.py new file mode 100644 index 0000000..e4761fa --- /dev/null +++ b/test/test_issue_102_attachment.py @@ -0,0 +1,56 @@ +import numpy as np +import pyvista as pv +import pytest + +from svv.domain.domain import Domain +from svv.domain.routines.tetrahedralize import _symmetric_surface_distance + + +def test_issue_102_attachment_completes_domain_build(issue_102_stl_path): + surface = pv.read(issue_102_stl_path) + source_points = np.asarray(surface.points).copy() + source_faces = np.asarray(surface.faces).copy() + source_diagonal = np.linalg.norm( + np.asarray(surface.bounds)[1::2] - np.asarray(surface.bounds)[::2] + ) + + domain = Domain(surface) + domain.create() + domain.solve() + domain.build(resolution=25) + + assert domain.mesh is not None + assert domain.mesh.n_cells > 0 + assert domain.mesh_tree is not None + assert domain.mesh_tree_2 is not None + assert domain.mesh_build_report.selected_strategy == "meshfix" + assert np.isfinite(domain.mesh_nodes).all() + assert domain.mesh_vertices.min() >= 0 + assert domain.mesh_vertices.max() < domain.mesh_nodes.shape[0] + + probabilities = np.asarray(domain.mesh.cell_data["Normalized_Volume"]) + assert np.isfinite(probabilities).all() + assert (probabilities >= 0).all() + assert probabilities.sum() == pytest.approx(1.0) + assert domain.cumulative_probability[-1] == pytest.approx(1.0) + + cell_volumes = np.asarray(domain.mesh.cell_data["Volume"]) + total_volume = float(cell_volumes.sum()) + assert np.isfinite(total_volume) + assert total_volume > 0 + selected_volume = abs(float(domain.boundary.volume)) + assert abs(total_volume - selected_volume) / selected_volume <= 0.005 + + assert domain.boundary.is_manifold + assert domain.boundary.n_open_edges == 0 + extracted = domain.mesh.extract_surface().triangulate() + assert _symmetric_surface_distance(domain.boundary, extracted) <= 0.01 * source_diagonal + + assert np.array_equal(domain.original_boundary.points, source_points) + assert np.array_equal(domain.original_boundary.faces, source_faces) + assert np.array_equal(surface.points, source_points) + assert np.array_equal(surface.faces, source_faces) + + boundary_point = domain.get_boundary_points(1) + assert boundary_point.shape == (1, 3) + assert np.isfinite(boundary_point).all() diff --git a/test/test_mesh_diagnostics.py b/test/test_mesh_diagnostics.py new file mode 100644 index 0000000..197f617 --- /dev/null +++ b/test/test_mesh_diagnostics.py @@ -0,0 +1,139 @@ +import math +import json + +import pyvista as pv +import pytest + +import svv.domain.routines.mesh_diagnostics as mesh_diagnostics +from svv.domain.routines.mesh_diagnostics import ( + TetGenAttemptReport, + TetrahedralizationError, + TetrahedralizationReport, + summarize_surface, + summarize_tetgen_output, +) + + +def test_summarize_surface_reports_closed_cube_geometry(): + summary = summarize_surface(pv.Cube().triangulate()) + + assert summary.n_points == 8 + assert summary.n_triangles == 12 + assert summary.n_components == 1 + assert summary.is_all_triangles is True + assert summary.points_finite is True + assert summary.is_manifold is True + assert summary.n_open_edges == 0 + assert summary.bounds == (-0.5, 0.5, -0.5, 0.5, -0.5, 0.5) + assert summary.diagonal == math.sqrt(3.0) + assert summary.area == 6.0 + assert summary.volume == pytest.approx(1.0) + + +def test_summarize_tetgen_output_identifies_intersections_and_native_abort(): + stdout = """\ +Warning: A segment and a facet intersect. + segment: [38341,38340] tag(-1). + facet triangle: [1033,1035,1238] tag(-1) +Warning: A segment and a facet intersect. + segment: [38311,38310] tag(-1). + facet triangle: [37032,74417,74416] tag(-1) +Warning: Two facets exactly intersect. + 1st facet triangle: [1608,1810,75715] tag(-1). + 2nd facet triangle: [1605,1608,75721] tag(-1). + 151404 (12) subfaces are recovered (missing). +""" + + summary = summarize_tetgen_output( + stdout, + "free(): invalid next size (normal)\n", + -6, + ) + + assert summary.segment_facet_intersections == 2 + assert summary.facet_facet_intersections == 1 + assert summary.missing_subfaces == 12 + assert summary.native_abort is True + assert summary.return_code == -6 + assert summary.signal_name == "SIGABRT" + assert any("segment: [38341,38340]" in line for line in summary.examples) + + +def test_sigabrt_name_is_stable_when_host_signal_enum_uses_other_numbers( + monkeypatch, +): + def unsupported_signal_number(number): + raise ValueError(number) + + monkeypatch.setattr(mesh_diagnostics.signal, "Signals", unsupported_signal_number) + + summary = summarize_tetgen_output("", "", -6) + + assert summary.signal_name == "SIGABRT" + + +def test_failure_report_provides_actionable_summary_details_and_json(): + surface = summarize_surface(pv.Cube().triangulate()) + diagnostic = summarize_tetgen_output( + "Warning: A segment and a facet intersect.\n" + " segment: [4,5] tag(-1).\n" + " facet triangle: [1,2,3] tag(-1).\n", + "free(): invalid next size (normal)\n", + -6, + ) + attempt = TetGenAttemptReport( + strategy="original", + status="failed", + surface=surface, + duration_seconds=1.25, + recoverable=True, + tetgen_args=(), + tetgen_kwargs={"order": 1, "nobisect": True}, + diagnostics=diagnostic, + message="TetGen rejected the input surface.", + ) + report = TetrahedralizationReport( + source=surface, + attempts=[attempt], + selected_strategy=None, + selected_surface=None, + versions={"tetgen": "0.6.4"}, + ) + + summary = report.user_summary() + details = report.detailed_text() + payload = report.to_dict() + + assert "intersecting surface facets" in summary + assert "repair the source surface" in summary.lower() + assert "original" in details + assert "SIGABRT" in details + assert "nobisect=True" in details + assert "8 points" in details + assert "segment: [4,5]" in details + assert payload["source"]["n_points"] == 8 + assert payload["attempts"][0]["diagnostics"]["return_code"] == -6 + json.dumps(payload) + assert str(TetrahedralizationError(report)) == summary + + +def test_tetgen_output_is_bounded_without_losing_python_failure_classification(): + summary = summarize_tetgen_output( + "x" * 40000, + "Traceback (most recent call last):\nRuntimeError: Unknown exception\n", + 1, + ) + + assert summary.python_exception is True + assert summary.native_abort is False + assert summary.signal_name is None + assert len(summary.stdout) < 33000 + assert "[truncated" in summary.stdout + + +def test_tetgen_output_identifies_windows_native_crash_status(): + summary = summarize_tetgen_output("", "", 0xC0000005) + + assert summary.native_abort is True + assert summary.return_code == 0xC0000005 + assert summary.signal_name == "NTSTATUS_0xC0000005" diff --git a/test/test_tetgen_worker.py b/test/test_tetgen_worker.py new file mode 100644 index 0000000..8cbe0af --- /dev/null +++ b/test/test_tetgen_worker.py @@ -0,0 +1,464 @@ +import importlib +import json +import os +from pathlib import Path +import signal +import subprocess +import sys + +import numpy as np +import pyvista as pv +import pytest + +from svv.domain.routines.mesh_diagnostics import TetGenWorkerError + + +tetrahedralize_mod = importlib.import_module("svv.domain.routines.tetrahedralize") +tetgen_worker_mod = importlib.import_module("svv.domain.routines.tetgen_worker") + + +def _write_array_worker(tmp_path, nodes_expression, elements_expression): + worker = tmp_path / "array_worker.py" + worker.write_text( + "import sys\n" + "import numpy as np\n" + "nodes = {}\n".format(nodes_expression) + + "elements = {}\n".format(elements_expression) + + "np.savez(sys.argv[2], nodes=nodes, elems=elements)\n" + ) + return worker + + +def test_worker_confines_tetgen_artifacts_to_its_temporary_directory(tmp_path, monkeypatch): + caller_dir = tmp_path / "caller" + caller_dir.mkdir() + monkeypatch.chdir(caller_dir) + worker = tmp_path / "artifact_worker.py" + worker.write_text( + """\ +import pathlib +import sys +import numpy as np + +pathlib.Path('_skipped.node').write_text('diagnostic') +pathlib.Path('_skipped.face').write_text('diagnostic') +nodes = np.array([[0., 0., 0.], [1., 0., 0.], [0., 1., 0.], [0., 0., 1.]]) +elements = np.array([[0, 1, 2, 3]], dtype=np.int64) +np.savez(sys.argv[2], nodes=nodes, elems=elements) +""" + ) + + nodes, elements = tetrahedralize_mod._tetgen_worker_tetrahedralize( + pv.Tetrahedron().extract_surface(), + (), + {"verbose": 0}, + str(worker), + sys.executable, + ) + leaked = list(caller_dir.glob("_skipped.*")) + for path in leaked: + path.unlink() + + assert nodes.shape == (4, 3) + assert elements.shape == (1, 4) + assert leaked == [] + + +def test_worker_drains_large_output_without_deadlocking(tmp_path): + worker = tmp_path / "verbose_worker.py" + worker.write_text( + """\ +import sys +import numpy as np + +print('x' * 200000, flush=True) +nodes = np.array([[0., 0., 0.], [1., 0., 0.], [0., 1., 0.], [0., 0., 1.]]) +elements = np.array([[0, 1, 2, 3]], dtype=np.int64) +np.savez(sys.argv[2], nodes=nodes, elems=elements) +""" + ) + driver = tmp_path / "driver.py" + driver.write_text( + """\ +import importlib +import sys +import pyvista as pv + +module = importlib.import_module('svv.domain.routines.tetrahedralize') +module._tetgen_worker_tetrahedralize( + pv.Tetrahedron().extract_surface(), (), {'verbose': 0}, sys.argv[1], sys.executable +) +""" + ) + env = os.environ.copy() + env["PYTHONPATH"] = str(Path(tetrahedralize_mod.__file__).resolve().parents[3]) + + completed = subprocess.run( + [sys.executable, str(driver), str(worker)], + cwd=tmp_path, + env=env, + capture_output=True, + text=True, + timeout=20, + check=False, + ) + + assert completed.returncode == 0, completed.stderr + + +def test_worker_raises_typed_recoverable_error_for_geometry_rejection(tmp_path): + worker = tmp_path / "geometry_failure_worker.py" + worker.write_text( + """\ +import sys + +print('Warning: A segment and a facet intersect.') +print(' segment: [4,5] tag(-1).') +print(' facet triangle: [1,2,3] tag(-1).') +raise SystemExit(3) +""" + ) + + with pytest.raises(TetGenWorkerError) as error_info: + tetrahedralize_mod._tetgen_worker_tetrahedralize( + pv.Tetrahedron().extract_surface(), + (), + {"order": 1, "nobisect": True, "verbose": 0}, + str(worker), + sys.executable, + strategy="original", + ) + + error = error_info.value + assert error.recoverable is True + assert error.attempt.strategy == "original" + assert error.attempt.status == "failed" + assert error.attempt.diagnostics.return_code == 3 + assert error.attempt.diagnostics.segment_facet_intersections == 1 + + +def test_worker_treats_internal_tetgen_error_as_geometry_rejection(tmp_path): + worker = tmp_path / "internal_tetgen_failure_worker.py" + worker.write_text( + "raise RuntimeError('Internal TetGen error within `recoversubfaces`.')\n" + ) + + with pytest.raises(TetGenWorkerError) as error_info: + tetrahedralize_mod._tetgen_worker_tetrahedralize( + pv.Tetrahedron().extract_surface(), + (), + {"order": 1, "nobisect": True, "verbose": 0}, + str(worker), + sys.executable, + strategy="original", + ) + + error = error_info.value + assert error.recoverable is True + assert error.attempt.status == "failed" + assert error.attempt.diagnostics.python_exception is True + assert "Internal TetGen error" in error.attempt.diagnostics.stderr + + +def test_worker_treats_tetgen_self_intersection_error_as_geometry_rejection( + tmp_path, +): + worker = tmp_path / "self_intersection_failure_worker.py" + worker.write_text( + "raise RuntimeError('The input surface mesh contain self-intersections.')\n" + ) + + with pytest.raises(TetGenWorkerError) as error_info: + tetrahedralize_mod._tetgen_worker_tetrahedralize( + pv.Tetrahedron().extract_surface(), + (), + {"order": 1, "nobisect": True, "verbose": 0}, + str(worker), + sys.executable, + strategy="original", + ) + + error = error_info.value + assert error.recoverable is True + assert error.attempt.status == "failed" + assert "self-intersections" in error.attempt.diagnostics.stderr + + +def test_worker_classifies_missing_dependency_as_nonrecoverable(tmp_path): + worker = tmp_path / "dependency_failure_worker.py" + worker.write_text("import missing_issue_102_dependency\n") + + with pytest.raises(TetGenWorkerError) as error_info: + tetrahedralize_mod._tetgen_worker_tetrahedralize( + pv.Tetrahedron().extract_surface(), + (), + {"verbose": 0}, + str(worker), + sys.executable, + strategy="original", + ) + + error = error_info.value + assert error.recoverable is False + assert "infrastructure" in str(error).lower() + assert error.attempt.diagnostics.python_exception is True + assert "ModuleNotFoundError" in error.attempt.diagnostics.stderr + + +def test_self_intersection_in_worker_path_does_not_hide_dependency_failure(tmp_path): + worker = tmp_path / "self-intersection-dependency-worker.py" + worker.write_text("import missing_issue_102_dependency\n") + + with pytest.raises(TetGenWorkerError) as error_info: + tetrahedralize_mod._tetgen_worker_tetrahedralize( + pv.Tetrahedron().extract_surface(), + (), + {"verbose": 0}, + str(worker), + sys.executable, + strategy="original", + ) + + error = error_info.value + assert error.recoverable is False + assert "infrastructure" in str(error).lower() + assert error.attempt.diagnostics.python_exception is True + assert "ModuleNotFoundError" in error.attempt.diagnostics.stderr + + +def test_worker_rejects_nonfinite_result_arrays_as_infrastructure_failure(tmp_path): + worker = tmp_path / "invalid_result_worker.py" + worker.write_text( + """\ +import sys +import numpy as np + +nodes = np.array([[np.nan, 0., 0.], [1., 0., 0.], [0., 1., 0.], [0., 0., 1.]]) +elements = np.array([[0, 1, 2, 3]], dtype=np.int64) +np.savez(sys.argv[2], nodes=nodes, elems=elements) +""" + ) + + with pytest.raises(TetGenWorkerError) as error_info: + tetrahedralize_mod._tetgen_worker_tetrahedralize( + pv.Tetrahedron().extract_surface(), + (), + {"verbose": 0}, + str(worker), + sys.executable, + ) + + error = error_info.value + assert error.recoverable is False + assert error.attempt.status == "invalid-output" + assert "finite" in str(error).lower() + + +def test_worker_normalizes_one_based_connectivity_once(tmp_path): + worker = _write_array_worker( + tmp_path, + "np.array([[0., 0., 0.], [1., 0., 0.], [0., 1., 0.], [0., 0., 1.]])", + "np.array([[1, 2, 3, 4]], dtype=np.int64)", + ) + + _, elements = tetrahedralize_mod._tetgen_worker_tetrahedralize( + pv.Tetrahedron().extract_surface(), + (), + {"verbose": 0}, + str(worker), + sys.executable, + ) + + assert np.array_equal(elements, [[0, 1, 2, 3]]) + + +@pytest.mark.parametrize( + ("nodes_expression", "elements_expression", "message"), + [ + ( + "np.empty((0, 3))", + "np.array([[0, 1, 2, 3]], dtype=np.int64)", + "nodes", + ), + ( + "np.zeros((4, 2))", + "np.array([[0, 1, 2, 3]], dtype=np.int64)", + "nodes", + ), + ( + "np.zeros((4, 3))", + "np.empty((0, 4), dtype=np.int64)", + "elements", + ), + ( + "np.zeros((4, 3))", + "np.array([[0, 1, 2]], dtype=np.int64)", + "elements", + ), + ( + "np.zeros((4, 3))", + "np.array([[0., 1., 2., 3.]])", + "integer", + ), + ( + "np.zeros((4, 3))", + "np.array([[0, 1, 2, 4]], dtype=np.int64)", + "range", + ), + ( + "np.zeros((4, 3))", + "np.array([[-1, 1, 2, 3]], dtype=np.int64)", + "range", + ), + ], +) +def test_worker_rejects_malformed_result_arrays( + tmp_path, + nodes_expression, + elements_expression, + message, +): + worker = _write_array_worker(tmp_path, nodes_expression, elements_expression) + + with pytest.raises(TetGenWorkerError) as error_info: + tetrahedralize_mod._tetgen_worker_tetrahedralize( + pv.Tetrahedron().extract_surface(), + (), + {"verbose": 0}, + str(worker), + sys.executable, + ) + + assert error_info.value.recoverable is False + assert error_info.value.attempt.status == "invalid-output" + assert message in str(error_info.value).lower() + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX signal return codes are unavailable") +def test_worker_reports_native_signal_as_recoverable(tmp_path): + worker = tmp_path / "signal_worker.py" + worker.write_text( + "import os\n" + "import signal\n" + "os.kill(os.getpid(), signal.SIGABRT)\n" + ) + + with pytest.raises(TetGenWorkerError) as error_info: + tetrahedralize_mod._tetgen_worker_tetrahedralize( + pv.Tetrahedron().extract_surface(), + (), + {"verbose": 0}, + str(worker), + sys.executable, + ) + + diagnostics = error_info.value.attempt.diagnostics + assert error_info.value.recoverable is True + assert diagnostics.return_code == -signal.SIGABRT + assert diagnostics.signal_name == "SIGABRT" + + +def test_worker_wraps_launch_failure_as_infrastructure_error(tmp_path): + missing_worker = tmp_path / "missing_worker.py" + + with pytest.raises(TetGenWorkerError) as error_info: + tetrahedralize_mod._tetgen_worker_tetrahedralize( + pv.Tetrahedron().extract_surface(), + (), + {"verbose": 0}, + str(missing_worker), + str(tmp_path / "missing-python"), + ) + + assert error_info.value.recoverable is False + assert error_info.value.attempt.status == "infrastructure-error" + + +def test_worker_wraps_unserializable_configuration_as_infrastructure_error(tmp_path): + worker = tmp_path / "unused_worker.py" + worker.write_text("pass\n") + + with pytest.raises(TetGenWorkerError) as error_info: + tetrahedralize_mod._tetgen_worker_tetrahedralize( + pv.Tetrahedron().extract_surface(), + (), + {"unsupported": object()}, + str(worker), + sys.executable, + ) + + assert error_info.value.recoverable is False + assert error_info.value.attempt.status == "infrastructure-error" + + +def test_worker_removes_temporary_directory_after_failure(tmp_path, monkeypatch): + temp_root = tmp_path / "worker-temp" + temp_root.mkdir() + monkeypatch.setattr(tetrahedralize_mod.tempfile, "tempdir", str(temp_root)) + worker = tmp_path / "failing_worker.py" + worker.write_text("raise SystemExit(7)\n") + + with pytest.raises(TetGenWorkerError): + tetrahedralize_mod._tetgen_worker_tetrahedralize( + pv.Tetrahedron().extract_surface(), + (), + {"verbose": 0}, + str(worker), + sys.executable, + ) + + assert list(temp_root.iterdir()) == [] + + +def test_worker_wraps_missing_result_archive_as_infrastructure_failure(tmp_path): + worker = tmp_path / "missing_result_worker.py" + worker.write_text("pass\n") + + with pytest.raises(TetGenWorkerError) as error_info: + tetrahedralize_mod._tetgen_worker_tetrahedralize( + pv.Tetrahedron().extract_surface(), + (), + {"verbose": 0}, + str(worker), + sys.executable, + ) + + error = error_info.value + assert error.recoverable is False + assert error.attempt.status == "invalid-output" + assert "invalid output" in str(error).lower() + + +@pytest.mark.parametrize("return_tuple", [True, False]) +def test_tetgen_worker_main_saves_supported_result_formats( + tmp_path, + monkeypatch, + return_tuple, +): + surface_path = tmp_path / "surface.vtp" + output_path = tmp_path / "tet.npz" + config_path = tmp_path / "config.json" + pv.Tetrahedron().extract_surface().save(surface_path) + config_path.write_text(json.dumps({"args": [], "kwargs": {"verbose": 0}})) + expected_nodes = np.array( + [[0.0, 0.0, 0.0], [1.0, 0.0, 0.0], [0.0, 1.0, 0.0], [0.0, 0.0, 1.0]] + ) + expected_elements = np.array([[0, 1, 2, 3]], dtype=np.int64) + + class TetGenResult: + def __init__(self, surface): + self.node = expected_nodes + self.elem = expected_elements + + def tetrahedralize(self, *args, **kwargs): + if return_tuple: + return self.node, self.elem + return pv.UnstructuredGrid() + + monkeypatch.setattr(tetgen_worker_mod.tetgen, "TetGen", TetGenResult) + + tetgen_worker_mod.main(str(surface_path), str(output_path), str(config_path)) + + with np.load(output_path) as result: + assert np.array_equal(result["nodes"], expected_nodes) + assert np.array_equal(result["elems"], expected_elements) diff --git a/test/test_tetrahedralize_recovery.py b/test/test_tetrahedralize_recovery.py new file mode 100644 index 0000000..c9fc99e --- /dev/null +++ b/test/test_tetrahedralize_recovery.py @@ -0,0 +1,675 @@ +import importlib +import os +import sys +from dataclasses import replace + +import numpy as np +import pyvista as pv +import pytest + +from svv.domain.routines.mesh_diagnostics import ( + TetGenAttemptReport, + TetGenWorkerError, + TetrahedralizationError, + summarize_surface, + summarize_tetgen_output, +) + + +tetrahedralize_mod = importlib.import_module("svv.domain.routines.tetrahedralize") + + +def _geometry_worker_error(surface, strategy, tet_kwargs): + return TetGenWorkerError( + TetGenAttemptReport( + strategy=strategy, + status="failed", + surface=summarize_surface(surface), + duration_seconds=0.01, + recoverable=True, + tetgen_args=(), + tetgen_kwargs=dict(tet_kwargs), + diagnostics=summarize_tetgen_output( + "Warning: A segment and a facet intersect.\n", "", 1 + ), + message="TetGen rejected the {} surface.".format(strategy), + ) + ) + + +def test_closed_manifold_self_intersection_recovers_with_meshfix( + closed_self_intersecting_surface, +): + result = tetrahedralize_mod.tetrahedralize( + closed_self_intersecting_surface, + order=1, + nobisect=True, + repair_max_distance_ratio=0.4, + remesh_on_failure=False, + return_result=True, + ) + + assert result.report.selected_strategy in {"original", "meshfix"} + if result.report.attempts[0].status == "failed": + assert result.report.selected_strategy == "meshfix" + else: + assert result.report.attempts[0].strategy == "original" + assert result.report.attempts[0].status == "succeeded" + assert result.grid.n_cells > 0 + assert np.isfinite(result.nodes).all() + assert result.elements.shape[1] == 4 + assert result.surface.is_manifold + assert result.surface.n_open_edges == 0 + assert set(np.unique(result.grid.celltypes)) == {pv.CellType.TETRA} + + +def test_clean_surface_keeps_original_fast_path_and_legacy_tuple(monkeypatch): + def unexpected_recovery(*args, **kwargs): + raise AssertionError("Recovery must not run after the original surface succeeds") + + monkeypatch.setattr( + tetrahedralize_mod, + "repair_surface_with_meshfix", + unexpected_recovery, + ) + monkeypatch.setattr( + tetrahedralize_mod, + "uniform_remesh_surface", + unexpected_recovery, + ) + surface = pv.Cube().triangulate() + + result = tetrahedralize_mod.tetrahedralize(surface, return_result=True) + legacy = tetrahedralize_mod.tetrahedralize(surface) + + assert result.report.selected_strategy == "original" + assert len(result.report.attempts) == 1 + assert result.report.attempts[0].status == "succeeded" + assert isinstance(legacy, tuple) + assert len(legacy) == 3 + assert legacy[0].n_cells > 0 + + +def test_recovery_does_not_mutate_or_share_storage_with_input( + closed_self_intersecting_surface, +): + original_points = closed_self_intersecting_surface.points.copy() + original_faces = closed_self_intersecting_surface.faces.copy() + + result = tetrahedralize_mod.tetrahedralize( + closed_self_intersecting_surface, + repair_max_distance_ratio=0.4, + remesh_on_failure=False, + return_result=True, + ) + result.surface.points[0] += 1.0 + + assert np.array_equal(closed_self_intersecting_surface.points, original_points) + assert np.array_equal(closed_self_intersecting_surface.faces, original_faces) + assert not np.shares_memory(result.surface.points, closed_self_intersecting_surface.points) + + +def test_default_repair_bound_rejects_large_surface_change( + closed_self_intersecting_surface, +): + with pytest.raises(ValueError, match="bounds|displacement"): + tetrahedralize_mod.repair_surface_with_meshfix( + closed_self_intersecting_surface, + max_distance_ratio=0.01, + ) + + +def test_recovery_validation_rejects_internal_displacement_with_unchanged_bounds(): + source = pv.Sphere(theta_resolution=20, phi_resolution=20) + candidate = source.copy(deep=True) + interior_index = int(np.argmin(np.abs(candidate.points[:, 2]))) + candidate.points[interior_index] *= 0.5 + + with pytest.raises(ValueError, match="displacement"): + tetrahedralize_mod.validate_recovery_surface( + source, + candidate, + max_distance_ratio=0.01, + ) + + +def test_recovery_validation_rejects_nonfinite_surface_distance(monkeypatch): + source = pv.Cube().triangulate() + candidate = source.copy(deep=True) + monkeypatch.setattr( + tetrahedralize_mod, + "_symmetric_surface_distance", + lambda *args: np.nan, + ) + + with pytest.raises(ValueError, match="distance.*finite"): + tetrahedralize_mod.validate_recovery_surface( + source, + candidate, + max_distance_ratio=0.01, + ) + + +@pytest.mark.parametrize("distances", [np.array([]), np.array([np.nan]), np.array([np.inf])]) +def test_symmetric_surface_distance_rejects_invalid_arrays(distances): + class FakeSurface: + def compute_implicit_distance(self, other): + return {"implicit_distance": distances} + + with pytest.raises(ValueError, match="distance arrays.*finite.*non-empty"): + tetrahedralize_mod._symmetric_surface_distance( + FakeSurface(), + FakeSurface(), + ) + + +def test_recovery_validation_rejects_nonfinite_bounds(monkeypatch): + source = pv.Cube().triangulate() + candidate = source.copy(deep=True) + real_summary = tetrahedralize_mod.summarize_surface + + def summary(surface): + result = real_summary(surface) + if surface is candidate: + return replace(result, bounds=(np.nan,) + result.bounds[1:]) + return result + + monkeypatch.setattr(tetrahedralize_mod, "summarize_surface", summary) + + with pytest.raises(ValueError, match="bounds.*finite"): + tetrahedralize_mod.validate_recovery_surface( + source, + candidate, + max_distance_ratio=0.01, + ) + + +def test_recovery_validation_rejects_nonfinite_allowed_distance(): + source = pv.Cube().triangulate() + + with pytest.raises(ValueError, match="allowed recovery distance.*finite"): + tetrahedralize_mod.validate_recovery_surface( + source, + source.copy(deep=True), + max_distance_ratio=np.finfo(float).max, + ) + + +def test_invalid_repair_candidate_is_rejected_before_tetgen(monkeypatch): + source = pv.Cube().triangulate() + invalid_candidate = pv.Plane().triangulate() + calls = [] + + def worker(surface, tet_args, tet_kwargs, worker_script, python_exe, *, strategy): + calls.append(strategy) + if strategy != "original": + raise AssertionError("Invalid recovery candidate reached TetGen") + raise TetGenWorkerError( + TetGenAttemptReport( + strategy="original", + status="failed", + surface=summarize_surface(surface), + duration_seconds=0.01, + recoverable=True, + tetgen_args=(), + tetgen_kwargs=dict(tet_kwargs), + diagnostics=summarize_tetgen_output( + "Warning: A segment and a facet intersect.\n", "", 1 + ), + message="TetGen rejected the original surface.", + ) + ) + + monkeypatch.setattr(tetrahedralize_mod, "_tetgen_worker_tetrahedralize", worker) + monkeypatch.setattr( + tetrahedralize_mod, + "repair_surface_with_meshfix", + lambda *args, **kwargs: invalid_candidate.copy(deep=True), + ) + + with pytest.raises(TetrahedralizationError) as error_info: + tetrahedralize_mod.tetrahedralize( + source, + remesh_on_failure=False, + return_result=True, + ) + + assert calls == ["original"] + assert [attempt.status for attempt in error_info.value.report.attempts] == [ + "failed", + "rejected", + ] + rejected = error_info.value.report.attempts[1] + assert rejected.surface.n_points == invalid_candidate.n_points + assert rejected.surface.n_cells == invalid_candidate.n_cells + + +def test_meshfix_rejection_report_describes_the_repaired_candidate(monkeypatch): + source = pv.Cube().triangulate() + + class DistortingMeshFix: + def __init__(self, points, faces): + self.v = np.asarray(points).copy() + self.f = np.asarray(faces).copy() + + def repair(self, **kwargs): + self.v[:, 0] += 10.0 + + def worker(surface, tet_args, tet_kwargs, worker_script, python_exe, *, strategy): + raise _geometry_worker_error(surface, strategy, tet_kwargs) + + monkeypatch.setattr(tetrahedralize_mod.pymeshfix, "MeshFix", DistortingMeshFix) + monkeypatch.setattr(tetrahedralize_mod, "_tetgen_worker_tetrahedralize", worker) + + with pytest.raises(TetrahedralizationError) as error_info: + tetrahedralize_mod.tetrahedralize(source, remesh_on_failure=False) + + rejected = error_info.value.report.attempts[1] + assert rejected.status == "rejected" + assert rejected.surface.bounds[0] > 9.0 + + +def test_unexpected_meshfix_error_is_not_relabelled_as_geometry_rejection(monkeypatch): + source = pv.Cube().triangulate() + + def worker(surface, tet_args, tet_kwargs, worker_script, python_exe, *, strategy): + raise _geometry_worker_error(surface, strategy, tet_kwargs) + + def broken_repair(*args, **kwargs): + raise ValueError("meshfix implementation failed") + + monkeypatch.setattr(tetrahedralize_mod, "_tetgen_worker_tetrahedralize", worker) + monkeypatch.setattr(tetrahedralize_mod, "repair_surface_with_meshfix", broken_repair) + + with pytest.raises(ValueError, match="meshfix implementation failed"): + tetrahedralize_mod.tetrahedralize(source, remesh_on_failure=False) + + +def test_unexpected_pyacvd_error_is_not_relabelled_as_geometry_rejection(monkeypatch): + source = pv.Cube().triangulate() + + def worker(surface, tet_args, tet_kwargs, worker_script, python_exe, *, strategy): + raise _geometry_worker_error(surface, strategy, tet_kwargs) + + def broken_remesh(*args, **kwargs): + raise RuntimeError("remesher implementation failed") + + monkeypatch.setattr(tetrahedralize_mod, "_tetgen_worker_tetrahedralize", worker) + monkeypatch.setattr(tetrahedralize_mod, "uniform_remesh_surface", broken_remesh) + + with pytest.raises(RuntimeError, match="remesher implementation failed"): + tetrahedralize_mod.tetrahedralize(source, repair_on_failure=False) + + +def test_open_pyacvd_candidate_is_repaired_before_tetgen(monkeypatch): + source = pv.Cube().triangulate() + face_rows = source.faces.reshape(-1, 4) + open_surface = pv.PolyData(source.points.copy(), face_rows[:-1].copy()) + worker_calls = [] + repair_calls = [] + + def worker(surface, tet_args, tet_kwargs, worker_script, python_exe, *, strategy): + worker_calls.append(strategy) + if strategy == "original": + raise TetGenWorkerError( + TetGenAttemptReport( + strategy=strategy, + status="failed", + surface=summarize_surface(surface), + duration_seconds=0.01, + recoverable=True, + tetgen_args=(), + tetgen_kwargs=dict(tet_kwargs), + diagnostics=summarize_tetgen_output( + "Warning: A segment and a facet intersect.\n", "", 1 + ), + message="TetGen rejected the original surface.", + ) + ) + if strategy != "pyacvd_meshfix": + raise AssertionError("Open PyACVD surface reached TetGen") + nodes = np.array( + [[0.0, 0.0, 0.0], [0.25, 0.0, 0.0], [0.0, 0.25, 0.0], [0.0, 0.0, 0.25]] + ) + return nodes, np.array([[0, 1, 2, 3]], dtype=np.int64) + + def repair(surface, **kwargs): + repair_calls.append(surface.n_open_edges) + if len(repair_calls) == 1: + return open_surface.copy(deep=True) + return source.copy(deep=True) + + monkeypatch.setattr(tetrahedralize_mod, "_tetgen_worker_tetrahedralize", worker) + monkeypatch.setattr(tetrahedralize_mod, "repair_surface_with_meshfix", repair) + monkeypatch.setattr( + tetrahedralize_mod, + "uniform_remesh_surface", + lambda *args, **kwargs: open_surface.copy(deep=True), + ) + + result = tetrahedralize_mod.tetrahedralize(source, return_result=True) + + assert result.report.selected_strategy == "pyacvd_meshfix" + assert worker_calls == ["original", "pyacvd_meshfix"] + assert repair_calls == [0, open_surface.n_open_edges] + + +@pytest.mark.parametrize("switches", [None, "pqQ"]) +def test_quiet_unknown_failure_runs_isolated_geometry_diagnostic( + tmp_path, + monkeypatch, + switches, +): + worker = tmp_path / "diagnostic_worker.py" + worker.write_text( + """\ +import json +import pathlib +import sys + +with open(sys.argv[3], 'r') as stream: + options = json.load(stream)['kwargs'] +switches = options.get('switches') +if switches: + diagnostic_enabled = 'd' in switches and 'V' in switches and 'Q' not in switches +else: + diagnostic_enabled = options.get('diagnose') and options.get('quiet') is False +if diagnostic_enabled: + pathlib.Path('_skipped.node').write_text('diagnostic') + print('Warning: A segment and a facet intersect.') + print(' segment: [4,5] tag(-1).') + print(' facet triangle: [1,2,3] tag(-1).') + raise SystemExit(3) +print('RuntimeError: Failed to tetrahedralize: Unknown exception', file=sys.stderr) +raise SystemExit(1) +""" + ) + monkeypatch.chdir(tmp_path) + + tetgen_options = {"switches": switches} if switches is not None else {} + with pytest.raises(TetrahedralizationError) as error_info: + tetrahedralize_mod.tetrahedralize( + pv.Cube().triangulate(), + worker_script=str(worker), + python_exe=sys.executable, + repair_on_failure=False, + remesh_on_failure=False, + return_result=True, + **tetgen_options, + ) + + report = error_info.value.report + assert len(report.attempts) == 1 + assert report.attempts[0].diagnostics.segment_facet_intersections == 1 + assert "intersecting surface facets" in str(error_info.value) + assert not list(tmp_path.glob("_skipped.*")) + + +def test_relative_worker_path_is_resolved_before_temporary_chdir(tmp_path, monkeypatch): + worker = tmp_path / "relative_worker.py" + worker.write_text( + """\ +import numpy as np +import sys + +nodes = np.array( + [[0.0, 0.0, 0.0], [1.0, 0.0, 0.0], [0.0, 1.0, 0.0], [0.0, 0.0, 1.0]] +) +elems = np.array([[0, 1, 2, 3]], dtype=np.int64) +np.savez(sys.argv[2], nodes=nodes, elems=elems) +""" + ) + monkeypatch.chdir(tmp_path) + + result = tetrahedralize_mod.tetrahedralize( + pv.Cube().triangulate(), + worker_script="relative_worker.py", + python_exe=sys.executable, + repair_on_failure=False, + remesh_on_failure=False, + return_result=True, + ) + + assert result.grid.n_cells == 1 + + +def test_worker_launch_path_resolution_preserves_path_lookup(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + + worker, interpreter = tetrahedralize_mod._resolve_worker_launch_paths( + "worker.py", + os.path.join("env", "python"), + ) + _, path_interpreter = tetrahedralize_mod._resolve_worker_launch_paths( + "worker.py", + "python", + ) + + assert worker == str(tmp_path / "worker.py") + assert interpreter == str(tmp_path / "env" / "python") + assert path_interpreter == "python" + + +def test_opaque_native_abort_runs_isolated_geometry_diagnostic(tmp_path): + worker = tmp_path / "native_diagnostic_worker.py" + worker.write_text( + """\ +import json +import sys + +with open(sys.argv[3], 'r') as stream: + options = json.load(stream)['kwargs'] +if options.get('diagnose'): + print('Warning: Two facets exactly intersect.') + print(' 1st facet triangle: [1,2,3] tag(-1).') + print(' 2nd facet triangle: [4,5,6] tag(-1).') + raise SystemExit(3) +print('free(): invalid next size (normal)', file=sys.stderr) +raise SystemExit(1) +""" + ) + + with pytest.raises(TetrahedralizationError) as error_info: + tetrahedralize_mod.tetrahedralize( + pv.Cube().triangulate(), + worker_script=str(worker), + python_exe=sys.executable, + repair_on_failure=False, + remesh_on_failure=False, + ) + + diagnostic = error_info.value.report.attempts[0].diagnostics + assert diagnostic.facet_facet_intersections == 1 + assert "intersecting surface facets" in str(error_info.value) + + +def test_disabling_repair_and_remesh_returns_structured_direct_failure(monkeypatch): + def worker(surface, tet_args, tet_kwargs, worker_script, python_exe, *, strategy): + raise _geometry_worker_error(surface, strategy, tet_kwargs) + + monkeypatch.setattr(tetrahedralize_mod, "_tetgen_worker_tetrahedralize", worker) + + with pytest.raises(TetrahedralizationError) as error_info: + tetrahedralize_mod.tetrahedralize( + pv.Cube().triangulate(), + repair_on_failure=False, + remesh_on_failure=False, + ) + + assert [attempt.strategy for attempt in error_info.value.report.attempts] == [ + "original" + ] + + +def test_nongeometry_worker_failure_does_not_start_recovery(monkeypatch): + source = pv.Cube().triangulate() + infrastructure_attempt = TetGenAttemptReport( + strategy="original", + status="failed", + surface=summarize_surface(source), + duration_seconds=0.01, + recoverable=False, + tetgen_args=(), + tetgen_kwargs={"verbose": 0}, + diagnostics=summarize_tetgen_output( + "", "ModuleNotFoundError: missing dependency\n", 1 + ), + message="TetGen worker infrastructure failed for the original surface.", + ) + + def worker(*args, **kwargs): + raise TetGenWorkerError(infrastructure_attempt) + + def unexpected_recovery(*args, **kwargs): + raise AssertionError("Infrastructure failures must not start geometry recovery") + + monkeypatch.setattr(tetrahedralize_mod, "_tetgen_worker_tetrahedralize", worker) + monkeypatch.setattr( + tetrahedralize_mod, "repair_surface_with_meshfix", unexpected_recovery + ) + monkeypatch.setattr(tetrahedralize_mod, "uniform_remesh_surface", unexpected_recovery) + + with pytest.raises(TetGenWorkerError) as error_info: + tetrahedralize_mod.tetrahedralize(source) + + assert error_info.value.recoverable is False + + +def test_all_geometry_attempts_fail_with_ordered_aggregate_report(monkeypatch): + source = pv.Cube().triangulate() + + def worker(surface, tet_args, tet_kwargs, worker_script, python_exe, *, strategy): + raise _geometry_worker_error(surface, strategy, tet_kwargs) + + monkeypatch.setattr(tetrahedralize_mod, "_tetgen_worker_tetrahedralize", worker) + monkeypatch.setattr( + tetrahedralize_mod, + "repair_surface_with_meshfix", + lambda *args, **kwargs: source.copy(deep=True), + ) + monkeypatch.setattr( + tetrahedralize_mod, + "uniform_remesh_surface", + lambda *args, **kwargs: source.copy(deep=True), + ) + + with pytest.raises(TetrahedralizationError) as error_info: + tetrahedralize_mod.tetrahedralize(source) + + report = error_info.value.report + assert [attempt.strategy for attempt in report.attempts] == [ + "original", + "meshfix", + "pyacvd", + ] + assert report.selected_strategy is None + assert "intersecting surface facets" in report.user_summary() + + +def test_meshfix_repair_preserves_components_by_policy(monkeypatch): + source = pv.Cube().triangulate() + captured = {} + + class FakeMeshFix: + def __init__(self, points, faces): + self.v = np.asarray(points).copy() + self.f = np.asarray(faces).copy() + + def repair(self, **kwargs): + captured.update(kwargs) + + monkeypatch.setattr(tetrahedralize_mod.pymeshfix, "MeshFix", FakeMeshFix) + + repaired = tetrahedralize_mod.repair_surface_with_meshfix(source) + + assert repaired.n_points == source.n_points + assert captured["joincomp"] is False + assert captured["remove_smallest_components"] is False + + +def test_recovery_validation_rejects_component_loss(): + first = pv.Cube(center=(0.0, 0.0, 0.0)).triangulate() + second = pv.Cube(center=(3.0, 0.0, 0.0)).triangulate() + source = first.merge(second, merge_points=False) + + with pytest.raises(ValueError, match="connected-component count from 2 to 1"): + tetrahedralize_mod.validate_recovery_surface( + source, + first, + max_distance_ratio=1.0, + ) + + +@pytest.mark.parametrize("value", [0.0, -0.1, np.nan, np.inf]) +def test_repair_distance_ratio_must_be_finite_and_positive(value): + with pytest.raises(ValueError, match="finite and positive"): + tetrahedralize_mod.tetrahedralize( + pv.Cube().triangulate(), + repair_max_distance_ratio=value, + ) + + +@pytest.mark.parametrize( + "value", + [True, np.bool_(True), np.array(0.1), np.array([0.1]), "0.1", None], +) +def test_repair_distance_ratio_must_be_a_real_scalar(value): + with pytest.raises(ValueError, match="finite and positive scalar"): + tetrahedralize_mod.tetrahedralize( + pv.Cube().triangulate(), + repair_max_distance_ratio=value, + ) + + +@pytest.mark.parametrize("value", [-1.0, np.nan, np.inf]) +def test_remesh_clean_tolerance_must_be_finite_and_nonnegative(value): + with pytest.raises(ValueError, match="finite and non-negative"): + tetrahedralize_mod.tetrahedralize( + pv.Cube().triangulate(), + remesh_clean_tolerance=value, + ) + + +@pytest.mark.parametrize( + "value", + [True, np.bool_(True), np.array(0.1), np.array([0.1]), "0.1"], +) +def test_remesh_clean_tolerance_must_be_a_real_scalar(value): + with pytest.raises(ValueError, match="finite and non-negative scalar"): + tetrahedralize_mod.tetrahedralize( + pv.Cube().triangulate(), + remesh_clean_tolerance=value, + ) + + +@pytest.mark.parametrize( + ("option", "value", "message"), + [ + ("remesh_subdivisions", np.nan, "non-negative integer"), + ("remesh_subdivisions", 1.5, "non-negative integer"), + ("remesh_subdivisions", True, "non-negative integer"), + ("remesh_clusters", np.nan, "positive integer"), + ("remesh_clusters", 2.5, "positive integer"), + ("remesh_clusters", True, "positive integer"), + ], +) +def test_remesh_integer_controls_are_validated_before_tetgen( + monkeypatch, + option, + value, + message, +): + def unexpected_worker(*args, **kwargs): + raise AssertionError("Invalid recovery controls must fail before TetGen") + + monkeypatch.setattr( + tetrahedralize_mod, + "_tetgen_worker_tetrahedralize", + unexpected_worker, + ) + + with pytest.raises(ValueError, match=message): + tetrahedralize_mod.tetrahedralize( + pv.Cube().triangulate(), + **{option: value}, + )