diff --git a/__init__.py b/__init__.py index 321034b..1e960b1 100644 --- a/__init__.py +++ b/__init__.py @@ -36,6 +36,11 @@ def register(): print('Nodeslicer Loaded') def unregister(): + try: + tree.cancel_timer() + except Exception as e: + print(f"Error canceling timer: {e}") + try: if tree.depsgraph_update_handler in bpy.app.handlers.depsgraph_update_post: bpy.app.handlers.depsgraph_update_post.remove(tree.depsgraph_update_handler) diff --git a/core/nodeslicer/bpy_ns_types.py b/core/nodeslicer/bpy_ns_types.py index d909c72..bf1ccee 100644 --- a/core/nodeslicer/bpy_ns_types.py +++ b/core/nodeslicer/bpy_ns_types.py @@ -3,35 +3,53 @@ from typing import Any, Callable import bpy from bpy.props import PointerProperty, StringProperty, FloatProperty, IntProperty, EnumProperty, BoolProperty, FloatVectorProperty, CollectionProperty +def _is_btype(val: Any, type_name: str) -> bool: + if val is None: + return False + btype = getattr(bpy.types, type_name, None) + if isinstance(btype, type): + try: + if isinstance(val, btype): + return True + except TypeError: + pass + val_cls_name = getattr(val, "__class__", type(None)).__name__ + return val_cls_name == type_name or f"Dummy{type_name}" in val_cls_name + def notify_tree_changed(self, context): from . import tree - if tree._is_updating: + if getattr(tree, "_is_updating", False): return - tree._tree_changed = True node = None socket = None + changed_tree = None - if isinstance(self, bpy.types.Node): + if _is_btype(self, "NodeTree"): + changed_tree = self + elif _is_btype(self, "Node"): node = self - elif isinstance(self, bpy.types.NodeSocket): + changed_tree = getattr(self, "id_data", None) + elif _is_btype(self, "NodeSocket"): socket = self - node = self.node - elif isinstance(self, bpy.types.PropertyGroup): + node = getattr(self, "node", None) + changed_tree = getattr(self, "id_data", getattr(node, "id_data", None) if node else None) + elif _is_btype(self, "PropertyGroup"): try: id_data = self.id_data - if isinstance(id_data, bpy.types.NodeTree): + if _is_btype(id_data, "NodeTree"): + changed_tree = id_data path = self.path_from_id() if ".inputs[" in path or ".outputs[" in path: socket_path = path.split(".items[")[0] if ".items[" in path else path.rsplit(".", 1)[0] resolved = id_data.path_resolve(socket_path) - if isinstance(resolved, bpy.types.NodeSocket): + if _is_btype(resolved, "NodeSocket"): socket = resolved node = socket.node else: node_path = path.rsplit(".", 1)[0] resolved = id_data.path_resolve(node_path) - if isinstance(resolved, bpy.types.Node): + if _is_btype(resolved, "Node"): node = resolved except Exception: pass @@ -41,6 +59,8 @@ def notify_tree_changed(self, context): node = getattr(context, "node", None) if socket is None: socket = getattr(context, "socket", None) + if changed_tree is None and node: + changed_tree = getattr(node, "id_data", None) if socket and hasattr(socket, "update"): try: @@ -54,6 +74,9 @@ def notify_tree_changed(self, context): except Exception as e: print(f"Error updating node {node.name}: {e}") + if hasattr(tree, "request_auto_refresh"): + tree.request_auto_refresh(changed_tree) + class MockPropertyDescriptor: def __init__(self, default: Any = None): self.default = default diff --git a/core/nodeslicer/tree.py b/core/nodeslicer/tree.py index 3b53d05..629ec7f 100644 --- a/core/nodeslicer/tree.py +++ b/core/nodeslicer/tree.py @@ -9,101 +9,311 @@ from ...registry import register_class if TYPE_CHECKING: from bpy.stub_internal.rna_enums import OperatorReturnItems -def get_referenced_objects(node, visited=None): +def _get_scene_object(name: str) -> Any: + if not name: + return None + objects = getattr(bpy.data, "objects", None) + if objects is None: + return None + if isinstance(objects, dict): + return objects.get(name) + if "Mock" in type(objects).__name__: + return None + if hasattr(objects, "get"): + return objects.get(name) + if hasattr(objects, "__getitem__"): + try: + return objects[name] + except Exception: + return None + return None + +def _is_btype(val: Any, type_name: str) -> bool: + if val is None: + return False + btype = getattr(bpy.types, type_name, None) + if isinstance(btype, type): + try: + if isinstance(val, btype): + return True + except TypeError: + pass + val_cls_name = getattr(val, "__class__", type(None)).__name__ + return val_cls_name == type_name or f"Dummy{type_name}" in val_cls_name + +def get_referenced_data(node: Any, visited: set[int] | None = None) -> tuple[set[Any], set[Any]]: + """Traverse upstream from node and collect all referenced bpy.types.Object and bpy.types.Collection.""" if visited is None: visited = set() - - ptr = node.as_pointer() + + ptr = node.as_pointer() if hasattr(node, "as_pointer") and callable(node.as_pointer) else id(node) + if ptr in visited: + return set(), set() + visited.add(ptr) + + objs: set[Any] = set() + cols: set[Any] = set() + + # 1. Group Node: check internal nodes inside the group + if getattr(node, "bl_idname", "") == "SLICENODE_Group": + group_tree = getattr(node, "slicer_tree", None) or getattr(node, "node_tree", None) + if group_tree and hasattr(group_tree, "nodes"): + for internal_node in group_tree.nodes: + sub_objs, sub_cols = get_referenced_data(internal_node, visited) + objs.update(sub_objs) + cols.update(sub_cols) + + # 2. CollectionObjects Node + if getattr(node, "bl_idname", "") == "SLICENODE_CollectionObjects": + col = getattr(node, "collection", None) + if col: + cols.add(col) + for o in getattr(col, "all_objects", getattr(col, "objects", [])): + if _is_btype(o, "Object"): + objs.add(o) + for outp in getattr(node, "outputs", []): + obj = _get_scene_object(getattr(outp, "identifier", "")) or _get_scene_object(getattr(outp, "name", "")) + if obj: + objs.add(obj) + + # 3. Check node RNA pointer properties for Object or Collection + for prop in getattr(getattr(node, "bl_rna", None), "properties", []): + if getattr(prop, "type", "") == 'POINTER': + val = getattr(node, prop.identifier, None) + if _is_btype(val, "Object"): + objs.add(val) + elif _is_btype(val, "Collection"): + cols.add(val) + for o in getattr(val, "all_objects", getattr(val, "objects", [])): + if _is_btype(o, "Object"): + objs.add(o) + + # 4. Check all input sockets + for inp in getattr(node, "inputs", []): + val = getattr(inp, "value", getattr(inp, "default_value", None)) + if _is_btype(val, "Object"): + objs.add(val) + elif _is_btype(val, "Collection"): + cols.add(val) + + # Follow upstream links + if getattr(inp, "is_linked", False): + for link in getattr(inp, "links", []): + from_node = getattr(link, "from_node", None) + if from_node: + if getattr(from_node, "bl_idname", "") == "SLICENODE_CollectionObjects": + from_sock = getattr(link, "from_socket", None) + if from_sock: + obj = _get_scene_object(getattr(from_sock, "identifier", "")) or _get_scene_object(getattr(from_sock, "name", "")) + if obj: + objs.add(obj) + sub_objs, sub_cols = get_referenced_data(from_node, visited) + objs.update(sub_objs) + cols.update(sub_cols) + + # 5. Check all output sockets + for outp in getattr(node, "outputs", []): + val = getattr(outp, "value", getattr(outp, "default_value", None)) + if _is_btype(val, "Object"): + objs.add(val) + elif _is_btype(val, "Collection"): + cols.add(val) + + # 6. Expand all objects to include their recursive children and parents/ancestors + from ...infra.blender.blender_bridge import get_all_children + all_objs = set(objs) + for o in list(objs): + # Children + try: + for child in get_all_children(o): + all_objs.add(child) + except Exception: + pass + # Ancestors + curr = getattr(o, "parent", None) + seen_parents = set() + while curr and curr not in seen_parents and len(seen_parents) < 100: + seen_parents.add(curr) + all_objs.add(curr) + if "Mock" in type(curr).__name__: + break + curr = getattr(curr, "parent", None) + + return all_objs, cols + +def get_referenced_objects(node: Any, visited: set[int] | None = None) -> set[Any]: + objs, _ = get_referenced_data(node, visited) + return objs + +def get_referenced_trees(node: Any, visited: set[int] | None = None) -> set[Any]: + """Collect all NodeTree datablocks referenced upstream (e.g. via group nodes).""" + if visited is None: + visited = set() + + ptr = node.as_pointer() if hasattr(node, "as_pointer") and callable(node.as_pointer) else id(node) if ptr in visited: return set() visited.add(ptr) - objs = set() + trees = set() + if getattr(node, "bl_idname", "") == "SLICENODE_Group": + group_tree = getattr(node, "slicer_tree", None) or getattr(node, "node_tree", None) + if group_tree: + trees.add(group_tree) + if hasattr(group_tree, "nodes"): + for internal_node in group_tree.nodes: + trees.update(get_referenced_trees(internal_node, visited)) - # Special case: if this is a group node, we must also check nodes inside the group - if node.bl_idname == "SLICENODE_Group" and getattr(node, "slicer_tree", None): - for internal_node in node.slicer_tree.nodes: - objs.update(get_referenced_objects(internal_node, visited)) + for inp in getattr(node, "inputs", []): + if getattr(inp, "is_linked", False): + for link in getattr(inp, "links", []): + from_node = getattr(link, "from_node", None) + if from_node: + trees.update(get_referenced_trees(from_node, visited)) - # Check all input sockets for objects - for inp in node.inputs: - # If it's an object socket, it might have a value if not linked - if inp.bl_idname == "NodeSocketObject": - if not inp.is_linked and getattr(inp, "value", getattr(inp, "default_value", None)): - val = getattr(inp, "value", getattr(inp, "default_value", None)) - if isinstance(val, bpy.types.Object): - objs.add(val) - - # Always follow links to find more objects upstream - if inp.is_linked: - for link in inp.links: - # Special case: if from_node is CollectionObjects, the object is identified by the socket identifier - if link.from_node.bl_idname == "SLICENODE_CollectionObjects": - obj = bpy.data.objects.get(link.from_socket.identifier) - if obj: - objs.add(obj) - - objs.update(get_referenced_objects(link.from_node, visited)) - - # Also check output sockets of the node itself, - # as some nodes might "hold" an object (like InputObject) - for outp in node.outputs: - if outp.bl_idname == "NodeSocketObject": - val = getattr(outp, "value", getattr(outp, "default_value", None)) - if isinstance(val, bpy.types.Object): - objs.add(val) - elif node.bl_idname == "SLICENODE_CollectionObjects" and outp.is_linked: - obj = bpy.data.objects.get(outp.identifier) - if obj: - objs.add(obj) - - return objs + return trees _is_updating = False _tree_changed = False +_changed_trees: set[Any] = set() +_timer_scheduled = False -def tree_update_handler(scene, depsgraph): - global _tree_changed, _is_updating +def _has_real_timers() -> bool: + timers = getattr(bpy.app, "timers", None) + if timers is None: + return False + if "Mock" in type(timers).__name__: + return False + reg = getattr(timers, "register", None) + if reg is None or "Mock" in type(reg).__name__: + return False + return True - # Names of objects that have changed - updated_object_names = {update.id.name for update in depsgraph.updates if isinstance(update.id, bpy.types.Object)} +def cancel_timer() -> None: + global _timer_scheduled + if _timer_scheduled and hasattr(bpy.app, "timers") and hasattr(bpy.app.timers, "unregister"): + try: + bpy.app.timers.unregister(_timer_callback) + except Exception: + pass + _timer_scheduled = False - # Also check for mesh updates (happens in Edit Mode) - updated_meshes = {update.id.name for update in depsgraph.updates if isinstance(update.id, bpy.types.Mesh)} - if updated_meshes: - for obj in bpy.data.objects: - if obj.type == 'MESH' and obj.data.name in updated_meshes: - updated_object_names.add(obj.name) +def _timer_callback() -> float | None: + global _timer_scheduled, _is_updating + if _is_updating: + # Retry in 50ms if currently evaluating + return 0.05 + _timer_scheduled = False + run_auto_refresh() + return None - # Note: custom NodeTree updates often don't appear in depsgraph.updates - # so we rely on _tree_changed flag set by property update callbacks. - - if not updated_object_names and not _tree_changed: +def schedule_auto_refresh(delay: float = 0.02) -> None: + global _timer_scheduled + if _timer_scheduled: + return + if _has_real_timers(): + try: + _timer_scheduled = True + bpy.app.timers.register(_timer_callback, first_interval=delay, persistent=True) + return + except Exception: + _timer_scheduled = False + # Fallback if timers are not running / available (e.g. tests or non-interactive CLI) + if not _is_updating: + run_auto_refresh() + +def request_auto_refresh(changed_tree: Any = None) -> None: + global _tree_changed, _changed_trees + if _is_updating: + return + _tree_changed = True + if changed_tree is not None: + _changed_trees.add(changed_tree) + schedule_auto_refresh() + +def _tag_redraw() -> None: + try: + wm = getattr(bpy.context, "window_manager", None) + if wm: + for window in getattr(wm, "windows", []): + screen = getattr(window, "screen", None) + if screen: + for area in getattr(screen, "areas", []): + if getattr(area, "type", "") in ('VIEW_3D', 'NODE_EDITOR'): + area.tag_redraw() + except Exception: + pass + +def run_auto_refresh( + updated_object_names: set[str] | None = None, + updated_collection_names: set[str] | None = None, + changed_trees: set[Any] | None = None, + force_refresh: bool = False +) -> None: + global _is_updating, _tree_changed, _changed_trees + + if _is_updating: return - tree_needs_refresh = _tree_changed - _tree_changed = False # reset for next run + active_changed_trees = set(_changed_trees) + if changed_trees: + active_changed_trees.update(changed_trees) + _changed_trees.clear() - for tree in bpy.data.node_groups: - if tree.bl_idname != "SLICETREE_NodeTree": + tree_needs_refresh = force_refresh or _tree_changed + _tree_changed = False + + has_updates = tree_needs_refresh or bool(updated_object_names) or bool(updated_collection_names) + if not has_updates: + return + + # Collect all SLICETREE_NodeTree datablocks + trees_to_check: list[Any] = [] + node_groups = getattr(bpy.data, "node_groups", None) + if node_groups is not None: + trees_to_check.extend(list(node_groups)) + try: + space_tree = getattr(getattr(bpy.context, "space_data", None), "edit_tree", None) + if space_tree and space_tree not in trees_to_check: + trees_to_check.append(space_tree) + except Exception: + pass + + executed_any = False + for tree in trees_to_check: + if getattr(tree, "bl_idname", "") != "SLICETREE_NodeTree": continue - for node in tree.nodes: - if node.bl_idname == "SLICENODE_Executor" and getattr(node, "auto_refresh", False): - # Trigger if the tree changed OR if a referenced object changed - should_refresh = tree_needs_refresh - - if not should_refresh: - referenced = get_referenced_objects(node) - referenced_names = {obj.name for obj in referenced if obj} - if updated_object_names & referenced_names: + for node in getattr(tree, "nodes", []): + if getattr(node, "bl_idname", "") == "SLICENODE_Executor" and getattr(node, "auto_refresh", False): + should_refresh = False + + if tree_needs_refresh: + if not active_changed_trees: should_refresh = True + else: + owner_tree = getattr(node, "id_data", tree) + dep_trees = {owner_tree} | get_referenced_trees(node) + if active_changed_trees & dep_trees: + should_refresh = True + + if not should_refresh: + ref_objs, ref_cols = get_referenced_data(node) + if updated_object_names: + ref_names = {o.name for o in ref_objs if o and hasattr(o, "name")} + if updated_object_names & ref_names: + should_refresh = True + if not should_refresh and updated_collection_names: + ref_col_names = {c.name for c in ref_cols if c and hasattr(c, "name")} + if updated_collection_names & ref_col_names: + should_refresh = True if should_refresh: _is_updating = True try: with execution_cache(): node.evaluate() + executed_any = True except Exception as e: print(f"Auto-refresh failed: {e}") import traceback @@ -111,6 +321,43 @@ def tree_update_handler(scene, depsgraph): finally: _is_updating = False + if executed_any: + _tag_redraw() + +def tree_update_handler(scene, depsgraph): + global _tree_changed, _is_updating + if _is_updating: + return + + # Names of objects that have changed + updated_object_names = {update.id.name for update in depsgraph.updates if _is_btype(update.id, "Object")} + + # Also check for mesh updates (happens in Edit Mode) + updated_meshes = {update.id.name for update in depsgraph.updates if _is_btype(update.id, "Mesh")} + if updated_meshes: + objects = getattr(bpy.data, "objects", []) + for obj in objects: + if getattr(obj, "type", "") == 'MESH' and getattr(getattr(obj, "data", None), "name", None) in updated_meshes: + updated_object_names.add(obj.name) + + # Check for collection updates + updated_collection_names = {update.id.name for update in depsgraph.updates if _is_btype(update.id, "Collection")} + + # Check if any NodeTree updated in depsgraph + updated_trees = {update.id for update in depsgraph.updates if _is_btype(update.id, "NodeTree")} + changed_trees = {t for t in updated_trees if getattr(t, "bl_idname", "") == "SLICETREE_NodeTree"} + if changed_trees: + _tree_changed = True + + if not updated_object_names and not updated_collection_names and not _tree_changed: + return + + run_auto_refresh( + updated_object_names=updated_object_names, + updated_collection_names=updated_collection_names, + changed_trees=changed_trees + ) + @persistent def depsgraph_update_handler(scene, depsgraph): global _is_updating, _tree_changed @@ -158,6 +405,17 @@ class SLICETREE_NodeTree(NodeTree): bl_group_idname = "SLICENODE_Group" bl_use_group_interface = True + def update(self): + super_update = getattr(super(), "update", None) + if callable(super_update): + try: + super_update() + except Exception: + pass + if _is_updating: + return + request_auto_refresh(self) + def valid_link(self, from_socket: Any, to_socket: Any) -> bool: geom_sockets = {"SLICESOCK_Mesh", "SLICESOCK_Paths", "SLICESOCK_Geometry"} out_s, in_s = from_socket, to_socket @@ -232,7 +490,7 @@ class SLICENODE_OT_GroupToggle(Operator): # If we have a target group node, we try to enter it if target_node and target_node.bl_idname == "SLICENODE_Group": - group_tree = getattr(target_node, "slicer_tree", None) + group_tree = getattr(target_node, "slicer_tree", None) or getattr(target_node, "node_tree", None) if group_tree: path.append(group_tree, node=target_node) return {'FINISHED'} diff --git a/infra/blender/blender_bridge.py b/infra/blender/blender_bridge.py index cba7afa..1983086 100644 --- a/infra/blender/blender_bridge.py +++ b/infra/blender/blender_bridge.py @@ -1,9 +1,15 @@ from __future__ import annotations -def get_all_children(obj): +def get_all_children(obj, visited=None): + if visited is None: + visited = set() children = [] - for child in getattr(obj, "children", []): - children += [child] + get_all_children(child) - - children = list(set(children)) - return children \ No newline at end of file + children_attr = getattr(obj, "children", []) + if "Mock" in type(children_attr).__name__ or not isinstance(children_attr, (list, tuple, set)): + return [] + for child in children_attr: + if child not in visited: + visited.add(child) + children.append(child) + children.extend(get_all_children(child, visited)) + return list(set(children)) \ No newline at end of file diff --git a/tests/test_auto_refresh.py b/tests/test_auto_refresh.py new file mode 100644 index 0000000..20a377e --- /dev/null +++ b/tests/test_auto_refresh.py @@ -0,0 +1,357 @@ +import unittest +from unittest.mock import MagicMock +import sys +from pathlib import Path + +PROJECT_ROOT = Path(__file__).resolve().parent.parent +sys.path.append(str(PROJECT_ROOT.parent)) + +class DummyObject: + def __init__(self, name="TestObj", obj_type='MESH', parent=None, children=None): + self.name = name + self.type = obj_type + self.parent = parent + self.children = children or [] + + def as_pointer(self): + return id(self) + +class DummyCollection: + def __init__(self, name="Col", objects=None): + self.name = name + self.objects = objects or [] + self.all_objects = self.objects + + def as_pointer(self): + return id(self) + +if 'bpy' not in sys.modules: + class DummyBase: pass + mock_types = MagicMock() + mock_types.NodeSocket = type("NodeSocket", (DummyBase,), {}) + mock_types.Node = type("Node", (DummyBase,), {}) + mock_types.NodeCustomGroup = type("NodeCustomGroup", (DummyBase,), {}) + mock_types.NodeTree = type("NodeTree", (DummyBase,), {}) + mock_types.PropertyGroup = type("PropertyGroup", (DummyBase,), {}) + mock_types.Operator = type("Operator", (DummyBase,), {}) + mock_types.Panel = type("Panel", (DummyBase,), {}) + mock_types.Menu = type("Menu", (DummyBase,), {}) + mock_types.Object = DummyObject + mock_types.Collection = DummyCollection + + mock_bpy = MagicMock() + mock_bpy.types = mock_types + sys.modules['bpy'] = mock_bpy + sys.modules['bpy.types'] = mock_types + sys.modules['bpy.props'] = MagicMock() + sys.modules['bpy.utils'] = MagicMock() + sys.modules['bpy.utils.previews'] = MagicMock() + sys.modules['bpy.app'] = MagicMock() + sys.modules['bpy.app.handlers'] = MagicMock() +else: + if 'bpy.props' not in sys.modules: + sys.modules['bpy.props'] = MagicMock() + +class DummyNodeSocket: + def __init__(self, name="", bl_idname="NodeSocketObject", node=None, value=None): + self.name = name + self.identifier = name + self.bl_idname = bl_idname + self.node = node + self.is_linked = False + self.links = [] + self.value = value + self.default_value = value + + def as_pointer(self): + return id(self) + +class DummyLink: + def __init__(self, from_socket, to_socket): + self.from_socket = from_socket + self.from_node = getattr(from_socket, "node", None) + self.to_socket = to_socket + self.to_node = getattr(to_socket, "node", None) + +class DummyNode: + def __init__(self, name="Node", bl_idname="DummyNode", id_data=None): + self.name = name + self.bl_idname = bl_idname + self.id_data = id_data + self.inputs = [] + self.outputs = [] + self.bl_rna = MagicMock() + self.bl_rna.properties = [] + + def as_pointer(self): + return id(self) + +class DummyGroupNode(DummyNode): + def __init__(self, name="GroupNode", slicer_tree=None, node_tree=None, id_data=None): + super().__init__(name, "SLICENODE_Group", id_data) + self.slicer_tree = slicer_tree + self.node_tree = node_tree + +class DummyExecutorNode(DummyNode): + def __init__(self, name="Executor", auto_refresh=True, id_data=None): + super().__init__(name, "SLICENODE_Executor", id_data) + self.auto_refresh = auto_refresh + self.evaluated = False + self.evaluate_count = 0 + + def evaluate(self, operator=None): + self.evaluated = True + self.evaluate_count += 1 + return "ok" + +class DummyNodeTree: + def __init__(self, name="Tree"): + self.name = name + self.bl_idname = "SLICETREE_NodeTree" + self.nodes = [] + + def as_pointer(self): + return id(self) + + +# Import tree functions +from NodeSlicer.core.nodeslicer.tree import ( + get_referenced_data, + get_referenced_objects, + get_referenced_trees, + run_auto_refresh, + request_auto_refresh, + SLICETREE_NodeTree +) +import NodeSlicer.core.nodeslicer.tree as tree_mod +import bpy + + +class TestAutoRefresh(unittest.TestCase): + def setUp(self): + tree_mod._is_updating = False + tree_mod._tree_changed = False + tree_mod._changed_trees.clear() + tree_mod._timer_scheduled = False + + def tearDown(self): + tree_mod._is_updating = False + tree_mod._tree_changed = False + tree_mod._changed_trees.clear() + tree_mod._timer_scheduled = False + + def test_referenced_objects_direct_and_upstream(self): + obj1 = DummyObject("Obj1") + obj2 = DummyObject("Obj2") + + node_a = DummyNode("NodeA") + out_sock_a = DummyNodeSocket("Out", "NodeSocketObject", node=node_a, value=obj1) + node_a.outputs.append(out_sock_a) + + node_b = DummyNode("NodeB") + in_sock_b = DummyNodeSocket("In", "NodeSocketObject", node=node_b) + node_b.inputs.append(in_sock_b) + + # Link A to B + link = DummyLink(out_sock_a, in_sock_b) + in_sock_b.is_linked = True + in_sock_b.links.append(link) + + # Direct unlinked socket on B with obj2 + in_sock_b2 = DummyNodeSocket("In2", "SLICESOCK_Object", node=node_b, value=obj2) + node_b.inputs.append(in_sock_b2) + + ref_objs = get_referenced_objects(node_b) + self.assertIn(obj1, ref_objs) + self.assertIn(obj2, ref_objs) + + def test_referenced_objects_with_hierarchy(self): + parent_empty = DummyObject("ParentRig") + child_mesh = DummyObject("ChildDetail") + main_obj = DummyObject("MainMesh", parent=parent_empty, children=[child_mesh]) + parent_empty.children = [main_obj] + child_mesh.parent = main_obj + + node = DummyNode("MeshNode") + sock = DummyNodeSocket("Obj", "NodeSocketObject", node=node, value=main_obj) + node.inputs.append(sock) + + ref_objs = get_referenced_objects(node) + self.assertIn(main_obj, ref_objs) + self.assertIn(parent_empty, ref_objs) + self.assertIn(child_mesh, ref_objs) + + def test_referenced_objects_in_node_groups(self): + obj_inner = DummyObject("InnerSuzanne") + obj_outer = DummyObject("OuterCube") + + # Group tree + group_tree = DummyNodeTree("GroupTree") + inner_node = DummyNode("InnerNode") + inner_sock = DummyNodeSocket("Obj", "NodeSocketObject", node=inner_node, value=obj_inner) + inner_node.inputs.append(inner_sock) + group_tree.nodes.append(inner_node) + + # Group Node instance + group_node = DummyGroupNode("GroupInstance", slicer_tree=group_tree) + outer_sock = DummyNodeSocket("OuterIn", "NodeSocketObject", node=group_node, value=obj_outer) + group_node.inputs.append(outer_sock) + + # Main node connected to group node + main_node = DummyNode("MainNode") + main_in = DummyNodeSocket("In", "SLICESOCK_Mesh", node=main_node) + group_out = DummyNodeSocket("Out", "SLICESOCK_Mesh", node=group_node) + group_node.outputs.append(group_out) + + link = DummyLink(group_out, main_in) + main_in.is_linked = True + main_in.links.append(link) + main_node.inputs.append(main_in) + + ref_objs = get_referenced_objects(main_node) + self.assertIn(obj_inner, ref_objs) + self.assertIn(obj_outer, ref_objs) + + def test_referenced_objects_nested_groups(self): + obj_deep = DummyObject("DeepObj") + + # Inner-most group + deep_tree = DummyNodeTree("DeepTree") + deep_node = DummyNode("DeepNode") + deep_node.inputs.append(DummyNodeSocket("DeepSock", value=obj_deep)) + deep_tree.nodes.append(deep_node) + + # Middle group containing inner group + mid_tree = DummyNodeTree("MidTree") + mid_group = DummyGroupNode("MidGroupNode", node_tree=deep_tree) + mid_tree.nodes.append(mid_group) + + # Top group node + top_group = DummyGroupNode("TopGroupNode", slicer_tree=mid_tree) + + ref_objs = get_referenced_objects(top_group) + self.assertIn(obj_deep, ref_objs) + + def test_get_referenced_trees(self): + # Two group trees: GroupA and GroupB + tree_b = DummyNodeTree("TreeB") + node_b = DummyGroupNode("NodeB", slicer_tree=tree_b) + + tree_a = DummyNodeTree("TreeA") + node_a = DummyGroupNode("NodeA", slicer_tree=tree_a) + sock_a = DummyNodeSocket("A_Out", node=node_a) + node_a.outputs.append(sock_a) + + executor = DummyExecutorNode("Executor") + exec_in = DummyNodeSocket("In", node=executor) + exec_in.is_linked = True + exec_in.links.append(DummyLink(sock_a, exec_in)) + executor.inputs.append(exec_in) + + trees = get_referenced_trees(executor) + self.assertIn(tree_a, trees) + self.assertNotIn(tree_b, trees) + + def test_collection_objects_referenced(self): + obj_c1 = DummyObject("C1") + obj_c2 = DummyObject("C2") + col = DummyCollection("TestCollection", [obj_c1, obj_c2]) + + col_node = DummyNode("ColNode", bl_idname="SLICENODE_CollectionObjects") + col_node.collection = col + + # Output sockets named C1, C2 + out_sock1 = DummyNodeSocket("C1", node=col_node) + out_sock2 = DummyNodeSocket("C2", node=col_node) + col_node.outputs.extend([out_sock1, out_sock2]) + + ref_objs, ref_cols = get_referenced_data(col_node) + self.assertIn(col, ref_cols) + self.assertIn(obj_c1, ref_objs) + self.assertIn(obj_c2, ref_objs) + + def test_run_auto_refresh_on_object_change(self): + main_tree = DummyNodeTree("MainTree") + executor = DummyExecutorNode("Executor", auto_refresh=True, id_data=main_tree) + target_obj = DummyObject("TargetCube") + + in_sock = DummyNodeSocket("MeshIn", node=executor, value=target_obj) + executor.inputs.append(in_sock) + main_tree.nodes.append(executor) + + # Mock bpy.data.node_groups + old_groups = getattr(tree_mod.bpy.data, "node_groups", None) + tree_mod.bpy.data.node_groups = [main_tree] + bpy.data.node_groups = [main_tree] + try: + # 1. Unrelated object update -> should NOT evaluate + run_auto_refresh(updated_object_names={"UnrelatedObject"}) + self.assertFalse(executor.evaluated) + + # 2. Related object update -> SHOULD evaluate + run_auto_refresh(updated_object_names={"TargetCube"}) + self.assertTrue(executor.evaluated) + self.assertEqual(executor.evaluate_count, 1) + + # 3. auto_refresh = False -> should NOT evaluate + executor.evaluated = False + executor.auto_refresh = False + run_auto_refresh(updated_object_names={"TargetCube"}) + self.assertFalse(executor.evaluated) + finally: + tree_mod.bpy.data.node_groups = old_groups + bpy.data.node_groups = old_groups + + def test_run_auto_refresh_on_group_tree_change(self): + group_tree = DummyNodeTree("GroupTree") + main_tree = DummyNodeTree("MainTree") + unrelated_tree = DummyNodeTree("UnrelatedTree") + + group_node = DummyGroupNode("GroupInstance", slicer_tree=group_tree, id_data=main_tree) + g_out = DummyNodeSocket("Out", node=group_node) + group_node.outputs.append(g_out) + + executor = DummyExecutorNode("Executor", auto_refresh=True, id_data=main_tree) + e_in = DummyNodeSocket("In", node=executor) + e_in.is_linked = True + e_in.links.append(DummyLink(g_out, e_in)) + executor.inputs.append(e_in) + main_tree.nodes.extend([group_node, executor]) + + old_groups = getattr(tree_mod.bpy.data, "node_groups", None) + tree_mod.bpy.data.node_groups = [main_tree, group_tree, unrelated_tree] + bpy.data.node_groups = [main_tree, group_tree, unrelated_tree] + try: + # 1. Change to unrelated tree -> should NOT evaluate + run_auto_refresh(changed_trees={unrelated_tree}, force_refresh=True) + self.assertFalse(executor.evaluated) + + # 2. Change to group_tree -> SHOULD evaluate + run_auto_refresh(changed_trees={group_tree}, force_refresh=True) + self.assertTrue(executor.evaluated) + self.assertEqual(executor.evaluate_count, 1) + + # 3. Change to main_tree -> SHOULD evaluate + executor.evaluated = False + run_auto_refresh(changed_trees={main_tree}, force_refresh=True) + self.assertTrue(executor.evaluated) + self.assertEqual(executor.evaluate_count, 2) + finally: + tree_mod.bpy.data.node_groups = old_groups + bpy.data.node_groups = old_groups + + def test_request_auto_refresh_guard(self): + # When _is_updating is True, request_auto_refresh must do nothing + tree_mod._is_updating = True + tree_mod._tree_changed = False + tree_mod._timer_scheduled = True + request_auto_refresh() + self.assertFalse(tree_mod._tree_changed) + + tree_mod._is_updating = False + request_auto_refresh() + self.assertTrue(tree_mod._tree_changed) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_collection_objects.py b/tests/test_collection_objects.py index cd7e6dd..2dac461 100644 --- a/tests/test_collection_objects.py +++ b/tests/test_collection_objects.py @@ -164,6 +164,8 @@ class TestCollectionObjectsSorting(unittest.TestCase): def test_evaluate_output(self): obj_a = DummyObject("Apple", 'MESH') bpy.data.objects = {"Apple": obj_a} + from NodeSlicer.core.nodeslicer.nodes import inputs as inp_module + inp_module.bpy.data.objects = {"Apple": obj_a} collection = DummyCollection("Fruits", [obj_a]) self.node.collection = collection self.node._update_outputs() diff --git a/tests/test_node_groups.py b/tests/test_node_groups.py index db72f1e..21d56ee 100644 --- a/tests/test_node_groups.py +++ b/tests/test_node_groups.py @@ -56,16 +56,24 @@ mock_types.NodeGroupInput = DummyNodeGroupInput mock_types.NodeGroupOutput = DummyNodeGroupOutput mock_types.NodeCustomGroup = DummyGroupNode -mock_bpy = MagicMock() -mock_bpy.types = mock_types - -sys.modules['bpy'] = mock_bpy -sys.modules['bpy.types'] = mock_types -sys.modules['bpy.props'] = MagicMock() -sys.modules['bpy.utils'] = MagicMock() -sys.modules['bpy.utils.previews'] = MagicMock() -sys.modules['bpy.app'] = MagicMock() -sys.modules['bpy.app.handlers'] = MagicMock() +if 'bpy' not in sys.modules: + mock_bpy = MagicMock() + mock_bpy.types = mock_types + sys.modules['bpy'] = mock_bpy + sys.modules['bpy.types'] = mock_types + sys.modules['bpy.props'] = MagicMock() + sys.modules['bpy.utils'] = MagicMock() + sys.modules['bpy.utils.previews'] = MagicMock() + sys.modules['bpy.app'] = MagicMock() + sys.modules['bpy.app.handlers'] = MagicMock() +else: + mock_bpy = sys.modules['bpy'] + mock_bpy.types.NodeSocket = DummyNodeSocket + mock_bpy.types.Node = DummyNode + mock_bpy.types.NodeGroup = DummyNodeGroup + mock_bpy.types.NodeGroupInput = DummyNodeGroupInput + mock_bpy.types.NodeGroupOutput = DummyNodeGroupOutput + mock_bpy.types.NodeCustomGroup = DummyGroupNode from NodeSlicer.infra.blender.blender_slice_nodes_generic import ( _resolve_group_output,