from __future__ import annotations import ast import sys import time from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parent.parent if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) EXPECTED_FACE_ISOLATED_OPERATIONS = { "push_pull_face", "move_face_plane_offset_local", "resize_face_area_local", "resize_face_area", "resize_face_size_local", "resize_face_size_owning_scale", "move_face_center_local", "resize_shell_thickness", "resize_shell_thickness_owning_scale", "resize_cylindrical_height", "resize_cylindrical_boss_height", "resize_cylindrical_height_owning_scale", "resize_cone_reference_radius", "resize_cone_semi_angle", "resize_sphere_radius", "resize_torus_radius", } def _source_tree(relative_path: str) -> ast.Module: return ast.parse((PROJECT_ROOT / relative_path).read_text(encoding="utf-8")) def _function_node(tree: ast.Module, name: str) -> ast.FunctionDef: for node in ast.walk(tree): if isinstance(node, ast.FunctionDef) and node.name == name: return node raise SystemExit(f"function {name!r} was not found") def _constant_string_set(node: ast.AST, variable_name: str) -> set[str]: for child in ast.walk(node): if not isinstance(child, ast.Assign): continue if not any(isinstance(target, ast.Name) and target.id == variable_name for target in child.targets): continue if not isinstance(child.value, (ast.Set, ast.Tuple, ast.List)): raise SystemExit(f"{variable_name} should be a literal collection") values: set[str] = set() for item in child.value.elts: if not isinstance(item, ast.Constant) or not isinstance(item.value, str): raise SystemExit(f"{variable_name} contains a non-string item: {ast.dump(item)}") values.add(item.value) return values raise SystemExit(f"{variable_name} assignment was not found") def _worker_operation_set(node: ast.AST) -> set[str]: operations: set[str] = set() for child in ast.walk(node): if not isinstance(child, ast.Compare): continue if not isinstance(child.left, ast.Name) or child.left.id != "operation": continue if len(child.ops) != 1 or not isinstance(child.ops[0], ast.Eq): continue if len(child.comparators) != 1: continue comparator = child.comparators[0] if isinstance(comparator, ast.Constant) and isinstance(comparator.value, str): operations.add(comparator.value) return operations def _assert_same(label: str, actual: set[str], expected: set[str]) -> None: missing = sorted(expected - actual) extra = sorted(actual - expected) if missing or extra: raise SystemExit(f"{label} mismatch: missing={missing}, extra={extra}") PROPERTY_FACE_ACTION_TO_ISOLATED_OPERATION = { "push_pull_face": "push_pull_face", "move_selected_face_plane_position_local": "move_face_plane_offset_local", "resize_face_area_local": "resize_face_area_local", "resize_face_area": "resize_face_area", "resize_face_width_local": "resize_face_size_local", "resize_face_height_local": "resize_face_size_local", "resize_face_width_owning_scale": "resize_face_size_owning_scale", "resize_face_height_owning_scale": "resize_face_size_owning_scale", "move_selected_face_center_local": "move_face_center_local", "resize_shell_thickness": "resize_shell_thickness", "resize_shell_thickness_owning_scale": "resize_shell_thickness_owning_scale", "resize_cylinder_height": "resize_cylindrical_height", "resize_boss_height": "resize_cylindrical_boss_height", "resize_cylindrical_height_owning_scale": "resize_cylindrical_height_owning_scale", "resize_cone_reference_radius": "resize_cone_reference_radius", "resize_cone_semi_angle": "resize_cone_semi_angle", "resize_sphere_radius": "resize_sphere_radius", "resize_torus_major_radius": "resize_torus_radius", "resize_torus_minor_radius": "resize_torus_radius", } PROPERTY_FACE_ACTIONS_ALLOWED_IN_MAIN_THREAD = { # Pure owning-shape affine translations are intentionally lightweight and do # not enter the OCC rebuild/Boolean isolation worker. "move_selected_face_center", "move_selected_face_plane_position_by_translation", } ACTION_IMPLEMENTATION_METHOD = { "resize_face_width_local": "_resize_face_size_local", "resize_face_height_local": "_resize_face_size_local", "resize_face_width_owning_scale": "_resize_face_size_owning_scale", "resize_face_height_owning_scale": "_resize_face_size_owning_scale", "resize_torus_major_radius": "_resize_torus_radius", "resize_torus_minor_radius": "_resize_torus_radius", } def _property_spec_actions_for_keys(specs: list[dict[str, object]], keys: set[str]) -> set[str]: actions: set[str] = set() for spec in specs: key = str(spec.get("key", "")) if key not in keys: continue action = str(spec.get("action") or "") if action: actions.add(action) scope_modes = spec.get("scope_modes") if isinstance(scope_modes, dict): for mode in scope_modes.values(): if not isinstance(mode, dict): continue action = str(mode.get("action") or "") if action: actions.add(action) return actions def _face_property_actions_from_specs() -> set[str]: from verify_property_editor_specs import _specs plane_info = { "surface": "plane", "area": 100.0, "area_center": (5.0, 5.0, 0.0), "bbox_center": (5.0, 5.0, 0.0), "local_face_width": 10.0, "local_face_height": 10.0, "local_face_size_center": (5.0, 5.0, 0.0), "plane_origin": (0.0, 0.0, 0.0), "push_pull_outward_direction": (0.0, 0.0, 1.0), "normal": (0.0, 0.0, 1.0), "topology_relation_depth": 1, "topology_relation_status": "ready", "same_domain_face_count": 1, "first_level_boundary_edge_count": 4, "first_level_boundary_vertex_count": 4, "first_level_adjacent_face_count": 4, } shell_info = { **plane_info, "shell_region_status": "candidate", "shell_thickness_estimate": 2.0, "shell_current_thickness": 2.0, "shell_signed_thickness": 2.0, "shell_opposite_face_id": 2, "shell_overlap_ratio_estimate": 1.0, } generic_cylinder_info = { "surface": "cylinder", "feature_guess": "", "diameter": 6.0, "radius": 3.0, "angular_span": 6.283185307179586, "height_estimate": 11.0, "same_domain_height_estimate": 11.0, "area": 207.0, "area_center": (0.0, 0.0, 5.5), "bbox_center": (0.0, 0.0, 5.5), "axis": (0.0, 0.0, 1.0), "axis_point": (0.0, 0.0, 0.0), "axis_center": (0.0, 0.0, 5.5), } boss_info = { **generic_cylinder_info, "feature_guess": "boss/outer-round candidate", "height_estimate": 5.0, "same_domain_height_estimate": 5.0, "feature_start_end_face_ids": (1,), } cases = ( ( plane_info, { "area", "local_face_width", "local_face_height", "face_center_position", "face_target_normal_position", }, ), (shell_info, {"shell_thickness_estimate"}), (generic_cylinder_info, {"cylinder_height"}), (boss_info, {"boss_height"}), ( { "surface": "cone", "reference_radius": 5.0, "diameter": 10.0, "radius": 5.0, "semi_angle": 0.5, "area": 100.0, "area_center": (0.0, 0.0, 0.0), "bbox_center": (0.0, 0.0, 0.0), "axis": (0.0, 0.0, 1.0), "axis_point": (0.0, 0.0, 0.0), }, {"cone_reference_radius", "cone_reference_diameter", "cone_semi_angle_degrees"}, ), ( { "surface": "sphere", "radius": 5.0, "diameter": 10.0, "area": 100.0, "area_center": (0.0, 0.0, 0.0), "bbox_center": (0.0, 0.0, 0.0), "center": (0.0, 0.0, 0.0), }, {"sphere_radius", "sphere_diameter"}, ), ( { "surface": "torus", "major_radius": 8.0, "minor_radius": 2.0, "major_diameter": 16.0, "minor_diameter": 4.0, "area": 100.0, "area_center": (0.0, 0.0, 0.0), "bbox_center": (0.0, 0.0, 0.0), "center": (0.0, 0.0, 0.0), }, {"torus_major_radius", "torus_major_diameter", "torus_minor_radius", "torus_minor_diameter"}, ), ) actions: set[str] = set() for info, keys in cases: actions.update(_property_spec_actions_for_keys(_specs(info), keys)) return actions def _method_isolation_operations(tree: ast.Module, method_name: str) -> set[str]: method = _function_node(tree, method_name) operations: set[str] = set() for child in ast.walk(method): if not isinstance(child, ast.Call): continue callee = child.func if not isinstance(callee, ast.Attribute) or callee.attr != "_isolation_for_plan": continue if len(child.args) < 2: raise SystemExit(f"{method_name} calls _isolation_for_plan without an operation argument") operation_arg = child.args[1] if not isinstance(operation_arg, ast.Constant) or not isinstance(operation_arg.value, str): raise SystemExit(f"{method_name} should pass a literal isolated operation name") operations.add(operation_arg.value) return operations def _assert_property_face_actions_are_isolated(actions_tree: ast.Module) -> None: property_actions = _face_property_actions_from_specs() covered = set(PROPERTY_FACE_ACTION_TO_ISOLATED_OPERATION) | PROPERTY_FACE_ACTIONS_ALLOWED_IN_MAIN_THREAD missing = sorted(property_actions - covered) stale = sorted(set(PROPERTY_FACE_ACTION_TO_ISOLATED_OPERATION) - property_actions) if missing or stale: raise SystemExit(f"Face property action contract mismatch: missing={missing}, stale={stale}") for action, isolated_operation in sorted(PROPERTY_FACE_ACTION_TO_ISOLATED_OPERATION.items()): if isolated_operation not in EXPECTED_FACE_ISOLATED_OPERATIONS: raise SystemExit(f"{action} maps to non-worker isolated operation {isolated_operation!r}") method_name = ACTION_IMPLEMENTATION_METHOD.get(action, action) actual_operations = _method_isolation_operations(actions_tree, method_name) if isolated_operation not in actual_operations: raise SystemExit( f"{action} should enter isolated operation {isolated_operation!r} through {method_name}; " f"actual={sorted(actual_operations)}" ) def main() -> int: worker_execute = _function_node(_source_tree("step_editor/isolated_edit_worker.py"), "_execute") worker_ops = _worker_operation_set(worker_execute) _assert_same("isolated worker operations", worker_ops, EXPECTED_FACE_ISOLATED_OPERATIONS) actions_tree = _source_tree("step_editor/window_actions.py") isolation_func = _function_node(actions_tree, "_isolation_for_plan") ui_ops = _constant_string_set(isolation_func, "isolated_face_operations") _assert_same("UI isolation operations", ui_ops, EXPECTED_FACE_ISOLATED_OPERATIONS) _assert_same("UI/worker isolation operation contract", ui_ops, worker_ops) _assert_property_face_actions_are_isolated(actions_tree) from step_editor.window_actions import WindowActionMixin class _Probe(WindowActionMixin): pass probe = _Probe() for risk in ("low", "medium", "high"): for operation in sorted(EXPECTED_FACE_ISOLATED_OPERATIONS): isolation = probe._isolation_for_plan( {"risk": risk}, operation, [1, 2, 3], timeout_seconds=7.5, ) if isolation is None: raise SystemExit(f"{operation}/{risk} should request isolated execution") if isolation.get("operation") != operation: raise SystemExit(f"{operation}/{risk} returned wrong operation: {isolation}") if isolation.get("args") != [1, 2, 3]: raise SystemExit(f"{operation}/{risk} returned wrong args: {isolation}") if float(isolation.get("timeout_seconds") or 0.0) != 7.5: raise SystemExit(f"{operation}/{risk} returned wrong timeout: {isolation}") if str(isolation.get("reason") or "") != f"{risk}-risk-face-occ-edit": raise SystemExit(f"{operation}/{risk} returned wrong reason: {isolation}") if probe._isolation_for_plan({"risk": "blocked"}, "push_pull_face", [0, 1]) is not None: raise SystemExit("blocked plans should not enter isolated execution") if probe._isolation_for_plan({"risk": "high"}, "resize_cylindrical_hole", [0, 20]) is not None: raise SystemExit("non-Face feature operations should not be covered by the Face isolation contract") if probe._isolation_for_plan({"risk": "high"}, "unknown_operation", []) is not None: raise SystemExit("unknown operations should not enter isolated execution") from step_editor.model import StepModel complex_path = PROJECT_ROOT / "assets" / "models" / "geom_extract.step" if complex_path.exists(): model = StepModel.load(complex_path) complex_face_id = 1 complex_probe = _Probe() complex_probe.model = model complex_probe.step_path = complex_path complex_probe.selected_face_id = complex_face_id complex_probe.current_info_values = model.quick_face_info(complex_face_id) checks = ( ("area local", lambda: complex_probe._quick_blocked_face_area_local_plan(complex_face_id, 2000.0), "blocked"), ( "width local", lambda: complex_probe._quick_blocked_face_size_local_plan(complex_face_id, 50.0, "width"), "blocked", ), ( "offset local", lambda: complex_probe._quick_blocked_face_plane_offset_local_plan(complex_face_id, 50.0), "blocked", ), ( "center local", lambda: complex_probe._quick_blocked_face_center_local_plan(complex_face_id, (0.0, 50.0, 0.0)), "blocked", ), ( "shell thickness", lambda: complex_probe._deferred_shell_thickness_plan_if_needed(complex_face_id, 10.0), "caution", ), ) for label, callback, expected_status in checks: started = time.perf_counter() plan = callback() elapsed = time.perf_counter() - started if plan is None: raise SystemExit(f"{label} complex Face UI preflight should return a quick plan") if elapsed > 0.05: raise SystemExit(f"{label} complex Face UI preflight should be quick, elapsed={elapsed:.3f}s") if str(plan.get("status")) != expected_status: raise SystemExit(f"{label} complex Face UI preflight returned wrong status: {plan}") if expected_status == "blocked" and not plan.get("ui_quick_blocked_local_face_plan"): raise SystemExit(f"{label} should be marked as a quick local-Face blocker: {plan}") if label == "shell thickness" and not plan.get("ui_deferred_shell_thickness_plan"): raise SystemExit(f"{label} should defer full shell-thickness planning: {plan}") uncached_probe = _Probe() uncached_probe.model = model uncached_probe.step_path = complex_path uncached_probe.selected_face_id = complex_face_id uncached_probe.current_info_values = {} uncached_checks = ( ("area local uncached", lambda: uncached_probe._quick_blocked_face_area_local_plan(complex_face_id, 2000.0)), ( "offset local uncached", lambda: uncached_probe._quick_blocked_face_plane_offset_local_plan(complex_face_id, 50.0), ), ) for label, callback in uncached_checks: started = time.perf_counter() plan = callback() elapsed = time.perf_counter() - started if elapsed > 0.05: raise SystemExit(f"{label} should use quick_face_info fallback, elapsed={elapsed:.3f}s") if plan is None or plan.get("status") != "blocked": raise SystemExit(f"{label} should still return a quick blocked plan: {plan}") if not plan.get("ui_quick_blocked_local_face_plan"): raise SystemExit(f"{label} should be marked as a quick local-Face blocker: {plan}") multi_boundary_face_id = 594 multi_boundary_probe = _Probe() multi_boundary_probe.model = model multi_boundary_probe.step_path = complex_path multi_boundary_probe.selected_face_id = multi_boundary_face_id multi_boundary_probe.current_info_values = model.quick_face_info(multi_boundary_face_id) started = time.perf_counter() inward_plan = multi_boundary_probe._push_pull_plan_for_action(multi_boundary_face_id, -1.0) elapsed = time.perf_counter() - started if elapsed > 0.05: raise SystemExit(f"multi-boundary inward push/pull UI preflight should be quick, elapsed={elapsed:.3f}s") if inward_plan.get("status") != "blocked": raise SystemExit(f"multi-boundary inward push/pull UI preflight should be blocked: {inward_plan}") if not inward_plan.get("ui_quick_blocked_push_pull_plan"): raise SystemExit(f"multi-boundary inward push/pull should be marked as a quick blocker: {inward_plan}") message = str(inward_plan.get("message") or "") if "二级关系" not in message or "通用 OCCT 布尔" not in message: raise SystemExit( f"multi-boundary inward push/pull blocker should explain topology depth and Boolean risk: " f"{inward_plan}" ) for required in ("Face 594", "内边界", "一级边界 Edge", "一级相邻 Face"): if required not in message: raise SystemExit( f"multi-boundary inward push/pull blocker should include readable topology evidence " f"{required!r}: {inward_plan}" ) diagnostics = multi_boundary_probe._edit_failure_diagnostics( { "operation_name": "推拉平面", "target": f"Face {multi_boundary_face_id}", "parameters": { "surface": inward_plan.get("surface"), "semantic_distance": -1.0, "current_plane_position": inward_plan.get("current_plane_position"), "target_plane_position": inward_plan.get("target_plane_position"), "edit_strategy_label": inward_plan.get("edit_strategy_label"), "push_pull_message": inward_plan.get("message"), "push_pull_risk": inward_plan.get("risk"), "selected_boundary_wires": inward_plan.get("selected_boundary_wires"), "selected_inner_boundary_wires": inward_plan.get("selected_inner_boundary_wires"), "selected_has_inner_boundaries": inward_plan.get("selected_has_inner_boundaries"), "same_domain_face_count": inward_plan.get("same_domain_face_count"), "first_level_boundary_edge_count": inward_plan.get("first_level_boundary_edge_count"), "first_level_boundary_vertex_count": inward_plan.get("first_level_boundary_vertex_count"), "first_level_adjacent_face_count": inward_plan.get("first_level_adjacent_face_count"), "first_level_topology_note": inward_plan.get("first_level_topology_note"), "topology_ignored_relation_note": inward_plan.get("topology_ignored_relation_note"), "planar_cap_extension_kind": inward_plan.get("planar_cap_extension_kind"), "planar_cap_extension_method": inward_plan.get("planar_cap_extension_method"), }, "isolation": { "operation": "push_pull_face", "args": [multi_boundary_face_id, -1.0], "timeout_seconds": 180.0, "reason": "high-risk-face-occ-edit", }, } ) for required in ("诊断信息", "操作: 推拉平面 / Face 594", "一级关系证据", "可能原因", "二级", "隔离保护"): if required not in diagnostics: raise SystemExit(f"edit failure diagnostics should include {required!r}: {diagnostics}") if multi_boundary_probe._isolation_for_plan(inward_plan, "push_pull_face", [multi_boundary_face_id, -1.0]) is not None: raise SystemExit(f"blocked multi-boundary inward push/pull should not enter isolated execution: {inward_plan}") print( "Face UI isolation contract ok: " f"{len(EXPECTED_FACE_ISOLATED_OPERATIONS)} operations are shared by UI and worker." ) return 0 if __name__ == "__main__": raise SystemExit(main())