# noqa PLC0302
from typing import List

import omni.graph.core as og
import omni.graph.core._unstable as ogu
import omni.graph.core.tests as ogts
import omni.usd
from pxr import Sdf

COMPOUND_SUBGRAPH_NODE_TYPE = "omni.graph.nodes.CompoundSubgraph"
DEFAULT_SUBGRAPH_NAME = "Subgraph"


# -----------------------------------------------------------------------------
class TestCompoundSubgraphs(ogts.OmniGraphTestCase):
    """Tests functionality using subgraph-style compound nodes"""

    _test_graph_path = "/World/TestGraph"

    # -------------------------------------------------------------------------
    async def setUp(self):
        await super().setUp()
        omni.kit.commands.set_logging_enabled(False)

    # -------------------------------------------------------------------------
    async def tearDown(self):
        omni.timeline.get_timeline_interface().stop()
        await super().tearDown()
        omni.kit.commands.set_logging_enabled(True)

    # -------------------------------------------------------------------------
    def _get_input_attributes(self, node: og.Node) -> List[og.Attribute]:
        return [
            attr
            for attr in node.get_attributes()
            if attr.get_port_type() == og.AttributePortType.ATTRIBUTE_PORT_TYPE_INPUT
        ]

    # -------------------------------------------------------------------------
    def _get_output_attributes(self, node: og.Node) -> List[og.Attribute]:
        return [
            attr
            for attr in node.get_attributes()
            if attr.get_port_type() == og.AttributePortType.ATTRIBUTE_PORT_TYPE_OUTPUT
        ]

    # -------------------------------------------------------------------------
    async def test_compound_subgraph_placeholder_node(self):
        """Validation For the placeholder node used for compound subgraph nodes"""

        controller = og.Controller(update_usd=True)
        keys = controller.Keys
        (_, (node,), _, _) = controller.edit(
            self._test_graph_path, {keys.CREATE_NODES: ("PlaceHolder", COMPOUND_SUBGRAPH_NODE_TYPE)}
        )

        self.assertTrue(node.is_valid())
        self.assertTrue(node.is_compound_node())

        def is_input_or_output(attr):
            return attr.get_port_type() in [
                og.AttributePortType.ATTRIBUTE_PORT_TYPE_INPUT,
                og.AttributePortType.ATTRIBUTE_PORT_TYPE_OUTPUT,
            ]

        self.assertEqual(0, len(list(filter(is_input_or_output, node.get_attributes()))))

        stage = omni.usd.get_context().get_stage()
        self.assertTrue(stage.GetPrimAtPath(f"{node.get_prim_path()}/{DEFAULT_SUBGRAPH_NAME}").IsValid())

    # ------------------------------------------------------------------------
    async def test_create_compound_subgraph_command(self):
        """Validates the create compound subgraph command, including undo and redo"""
        controller = og.Controller(update_usd=True)
        keys = controller.Keys
        (graph, _, _, _) = controller.edit(self._test_graph_path, {keys.CREATE_NODES: ("Add", "omni.graph.nodes.Add")})

        # create a compound subgraph with the default name
        stage = omni.usd.get_context().get_stage()

        # validate the node is created
        ogu.cmds.CreateCompoundSubgraph(graph=graph, node_name="Node1", subgraph_name=DEFAULT_SUBGRAPH_NAME)
        node = graph.get_node(f"{self._test_graph_path}/Node1")
        subgraph_path = f"{node.get_prim_path()}/{DEFAULT_SUBGRAPH_NAME}"
        self.assertTrue(node.is_valid())
        self.assertTrue(stage.GetPrimAtPath(subgraph_path).IsValid())

        omni.kit.undo.undo()
        node = graph.get_node(f"{self._test_graph_path}/Node1")
        self.assertFalse(node.is_valid())
        self.assertFalse(stage.GetPrimAtPath(subgraph_path).IsValid())

        omni.kit.undo.redo()
        node = graph.get_node(f"{self._test_graph_path}/Node1")
        self.assertTrue(node.is_valid())
        self.assertTrue(stage.GetPrimAtPath(subgraph_path).IsValid())

        # validate with a different subgraph name
        subgraph2_name = f"{DEFAULT_SUBGRAPH_NAME}_02"
        ogu.cmds.CreateCompoundSubgraph(graph=graph, node_name="Node2", subgraph_name=subgraph2_name)
        node = graph.get_node(f"{self._test_graph_path}/Node2")
        subgraph_path = f"{node.get_prim_path()}/{subgraph2_name}"
        self.assertTrue(node.is_valid())
        self.assertTrue(stage.GetPrimAtPath(subgraph_path).IsValid())

    # ------------------------------------------------------------------------
    async def test_compound_subgraph_loaded_from_file(self):
        """Validates that a compound node with a subgraph can be loaded and evaluated"""
        (result, error) = await ogts.load_test_file("TestCompoundSubgraph.usda", use_caller_subdirectory=True)
        self.assertTrue(result, f"{error}")

        # The file contains an outer graph of
        # [const_double] -> [compound] -> [to_string]
        # Followed by an inner compound containing an add node

        graph = og.get_graph_by_path("/World/PushGraph")
        self.assertTrue(graph.is_valid())
        self.assertEquals(1, len(graph.get_subgraphs()))

        compound_node = graph.get_node("/World/PushGraph/compound")
        self.assertTrue(compound_node.is_valid())
        self.assertTrue(compound_node.is_compound_node())
        self.assertTrue(compound_node.get_node_type().is_valid())

        result_node = graph.get_node("/World/PushGraph/to_string")
        self.assertTrue(result_node.is_valid())
        output_attr = result_node.get_attribute("outputs:converted")
        self.assertTrue(output_attr.is_valid())

        await og.Controller.evaluate(graph)

        self.assertEqual("20", og.Controller.get(output_attr))

    # ------------------------------------------------------------------------
    async def test_replace_with_compound_subgraph(self):
        """Tests for replacing a portion of a graph with a subgraph"""

        compound_node_name = "Compound"
        controller = og.Controller(update_usd=True)
        keys = controller.Keys
        (graph, (_, _, _, add, mul, _), _, _) = controller.edit(
            self._test_graph_path,
            {
                keys.CREATE_NODES: [
                    ("ConstDouble1", "omni.graph.nodes.ConstantDouble"),
                    ("ConstDouble2", "omni.graph.nodes.ConstantDouble"),
                    ("ConstDouble3", "omni.graph.nodes.ConstantDouble"),
                    ("Add", "omni.graph.nodes.Add"),
                    ("Mul", "omni.graph.nodes.Multiply"),
                    ("ToDouble", "omni.graph.nodes.ToDouble"),
                ],
                keys.CONNECT: [
                    ("ConstDouble1.inputs:value", "Add.inputs:a"),
                    ("ConstDouble2.inputs:value", "Add.inputs:b"),
                    ("ConstDouble3.inputs:value", "Mul.inputs:b"),
                    ("Add.outputs:sum", "Mul.inputs:a"),
                    ("Mul.outputs:product", "ToDouble.inputs:value"),
                ],
                keys.SET_VALUES: [
                    ("ConstDouble1.inputs:value", 1.0),
                    ("ConstDouble2.inputs:value", 2.0),
                    ("ConstDouble3.inputs:value", 3.0),
                ],
            },
        )

        # evaluate the graph with a given constant value.
        async def validate_graph(c1_val):
            # update the constant and verify the graph is still functioning
            c1_attr = og.Controller.attribute(f"{self._test_graph_path}/ConstDouble1.inputs:value")
            og.Controller.set(c1_attr, c1_val, undoable=False)

            await og.Controller.evaluate(graph)
            res_attr = og.Controller.attribute(f"{self._test_graph_path}/ToDouble.outputs:converted")
            self.assertEqual((c1_val + 2.0) * 3.0, og.Controller.get(res_attr))

        await validate_graph(1.0)

        # the command will invalidate any og references to the graph
        ogu.cmds.ReplaceWithCompoundSubgraph(
            nodes=[add.get_prim_path(), mul.get_prim_path()], compound_name=compound_node_name
        )

        # validate the replacement generated the objects we are expecting
        graph = og.Controller.graph(self._test_graph_path)
        sub_graph_path = f"{self._test_graph_path}/{compound_node_name}/{DEFAULT_SUBGRAPH_NAME}"
        self.assertTrue(graph.is_valid())
        self.assertTrue(og.Controller.graph(sub_graph_path).is_valid())
        self.assertIsNone(og.get_node_by_path(f"{self._test_graph_path}/Add"))
        self.assertIsNone(og.get_node_by_path(f"{self._test_graph_path}/Mul"))
        self.assertTrue(og.Controller.node(f"{sub_graph_path}/Add").is_valid())
        self.assertTrue(og.Controller.node(f"{sub_graph_path}/Mul").is_valid())

        # update the constant and verify the graph is still functioning
        await validate_graph(2.0)

        # apply an undo
        omni.kit.undo.undo()

        graph = og.Controller.graph(self._test_graph_path)
        sub_graph_path = f"{self._test_graph_path}/{compound_node_name}/{DEFAULT_SUBGRAPH_NAME}"
        self.assertTrue(graph.is_valid())
        self.assertIsNone(og.get_graph_by_path(sub_graph_path))
        self.assertTrue(og.Controller.node(f"{self._test_graph_path}/Add").is_valid())
        self.assertTrue(og.Controller.node(f"{self._test_graph_path}/Mul").is_valid())
        self.assertIsNone(og.get_node_by_path(f"{sub_graph_path}/Add"))
        self.assertIsNone(og.get_node_by_path(f"{sub_graph_path}/Mul"))

        # update the constant and verify the graph is still functioning
        await validate_graph(3.0)

        # and a redo
        omni.kit.undo.redo()

        graph = og.Controller.graph(self._test_graph_path)
        sub_graph_path = f"{self._test_graph_path}/{compound_node_name}/{DEFAULT_SUBGRAPH_NAME}"
        self.assertTrue(graph.is_valid())
        self.assertTrue(og.Controller.graph(sub_graph_path).is_valid())
        self.assertIsNone(og.get_node_by_path(f"{self._test_graph_path}/Add"))
        self.assertIsNone(og.get_node_by_path(f"{self._test_graph_path}/Mul"))
        self.assertTrue(og.Controller.node(f"{sub_graph_path}/Add").is_valid())
        self.assertTrue(og.Controller.node(f"{sub_graph_path}/Mul").is_valid())

        # update the constant and verify the graph is still functioning
        await validate_graph(4.0)

    # ------------------------------------------------------------------------
    async def test_replace_with_compound_subgraph_input_resolved_types(self):
        """
        Tests which inputs are exposed in a generated compound graph
        When all the types are resolved
        """

        def _create_compound_graph(evaluator_name):

            controller = og.Controller(update_usd=True)
            keys = controller.Keys
            graph_path = f"{self._test_graph_path}_{evaluator_name}"

            # simple graph
            # [ConstDouble]->[AllTypes]->[AllTypes (via execution)]
            controller.edit({"graph_path": graph_path, "evaluator_name": evaluator_name})
            (_, (all1, all2, _), _, _) = controller.edit(
                graph_path,
                {
                    keys.CREATE_NODES: [
                        ("AllTypes1", "omni.graph.test.TestAllDataTypes"),
                        ("AllTypes2", "omni.graph.test.TestAllDataTypes"),
                        ("ConstDouble", "omni.graph.nodes.ConstantDouble"),
                    ],
                    keys.CONNECT: [
                        ("AllTypes1.outputs:a_execution", "AllTypes2.inputs:a_execution"),
                        ("ConstDouble.inputs:value", "AllTypes1.inputs:a_double"),
                    ],
                },
            )

            (success, node) = ogu.cmds.ReplaceWithCompoundSubgraph(nodes=[all1.get_prim_path(), all2.get_prim_path()])
            self.assertTrue(success)
            self.assertTrue(node.is_valid())

            return self._get_input_attributes(node)

        # for push, expect 1 exposed input, the constant double
        input_ports = _create_compound_graph("push")
        self.assertEquals(1, len(input_ports))
        self.assertEquals(og.Type(og.BaseDataType.DOUBLE), input_ports[0].get_resolved_type())

        # for action graph, the execution port should also be exposed
        input_ports = _create_compound_graph("execution")
        input_types = [input_port.get_resolved_type() for input_port in input_ports]
        self.assertEquals(2, len(input_ports))
        self.assertIn(og.Type(og.BaseDataType.DOUBLE), input_types)
        self.assertIn(og.Type(og.BaseDataType.UINT, 1, 0, og.AttributeRole.EXECUTION), input_types)

    # ------------------------------------------------------------------------
    async def test_replace_with_compound_subgraph_input_unresolved_types(self):
        """
        Tests the replace with compound subgraph exposes any unresolved types
        """

        # Partially connected graph of add nodes
        # [Add1]->[Add2]->[Add3]
        #  Add1 a,b are left unresolved
        #  Add2 a is connected, b is resolved
        #  Add3 a is connected, b is unresolved
        controller = og.Controller(update_usd=True)
        keys = controller.Keys
        (_, nodes, _, _) = controller.edit(
            self._test_graph_path,
            {
                keys.CREATE_NODES: [
                    ("Add1", "omni.graph.nodes.Add"),
                    ("Add2", "omni.graph.nodes.Add"),
                    ("Add3", "omni.graph.nodes.Add"),
                ],
                keys.CONNECT: [
                    ("Add1.outputs:sum", "Add2.inputs:a"),
                    ("Add2.outputs:sum", "Add3.inputs:a"),
                ],
            },
        )

        # manually resolve
        nodes[1].get_attribute("inputs:b").set_resolved_type(og.Type(og.BaseDataType.DOUBLE))

        # create the subgraph
        (success, node) = ogu.cmds.ReplaceWithCompoundSubgraph(nodes=[node.get_prim_path() for node in nodes])
        self.assertTrue(success)
        self.assertTrue(node.is_valid())

        inputs = self._get_input_attributes(node)
        outputs = self._get_output_attributes(node)
        self.assertEqual(3, len(inputs))
        self.assertEqual(1, len(outputs))

    # ------------------------------------------------------------------------
    async def test_replace_with_compound_subgraph_output_types(self):
        """
        Validates the behaviour of outputs when a compound replacement is made. If a node is a terminal node,
        it's outputs will be exposed
        """

        # create a graph with two nodes that contain a near complete subset of all types
        def _create_compound_graph(evaluator_name):

            controller = og.Controller(update_usd=True)
            keys = controller.Keys
            graph_path = f"{self._test_graph_path}_{evaluator_name}"

            # simple graph
            # [AllTypes]->[AllTypes] with one connection in between
            controller.edit({"graph_path": graph_path, "evaluator_name": evaluator_name})
            (_, (all1, all2), _, _) = controller.edit(
                graph_path,
                {
                    keys.CREATE_NODES: [
                        ("AllTypes1", "omni.graph.test.TestAllDataTypes"),
                        ("AllTypes2", "omni.graph.test.TestAllDataTypes"),
                    ],
                    keys.CONNECT: [("AllTypes1.outputs:a_double", "AllTypes2.inputs:a_double")],
                },
            )

            (success, node) = ogu.cmds.ReplaceWithCompoundSubgraph(nodes=[all1.get_prim_path(), all2.get_prim_path()])
            self.assertTrue(success)
            self.assertTrue(node.is_valid())

            output_types = [attr.get_resolved_type() for attr in self._get_output_attributes(node)]
            return output_types

        # in the case of a push graph, it will generate an output for each type, except execution
        output_types = _create_compound_graph("push")
        push_count = len(output_types)
        self.assertEqual(0, output_types.count(og.Type(og.BaseDataType.UINT, 1, 0, og.AttributeRole.EXECUTION)))
        self.assertEqual(1, output_types.count(og.Type(og.BaseDataType.DOUBLE)))
        self.assertEqual(1, output_types.count(og.Type(og.BaseDataType.INT)))

        # in the case of a push graph, it will generate an output for each type, except execution
        output_types = _create_compound_graph("execution")
        exec_count = len(output_types)
        self.assertEqual(1, output_types.count(og.Type(og.BaseDataType.UINT, 1, 0, og.AttributeRole.EXECUTION)))
        self.assertEqual(1, output_types.count(og.Type(og.BaseDataType.DOUBLE)))
        self.assertEqual(1, output_types.count(og.Type(og.BaseDataType.INT)))
        self.assertGreater(exec_count, push_count)

    # ------------------------------------------------------------------------
    async def test_replace_with_compound_subgraph_fan_in_out(self):
        """Validates replace with compound subgraph works with fan in/out"""

        controller = og.Controller(update_usd=True)
        keys = controller.Keys

        # Graph with Fan-In and Fan-Out
        # [OnImpulseEvent]-> |Sequence| -> [Sequence]
        # [OnImpulseEvent]-> |        | -> [Sequence]
        controller.edit({"graph_path": self._test_graph_path, "evaluator_name": "execution"})
        (_, (_, _, convert, _, _), _, _) = controller.edit(
            self._test_graph_path,
            {
                keys.CREATE_NODES: [
                    ("Input1", "omni.graph.action.OnImpulseEvent"),
                    ("Input2", "omni.graph.action.OnImpulseEvent"),
                    ("Convert", "omni.graph.action.Sequence"),
                    ("Output1", "omni.graph.action.Sequence"),
                    ("Output2", "omni.graph.action.Sequence"),
                ],
                keys.CONNECT: [
                    ("Input1.outputs:execOut", "Convert.inputs:execIn"),
                    ("Input2.outputs:execOut", "Convert.inputs:execIn"),
                    ("Convert.outputs:a", "Output1.inputs:execIn"),
                    ("Convert.outputs:a", "Output2.inputs:execIn"),
                ],
            },
        )

        (success, node) = ogu.cmds.ReplaceWithCompoundSubgraph(nodes=[convert.get_prim_path()])
        self.assertTrue(success)
        self.assertTrue(node.is_valid())
        inputs = self._get_input_attributes(node)
        outputs = self._get_output_attributes(node)
        self.assertEquals(1, len(inputs))
        self.assertEquals(1, len(outputs))
        self.assertEquals(2, inputs[0].get_upstream_connection_count())
        self.assertEquals(2, outputs[0].get_downstream_connection_count())

    # ------------------------------------------------------------------------
    async def test_replace_with_compound_subgraph_with_constant_nodes(self):
        """
        Validates replace with compound subgraph works with constant nodes,
        that use output only inputs
        """
        controller = og.Controller(update_usd=True)
        keys = controller.Keys
        # Create a graph with a connected constant and an unconnected constant
        # Converting the constants to a subgraph should create outputs for both
        # but keep the connection in tact
        (_, (c1, c2, _), _, _) = controller.edit(
            self._test_graph_path,
            {
                keys.CREATE_NODES: [
                    ("Const1", "omni.graph.nodes.ConstantDouble"),
                    ("Const2", "omni.graph.nodes.ConstantInt"),
                    ("Add", "omni.graph.nodes.Add"),
                ],
                keys.CONNECT: [
                    ("Const1.inputs:value", "Add.inputs:a"),
                ],
            },
        )

        (success, node) = ogu.cmds.ReplaceWithCompoundSubgraph(nodes=[c1.get_prim_path(), c2.get_prim_path()])
        self.assertTrue(success)
        self.assertTrue(node.is_valid())
        outputs = self._get_output_attributes(node)
        self.assertEquals(2, len(outputs))
        self.assertEquals(1, outputs[0].get_downstream_connection_count())
        self.assertEquals(0, outputs[1].get_downstream_connection_count())

    # ------------------------------------------------------------------------
    async def test_replace_with_compound_subgraph_catches_common_errors(self):
        """
        Validates ReplaceWithCompoundSubgraph handles typical error conditions
        """
        controller = og.Controller(update_usd=True)
        keys = controller.Keys

        (graph1, (add1,), _, _) = controller.edit(
            self._test_graph_path, {keys.CREATE_NODES: [("Add", "omni.graph.nodes.Add")]}
        )

        (_, (add2,), _, _) = controller.edit(
            f"{self._test_graph_path}_2", {keys.CREATE_NODES: [("Add2", "omni.graph.nodes.Add")]}
        )

        with ogts.ExpectedError():
            # no nodes means no compound is returned
            (success, node) = ogu.cmds.ReplaceWithCompoundSubgraph(nodes=[])
        self.assertFalse(success)
        self.assertIsNone(node)

        # not all nodes from the same graph
        with ogts.ExpectedError():
            (success, node) = ogu.cmds.ReplaceWithCompoundSubgraph(nodes=[add1.get_prim_path(), add2.get_prim_path()])
        self.assertFalse(success)
        self.assertIsNone(node)

        # invalid nodes
        with ogts.ExpectedError():
            (success, node) = ogu.cmds.ReplaceWithCompoundSubgraph(
                nodes=[add1.get_prim_path(), graph1.get_path_to_graph()]
            )
        self.assertFalse(success)
        self.assertIsNone(node)

    # ------------------------------------------------------------------------
    async def test_replace_with_compound_subgraph_with_relationship_inputs(self):
        """Validates ReplacewithCompoundSubgraph handles relationship inputs (bundles and targets)"""
        controller = og.Controller(update_usd=True)
        keys = controller.Keys

        # Graph with various types relationship inputs
        #   [BundleConstructor] -> [BundleInspector]
        #                          [BundleInspector (with target set)]
        #                          [BundleInspector (with target unset)]
        #   [GetGraphTargetPrim] -> [GetPrimPaths]
        #                           [GetPrimPaths (with target set)]
        #                           [GetPrimPaths (with target unset)]
        (_graph, _, _, _) = controller.edit(
            self._test_graph_path,
            {
                # create nodes that have bundle inputs and target inputs
                keys.CREATE_NODES: [
                    ("BundleCreator", "omni.graph.nodes.BundleConstructor"),
                    ("ConnectedInspector", "omni.graph.nodes.BundleInspector"),
                    ("SetInspector", "omni.graph.nodes.BundleInspector"),
                    ("UnsetInspector", "omni.graph.nodes.BundleInspector"),
                    ("GraphTargetPrim", "omni.graph.nodes.GetGraphTargetPrim"),
                    ("ConnectedTarget", "omni.graph.nodes.GetPrimPaths"),
                    ("SetTarget", "omni.graph.nodes.GetPrimPaths"),
                    ("UnsetTarget", "omni.graph.nodes.GetPrimPaths"),
                ],
                keys.CONNECT: [
                    ("BundleCreator.outputs_bundle", "ConnectedInspector.inputs:bundle"),
                    ("GraphTargetPrim.outputs:prim", "ConnectedTarget.inputs:prims"),
                ],
                keys.CREATE_PRIMS: [("Prim", "Cube")],
                keys.SET_VALUES: [
                    ("SetTarget.inputs:prims", f"{self._test_graph_path}/Prim"),
                ],
            },
        )

        # set the bundle manually
        stage = omni.usd.get_context().get_stage()
        stage.GetRelationshipAtPath(f"{self._test_graph_path}/SetInspector.inputs:bundle").SetTargets(
            [f"{self._test_graph_path}/Prim"]
        )

        # Replace the bundle inspector with a compound subgraph. Use names since a conversion can make
        # previous node handles invalid
        nodes_to_replace = [
            "ConnectedInspector",
            "SetInspector",
            "UnsetInspector",
            "ConnectedTarget",
            "SetTarget",
            "UnsetTarget",
        ]
        compounds = []
        for node_name in nodes_to_replace:
            node = og.Controller.node(f"{self._test_graph_path}/{node_name}")
            (success, compound_node) = ogu.cmds.ReplaceWithCompoundSubgraph(nodes=[node])
            self.assertTrue(success)
            self.assertTrue(compound_node.is_valid())
            compounds.append(compound_node.get_prim_path())

        def validate_result(compound_node_path, node_name, attr_name, expected_target):
            # make sure the inner node was created
            node = og.Controller.node(f"{compound_node_path}/Subgraph/{node_name}")
            self.assertTrue(node.is_valid())
            # make sure the relationship has the expected value
            rel = stage.GetRelationshipAtPath(f"{node.get_prim_path()}.{attr_name}")
            self.assertTrue(rel.IsValid())
            self.assertEquals(expected_target, rel.GetTargets())

        # validate the inner node have the expected connection
        validate_result(compounds[0], nodes_to_replace[0], "inputs:bundle", [f"{compounds[0]}.inputs:bundle"])
        validate_result(compounds[1], nodes_to_replace[1], "inputs:bundle", [f"{self._test_graph_path}/Prim"])
        validate_result(compounds[2], nodes_to_replace[2], "inputs:bundle", [])
        validate_result(compounds[3], nodes_to_replace[3], "inputs:prims", [f"{compounds[3]}.inputs:prims"])
        validate_result(compounds[4], nodes_to_replace[4], "inputs:prims", [f"{self._test_graph_path}/Prim"])
        validate_result(compounds[5], nodes_to_replace[5], "inputs:prims", [])

        def get_input_count(node_path: str):
            node = og.Controller.node(node_path)
            self.assertTrue(node.is_valid())
            return len(
                [
                    attr
                    for attr in node.get_attributes()
                    if attr.get_port_type() == og.AttributePortType.ATTRIBUTE_PORT_TYPE_INPUT
                ]
            )

        # validate that the new compounds have the correct number of inputs. They will have an input if they were connected
        # via a node connection
        self.assertEqual(get_input_count(compounds[0]), 1)
        self.assertEqual(get_input_count(compounds[1]), 0)
        self.assertEqual(get_input_count(compounds[2]), 0)
        self.assertEqual(get_input_count(compounds[3]), 1)
        self.assertEqual(get_input_count(compounds[4]), 0)
        self.assertEqual(get_input_count(compounds[5]), 0)

        # validate the inputs have the correct target values
        self.assertEquals(
            stage.GetRelationshipAtPath(f"{compounds[0]}.inputs:bundle").GetTargets(),
            [f"{self._test_graph_path}/BundleCreator/outputs_bundle"],
        )
        self.assertEquals(
            stage.GetRelationshipAtPath(f"{compounds[3]}.inputs:prims").GetTargets(),
            [f"{self._test_graph_path}/GraphTargetPrim.outputs:prim"],
        )

    # ------------------------------------------------------------------------
    async def test_promoted_bundle_inputs_retain_values(self):
        """Test that when a bundle input is promoted from a compound, it retains the value that was set for it"""

        controller = og.Controller(update_usd=True)
        keys = controller.Keys

        (_graph, (compound,), (prim,), node_map) = controller.edit(
            self._test_graph_path,
            {
                keys.CREATE_PRIMS: [("/World/Prim", "Cube")],
                keys.CREATE_NODES: [
                    (
                        "Compound",
                        {
                            keys.CREATE_NODES: ("Inspector", "omni.graph.nodes.BundleInspector"),
                        },
                    )
                ],
            },
        )

        # Set the bundle input on the compound node
        path = f"{node_map['Inspector'].get_prim_path()}.inputs:bundle"
        print(path)
        rel = omni.usd.get_context().get_stage().GetRelationshipAtPath(path)
        self.assertTrue(rel.IsValid())
        rel.SetTargets([prim.GetPath()])

        # Promote the input
        controller.edit(self._test_graph_path, {keys.PROMOTE_ATTRIBUTES: ("Inspector.inputs:bundle", "inputs:bundle")})

        # get the bundle relationship on the promoted node
        rel = omni.usd.get_context().get_stage().GetRelationshipAtPath(f"{compound.get_prim_path()}.inputs:bundle")
        self.assertTrue(rel.IsValid())
        self.assertEquals(str(rel.GetTargets()[0]), str(prim.GetPath()))

    # ------------------------------------------------------------------------
    async def test_replace_with_compound_subgraph_with_relationship_outputs(self):
        """Tests that ReplaceWithCompoundSubgraph works with relationship outputs produces errors as
        is is currently not supported.
        """

        #
        # [BundleConstructor]->[BundleInspector]

        controller = og.Controller(update_usd=True)
        keys = controller.Keys

        (_graph, (bundle_constructor, _bundle_inspector), _, _) = controller.edit(
            self._test_graph_path,
            {
                keys.CREATE_NODES: [
                    ("Constructor", "omni.graph.nodes.BundleConstructor"),
                    ("Inspector", "omni.graph.nodes.BundleInspector"),
                ],
                keys.CONNECT: [
                    ("Constructor.outputs_bundle", "Inspector.inputs:bundle"),
                ],
            },
        )

        # Attempt to replace the bundle constructor with a compound subgraph.
        with ogts.ExpectedError():
            (success, compound_node) = ogu.cmds.ReplaceWithCompoundSubgraph(nodes=[bundle_constructor])
            self.assertFalse(success)
            self.assertIsNone(compound_node)

    # ------------------------------------------------------------------------
    async def test_replace_with_subgraph_handles_target_types(self):
        """Tests that ReplaceWithCompoundSubgraph works with target inputs and outputs"""

        # Create a graph with the following setup
        #               |--------------------|   |-------------------|
        # [GetPrimPath]-|o-[GetPrimsAtPath]-o|-->|o-[GetPrimsPaths]-o|->[GetPrimsAtPath]
        #               |--------------------|   |-------------------|
        controller = og.Controller(update_usd=True)
        keys = controller.Keys

        (graph, nodes, (prim, prim1, prim2), _) = controller.edit(
            self._test_graph_path,
            {
                keys.CREATE_NODES: [
                    ("P0", "omni.graph.nodes.GetPrimPath"),
                    ("P1", "omni.graph.nodes.GetPrimsAtPath"),
                    ("P2", "omni.graph.nodes.GetPrimPaths"),
                    ("P3", "omni.graph.nodes.GetPrimsAtPath"),
                ],
                keys.CONNECT: [
                    ("P0.outputs:primPath", "P1.inputs:path"),
                    ("P1.outputs:prims", "P2.inputs:prims"),
                    ("P2.outputs:primPaths", "P3.inputs:path"),
                ],
                keys.CREATE_PRIMS: [
                    ("/World/Prim", "Cube"),
                    ("/World/Prim1", "Cube"),
                    ("/World/Prim2", "Cube"),
                ],
                keys.SET_VALUES: [
                    ("P0.inputs:prim", "/World/Prim"),
                ],
            },
        )

        # save the node paths, since the nodes can become invalid on a replace call
        node_paths = [og.Controller.node_path(node) for node in nodes]
        await og.Controller.evaluate(graph)

        attr = og.Controller.attribute("outputs:prims", nodes[3])
        self.assertEqual(str(prim.GetPath()), str(og.Controller.get(attr)[0]))

        # replace node with string input, path output
        p1 = og.Controller.node(node_paths[1])
        (success, c1) = ogu.cmds.ReplaceWithCompoundSubgraph(nodes=[p1])
        self.assertTrue(success)
        self.assertTrue(c1.is_valid())
        self.assertGreater(og.Controller.attribute(("inputs:path", c1)).get_upstream_connection_count(), 0)
        self.assertGreater(og.Controller.attribute(("inputs:path", c1)).get_downstream_connection_count(), 0)
        self.assertGreater(og.Controller.attribute(("outputs:prims", c1)).get_upstream_connection_count(), 0)
        self.assertGreater(og.Controller.attribute(("outputs:prims", c1)).get_downstream_connection_count(), 0)

        # evaluate with a new input and makes sure the output through the compounds is correct
        og.Controller.set(og.Controller.attribute(f"{node_paths[0]}/inputs:prim"), prim1.GetPath())
        await og.Controller.evaluate(graph)
        p3 = og.Controller.node(node_paths[3])
        attr = og.Controller.attribute("outputs:prims", p3)
        self.assertEqual(str(prim1.GetPath()), str(og.Controller.get(attr)[0]))

        # replace node with path input, path string output
        p2 = og.Controller.node(node_paths[2])
        (success, c2) = ogu.cmds.ReplaceWithCompoundSubgraph(nodes=[p2])
        self.assertTrue(success)
        self.assertTrue(c2.is_valid())
        self.assertGreater(og.Controller.attribute(("inputs:prims", c2)).get_upstream_connection_count(), 0)
        self.assertGreater(og.Controller.attribute(("inputs:prims", c2)).get_downstream_connection_count(), 0)
        self.assertGreater(og.Controller.attribute(("outputs:primPaths", c2)).get_upstream_connection_count(), 0)
        self.assertGreater(og.Controller.attribute(("outputs:primPaths", c2)).get_downstream_connection_count(), 0)

        # evaluate with a new input and makes sure the output through the compounds is correct
        og.Controller.set(og.Controller.attribute(f"{node_paths[0]}/inputs:prim"), prim2.GetPath())
        await og.Controller.evaluate(graph)
        p3 = og.Controller.node(node_paths[3])
        attr = og.Controller.attribute("outputs:prims", p3)
        self.assertEqual(str(prim2.GetPath()), str(og.Controller.get(attr)[0]))

    # ------------------------------------------------------------------------
    async def test_replace_with_subgraph_in_nested_compounds_types(self):
        """Tests that replace with compound subgraph works with nested compounds"""
        controller = og.Controller(update_usd=True)
        keys = controller.Keys
        (graph, (_,), _, nodes) = controller.edit(
            self._test_graph_path,
            {
                keys.CREATE_NODES: [
                    (
                        "OuterCompound",
                        {
                            keys.CREATE_NODES: [
                                ("Add", "omni.graph.nodes.Add"),
                            ],
                            keys.PROMOTE_ATTRIBUTES: [
                                ("Add.inputs:a", "inputs:a"),
                                ("Add.inputs:b", "inputs:b"),
                                ("Add.outputs:sum", "outputs:sum"),
                            ],
                        },
                    )
                ],
            },
        )

        graph_path = graph.get_path_to_graph()
        (success, c1) = ogu.cmds.ReplaceWithCompoundSubgraph(nodes=[nodes["Add"]])
        self.assertTrue(success)
        self.assertTrue(c1.is_valid())
        compound_graph = c1.get_compound_graph_instance()
        self.assertTrue(compound_graph.is_valid())
        self.assertTrue(compound_graph.get_node(f"{compound_graph.get_path_to_graph()}/Add").is_valid())
        outer_compound = og.Controller.node(f"{graph_path}/OuterCompound")
        self.assertEqual(c1.get_graph().get_owning_compound_node(), outer_compound)

        # create an extra level of compounds by replacing the new compound with a compound
        # [OuterCompound] -> [C1] -> [Add] becomes
        # [OuterCompound] -> [C2] -> [C1] -> [Add]
        (success, c2) = ogu.cmds.ReplaceWithCompoundSubgraph(nodes=[c1])
        self.assertTrue(success)
        self.assertTrue(c2.is_valid())
        outer_compound = og.Controller.node(f"{graph_path}/OuterCompound")
        self.assertEquals(c2.get_graph().get_owning_compound_node(), outer_compound)
        compound_graph = c2.get_compound_graph_instance()
        sub_graph_node = compound_graph.get_nodes()[0]
        self.assertTrue(sub_graph_node.is_compound_node())
        self.assertTrue(sub_graph_node.get_compound_graph_instance().is_valid())
        add_node = sub_graph_node.get_compound_graph_instance().get_nodes()[0]
        self.assertTrue(add_node.is_valid())
        self.assertEqual(Sdf.Path(add_node.get_prim_path()).name, "Add")

    # ---------------------------------------------------------------------
    async def test_subgraph_with_constants_evaluates(self):
        """Tests that a subgraph with no inputs evaluates correctly in a push graph setup"""

        # Create a subgraph/graph pair with the following setup
        #
        # |--------------------------------------|
        # | [Constant]----->|----------|         |
        # | [Constant]----->| Add      |-----> o | ---> [Magnitude]
        # |                 |----------|         |
        # | -------------------------------------|

        controller = og.Controller(update_usd=True)
        keys = controller.Keys
        (graph, (constant1, constant2, add, abs_node), _, _) = controller.edit(
            self._test_graph_path,
            {
                keys.CREATE_NODES: [
                    ("Constant1", "omni.graph.nodes.ConstantDouble"),
                    ("Constant2", "omni.graph.nodes.ConstantDouble"),
                    ("Add", "omni.graph.nodes.Add"),
                    ("Abs", "omni.graph.nodes.Magnitude"),
                ],
                keys.CONNECT: [
                    ("Constant1.inputs:value", "Add.inputs:a"),
                    ("Constant2.inputs:value", "Add.inputs:b"),
                    ("Add.outputs:sum", "Abs.inputs:input"),
                ],
                keys.SET_VALUES: [
                    ("Constant1.inputs:value", 5.0),
                    ("Constant2.inputs:value", 1.0),
                ],
            },
        )

        await og.Controller.evaluate(graph)
        attr = og.Controller.attribute(("outputs:magnitude", abs_node))
        self.assertEqual(6.0, og.Controller.get(attr))

        # push the variable nodes into a subgraph
        (_, compound_node) = ogu.cmds.ReplaceWithCompoundSubgraph(nodes=[constant1, constant2, add])
        self.assertIsNotNone(compound_node)
        self.assertTrue(compound_node.is_valid())

        # update the constant so we get different result
        constant = og.Controller.node(f"{compound_node.get_prim_path()}/Subgraph/Constant2")
        constant_attr = og.Controller.attribute(("inputs:value", constant))
        og.Controller.set(constant_attr, 2.0)

        await og.Controller.evaluate(graph)
        abs_node = og.Controller.node(f"{self._test_graph_path}/Abs")
        attr = og.Controller.attribute(("outputs:magnitude", abs_node))
        self.assertEqual(7.0, og.Controller.get(attr))

    # ---------------------------------------------------------------------
    async def test_attribute_promotion_supports_fan_in(self):
        """Tests than an attribute that supports fan-in can be promoted, even if connected or already promoted"""

        # Creates a graph where an execution pin is both connected internally and promoted twice
        # |---------------------------------------------------------------|
        # |                    --------------                             |
        # | o----------------->|o execIn    |                             |
        # | o----------------->|            |                             |
        # |                  /\--------------                             |
        # |   -----------     |                                           |
        # |   |execOut o|-----|                                           |
        # |   -----------                                                 |
        # |                                                               |
        # |---------------------------------------------------------------|

        controller = og.Controller(update_usd=True)
        keys = controller.Keys
        og.Controller.edit({"graph_path": self._test_graph_path, "evaluator_name": "execution"})
        (_, _, _, node_map) = controller.edit(
            self._test_graph_path,
            {
                keys.CREATE_NODES: [
                    (
                        "Compound",
                        {
                            keys.CREATE_NODES: [
                                ("NodeA", "omni.graph.action.Counter"),
                                ("NodeB", "omni.graph.action.Counter"),
                            ],
                            keys.CONNECT: [
                                ("NodeB.outputs:execOut", "NodeA.inputs:execIn"),
                            ],
                            keys.PROMOTE_ATTRIBUTES: [
                                ("NodeA.inputs:execIn", "execIn1"),
                                ("NodeA.inputs:execIn", "execIn2"),
                            ],
                        },
                    )
                ]
            },
        )

        compound_node = node_map["Compound"]
        self.assertEqual(len(self._get_input_attributes(compound_node)), 2)
        node_a = node_map["NodeA"]
        self.assertEqual(len(node_a.get_attribute("inputs:execIn").get_upstream_connections()), 3)

    # ---------------------------------------------------------------------
    async def test_attribute_promotion_rejects_fan_in(self):
        """Test that validates that promotion of non-fan-in attributes is rejected"""

        controller = og.Controller(update_usd=True)
        keys = controller.Keys
        with self.assertRaises(og.OmniGraphError):
            (_, _, _, _) = controller.edit(
                self._test_graph_path,
                {
                    keys.CREATE_NODES: [
                        (
                            "Compound",
                            {
                                keys.CREATE_NODES: [
                                    ("NodeA", "omni.graph.nodes.Add"),
                                ],
                                keys.PROMOTE_ATTRIBUTES: [
                                    ("NodeA.inputs:a", "a1"),
                                    ("NodeA.inputs:a", "a2"),
                                ],
                            },
                        )
                    ]
                },
            )

    # ---------------------------------------------------------------------
    async def test_compound_graph_exec_ports_support_fan_in(self):
        """Test that validates compounds can support fan in on exec ports"""

        controller = og.Controller(update_usd=True)
        keys = controller.Keys

        #             -------------     |----------------------------|
        # [OnTick]--->|Multigate o|---->|o-->[Counter]------------->o|
        #             |          o|---->|                            |
        #             |          o|     |                            |
        #             -------------     |----------------------------|
        og.Controller.edit({"graph_path": self._test_graph_path, "evaluator_name": "execution"})
        (_, (compound, mg, _), _, _) = controller.edit(
            self._test_graph_path,
            {
                keys.CREATE_NODES: [
                    (
                        "Compound",
                        {
                            keys.CREATE_NODES: [
                                ("Counter", "omni.graph.action.Counter"),
                            ],
                            keys.PROMOTE_ATTRIBUTES: [
                                ("Counter.inputs:execIn", "inputs:execIn"),
                                ("Counter.outputs:count", "outputs:count"),
                            ],
                        },
                    ),
                    ("Multigate", "omni.graph.action.Multigate"),
                    ("Tick", "omni.graph.action.OnTick"),
                ],
                keys.CREATE_ATTRIBUTES: [
                    ("Multigate.outputs:output1", og.Type(og.BaseDataType.UINT, 1, 0, og.AttributeRole.EXECUTION)),
                    ("Multigate.outputs:output2", og.Type(og.BaseDataType.UINT, 1, 0, og.AttributeRole.EXECUTION)),
                ],
                keys.CONNECT: [
                    ("Tick.outputs:tick", "Multigate.inputs:execIn"),
                    ("Multigate.outputs:output0", "Compound.inputs:execIn"),
                    ("Multigate.outputs:output1", "Compound.inputs:execIn"),
                ],
            },
        )

        omni.timeline.get_timeline_interface().play()

        # evaluate the graph a bunch of times. The counter should update 2 of every 3 times
        counter = compound.get_attribute("outputs:count")
        for _ in range(0, 8):
            await omni.kit.app.get_app().next_update_async()

        expected_counter = int(mg.get_compute_count() / 3) * 2 + (mg.get_compute_count() % 3)
        self.assertGreater(og.Controller.get(counter), 0)
        self.assertEqual(expected_counter, og.Controller.get(counter))

    # ---------------------------------------------------------------------
    async def test_rename_compound_subgraph(self):
        """Validates that a compound subgraph can be successfully renamed"""

        controller = og.Controller(update_usd=True)
        keys = controller.Keys
        (graph, _, _, node_map) = controller.edit(
            self._test_graph_path,
            {
                keys.CREATE_NODES: [
                    (
                        "Compound",
                        {
                            keys.CREATE_NODES: [
                                (
                                    "Compound2",
                                    {
                                        keys.CREATE_NODES: [
                                            ("Add", "omni.graph.nodes.Add"),
                                        ],
                                        keys.PROMOTE_ATTRIBUTES: [
                                            ("Add.inputs:a", "inputs:a"),
                                            ("Add.inputs:b", "inputs:b"),
                                            ("Add.outputs:sum", "outputs:sum"),
                                        ],
                                    },
                                ),
                            ],
                            keys.PROMOTE_ATTRIBUTES: [
                                ("Compound2.inputs:a", "inputs:a"),
                                ("Compound2.inputs:b", "inputs:b"),
                                ("Compound2.outputs:sum", "outputs:sum"),
                            ],
                        },
                    ),
                    ("Constant1", "omni.graph.nodes.ConstantDouble"),
                    ("Constant2", "omni.graph.nodes.ConstantDouble"),
                    ("Output", "omni.graph.nodes.Magnitude"),
                ],
                keys.CONNECT: [
                    ("Constant1.inputs:value", "Compound.inputs:a"),
                    ("Constant2.inputs:value", "Compound.inputs:b"),
                    ("Compound.outputs:sum", "Output.inputs:input"),
                ],
                keys.SET_VALUES: [
                    ("Constant1.inputs:value", 1.0),
                    ("Constant2.inputs:value", 2.0),
                ],
            },
        )

        # iterate the paths from deepest to shallowest
        paths = [
            node_map["Compound2"].get_prim_path(),
            node_map["Compound"].get_prim_path(),
        ]
        const_2_path = node_map["Constant2"].get_attribute("inputs:value").get_path()
        output_attr = node_map["Output"].get_attribute("outputs:magnitude").get_path()

        input_value = 2.0  # noqa SIM113
        expected_result = 3.0
        await og.Controller.evaluate(graph)
        self.assertEquals(expected_result, og.Controller.get(og.Controller.attribute(output_attr)))

        for path in paths:
            # rename the compound node specified by path
            compound = og.Controller.node(path).get_compound_graph_instance()
            old_name = compound.get_path_to_graph()
            new_name = old_name + "_renamed"
            omni.kit.commands.execute("RenameCompoundSubgraph", subgraph=compound, new_path=new_name)

            # increment the expected result
            input_value += 1.0
            og.Controller.set(og.Controller.attribute(const_2_path), input_value)
            expected_result = input_value + 1.0
            await og.Controller.evaluate(graph)

            # reload the compound node. the rename will have invalidated the node
            self.assertEquals(expected_result, og.Controller.get(og.Controller.attribute(output_attr)))

    # ---------------------------------------------------------------------
    async def tests_created_compounds_are_ineligible_for_delete(self):
        """Tests that compound subgraphs are flagged as ineligible for deletion"""
        controller = og.Controller(update_usd=True)
        keys = controller.Keys
        (_, (compound,), _, _) = controller.edit(
            self._test_graph_path,
            {keys.CREATE_NODES: [("Compound", {keys.CREATE_NODES: [("Add", "omni.graph.nodes.Add")]})]},
        )

        subgraph_path = compound.get_compound_graph_instance().get_path_to_graph()

        # suppress the warning. The command will succeed regardless, but print out a warning
        with ogts.ExpectedError():
            omni.kit.commands.execute("DeletePrims", paths=[subgraph_path])

        # validate that the prim still exists
        stage = omni.usd.get_context().get_stage()
        prim = stage.GetPrimAtPath(subgraph_path)
        self.assertTrue(prim.IsValid())
        self.assertTrue(compound.get_compound_graph_instance().is_valid())

    # ---------------------------------------------------------------------
    async def tests_broken_compounds_produces_error(self):
        """Tests that a broken compound will produce error msgs when evaluated"""
        controller = og.Controller(update_usd=True)
        keys = controller.Keys
        (graph, _, _, node_map) = controller.edit(
            self._test_graph_path,
            {
                keys.CREATE_NODES: [
                    (
                        "Compound",
                        {
                            keys.CREATE_NODES: [("Add", "omni.graph.nodes.Add")],
                            keys.PROMOTE_ATTRIBUTES: [
                                ("Add.inputs:a", "inputs:a"),
                                ("Add.inputs:b", "inputs:b"),
                                ("Add.outputs:sum", "outputs:sum"),
                            ],
                        },
                    ),
                    ("Constant1", "omni.graph.nodes.ConstantDouble"),
                ],
                keys.SET_VALUES: [
                    ("Constant1.inputs:value", 1.0),
                ],
                keys.CONNECT: [
                    ("Constant1.inputs:value", "Compound.inputs:a"),
                    ("Constant1.inputs:value", "Compound.inputs:b"),
                ],
            },
        )

        compound_path = node_map["Compound"].get_prim_path()
        subgraph_path = node_map["Compound"].get_compound_graph_instance().get_path_to_graph()
        stage = omni.usd.get_context().get_stage()
        prim = stage.GetPrimAtPath(subgraph_path)
        self.assertTrue(prim.IsValid())
        prim.SetMetadata("no_delete", False)
        omni.kit.commands.execute("DeletePrims", paths=[subgraph_path])

        # suppress the error that is expected
        with ogts.ExpectedError():
            await og.Controller.evaluate(graph)

        node = og.Controller.node(compound_path)
        self.assertTrue(node.is_valid())
        msgs = node.get_compute_messages(og.Severity.ERROR)
        self.assertTrue(len(msgs) > 0)

    # -----------------------------------------------------------------------------
    async def tests_rename_compound_command(self):
        """Tests the rename compound command"""
        controller = og.Controller(update_usd=True)
        keys = controller.Keys
        (_, _, _, node_map) = controller.edit(
            self._test_graph_path,
            {
                keys.CREATE_NODES: [
                    ("Compound1", {keys.CREATE_NODES: [("Add1", "omni.graph.nodes.Add")]}),
                    ("Compound2", {keys.CREATE_NODES: [("Add2", "omni.graph.nodes.Add")]}),
                    ("Compound3", {keys.CREATE_NODES: [("Add3", "omni.graph.nodes.Add")]}),
                    ("Compound4", {keys.CREATE_NODES: [("Add4", "omni.graph.nodes.Add")]}),
                    ("Compound5", {keys.CREATE_NODES: [("Add5", "omni.graph.nodes.Add")]}),
                ]
            },
        )

        graph_paths = [
            og.Controller.prim_path(node_map[f"Compound{idx}"].get_compound_graph_instance()) for idx in range(1, 6)
        ]

        # success case with undo
        og._unstable.cmds.RenameCompoundSubgraph(  # noqa: PLW0212
            subgraph=graph_paths[0], new_path=graph_paths[0] + "_renamed"
        )
        await omni.kit.app.get_app().next_update_async()
        self.assertTrue(og.Controller.graph(graph_paths[0] + "_renamed"))

        omni.kit.undo.undo()
        await omni.kit.app.get_app().next_update_async()
        self.assertTrue(og.Controller.graph(graph_paths[0]))

        omni.kit.undo.redo()
        await omni.kit.app.get_app().next_update_async()
        self.assertTrue(og.Controller.graph(graph_paths[0] + "_renamed"))

        # failure case - bad path
        with ogts.ExpectedError():
            (res, _) = og._unstable.cmds.RenameCompoundSubgraph(  # noqa: PLW0212
                subgraph=graph_paths[1], new_path="This is an invalid path"
            )
            self.assertFalse(res)

        # failure case - rename to an invalid name in immediate mode
        with ogts.ExpectedError():
            self.assertIsNone(
                og._unstable.cmds.imm.RenameCompoundSubgraph(  # noqa: PLW0212
                    subgraph=graph_paths[2], new_path="invalid name"
                )
            )

        # pass by graph
        og._unstable.cmds.RenameCompoundSubgraph(  # noqa: PLW0212
            subgraph=og.Controller.graph(graph_paths[3]), new_path=graph_paths[3] + "_renamed"
        )
        await omni.kit.app.get_app().next_update_async()
        self.assertTrue(og.Controller.graph(graph_paths[3] + "_renamed"))

        # failure case - path is not a child of the compound node
        with ogts.ExpectedError():
            self.assertIsNone(
                og._unstable.cmds.imm.RenameCompoundSubgraph(graph_paths[4], "/World/OtherPath")  # noqa: PLW0212
            )


# -----------------------------------------------------------------------------
class TestCompoundSubgraphAttributeCommands(ogts.OmniGraphTestCase):
    """Tests related to commands to add, remove and rename attributes from compound nodes
    containing a subgraph
    """

    _test_graph_path = "/World/TestGraph"

    # -------------------------------------------------------------------------
    async def setUp(self):
        await super().setUp()

        # Set up a subgraph node as follows
        #   The composedouble3 node as fixed-type inputs, so they aren't exposed
        #                |-------------------------------------|
        # [Constant] --> | o(a) -->|------|                    |
        #                |         | Add1 |                    |
        # [Constant] --> | o(b) -->|------| --> |------|       |
        #                |                      | Add2 | --> o |-->[Add3]
        #                | o(b_01) -----------> |------|       |
        #                |                                     |
        #                |            |----------------|       |
        #                |            | ComposeDouble3 | --> o |
        #                |            |----------------|       |
        #                |-------------------------------------|
        #
        # The inputs are a, b, b_01
        # The outputs are sum and double3

        controller = og.Controller(update_usd=True)
        keys = controller.Keys

        (
            _,
            (self._constant_node_1, self._constant_node_2, self._compound_node, self._add3_node),
            _,
            mapping,
        ) = controller.edit(
            self._test_graph_path,
            {
                keys.CREATE_NODES: [
                    ("Constant1", "omni.graph.nodes.ConstantDouble"),
                    ("Constant2", "omni.graph.nodes.ConstantDouble"),
                    (
                        "Compound",
                        {
                            keys.CREATE_NODES: [
                                ("Add1", "omni.graph.nodes.Add"),
                                ("Add2", "omni.graph.nodes.Add"),
                                ("ComposeDouble3", "omni.graph.test.ComposeDouble3C"),
                            ],
                            keys.CONNECT: [
                                ("Add1.outputs:sum", "Add2.inputs:a"),
                            ],
                            keys.SET_VALUES: [
                                ("ComposeDouble3.inputs:x", 5.0),
                            ],
                            keys.PROMOTE_ATTRIBUTES: [
                                ("Add1.inputs:a", "a"),
                                ("Add1.inputs:b", "b"),
                                ("Add2.inputs:b", "b_01"),
                                ("Add2.outputs:sum", "sum"),
                                ("ComposeDouble3.outputs:double3", "double3"),
                            ],
                        },
                    ),
                    ("Add3", "omni.graph.nodes.Add"),
                ],
                keys.CONNECT: [
                    ("Constant1.inputs:value", "Compound.inputs:a"),
                    ("Constant2.inputs:value", "Compound.inputs:b"),
                    ("Compound.outputs:sum", "Add3.inputs:a"),
                ],
                keys.SET_VALUES: [
                    ("Constant1.inputs:value", 5.0),
                ],
            },
        )

        self._add1_node = mapping["Add1"]
        self._add2_node = mapping["Add2"]
        self._compose_node = mapping["ComposeDouble3"]

        self._compound_input_1 = og.ObjectLookup.attribute(("inputs:a", self._compound_node))
        self._compound_input_2 = og.ObjectLookup.attribute(("inputs:b", self._compound_node))
        self._compound_input_3 = og.ObjectLookup.attribute(("inputs:b_01", self._compound_node))
        self._compound_output_1 = og.ObjectLookup.attribute(("outputs:sum", self._compound_node))
        self._compound_output_2 = og.ObjectLookup.attribute(("outputs:double3", self._compound_node))
        self._subgraph = og.ObjectLookup.graph(f"{self._compound_node.get_prim_path()}/Subgraph")

        # disable logging of command errors
        omni.kit.commands.set_logging_enabled(False)

    # -------------------------------------------------------------------------
    async def tearDown(self):
        await super().tearDown()
        omni.kit.commands.set_logging_enabled(True)

    # -------------------------------------------------------------------------
    async def test_create_input_command(self):
        """Test that input connections can be successfully added and works with undo and redo"""

        connect_to_attr = og.ObjectLookup.attribute(("inputs:x", self._compose_node))
        (success, attr) = ogu.cmds.CreateCompoundSubgraphInput(
            compound_node=self._compound_node, input_name="added_input", connect_to=connect_to_attr
        )

        self.assertTrue(success)
        self.assertTrue(attr.is_valid())
        self.assertTrue(attr.get_downstream_connections()[0], connect_to_attr)
        # validate the value was copied up to the graph
        self.assertEqual(og.Controller.get(attr), 5.0)

        omni.kit.undo.undo()
        self.assertTrue(self._compound_node.is_valid())
        with self.assertRaises(og.OmniGraphError):
            og.ObjectLookup.attribute(("inputs:added_input", self._compound_node))
        self.assertEqual(og.Controller.get(connect_to_attr), 5.0)

        omni.kit.undo.redo()
        self.assertTrue(self._compound_node.is_valid())
        attr = og.ObjectLookup.attribute(("inputs:added_input", self._compound_node))
        self.assertTrue(attr.is_valid())
        self.assertTrue(attr.get_downstream_connections()[0], connect_to_attr)
        self.assertEqual(og.Controller.get(attr), 5.0)

    # ---------------------------------------------------------------------------
    async def test_resolved_attribute_values_are_copied_on_create(self):
        """Tests that resolved attribute values are copied when used with create input"""

        # create a new node in the graph
        new_node = og.GraphController.create_node(
            node_id=("Add4", self._subgraph), node_type_id="omni.graph.nodes.Add", update_usd=True
        )

        attr = og.ObjectLookup.attribute(("inputs:a", new_node))
        attr.set_resolved_type(og.Type(og.BaseDataType.DOUBLE))
        og.Controller.set(attr, 10.0, update_usd=True)
        self.assertEqual(og.Controller.get(attr), 10.0)

        (success, compound_attr) = ogu.cmds.CreateCompoundSubgraphInput(
            compound_node=self._compound_node, input_name="added_input", connect_to=attr
        )

        self.assertTrue(success)
        self.assertTrue(compound_attr.is_valid())
        self.assertEqual(10, og.Controller.get(compound_attr))

    # ---------------------------------------------------------------------------
    async def test_create_input_error_cases(self):
        """Validates known failure cases for create input commands"""

        # 1) connecting to an attribute not in the subgraph
        with ogts.ExpectedError():
            (success, attr) = ogu.cmds.CreateCompoundSubgraphInput(
                compound_node=self._compound_node,
                input_name="added_input",
                connect_to=og.ObjectLookup.attribute("inputs:b", self._add3_node),
            )
        self.assertTrue(success)
        self.assertIsNone(attr)

        # 2) connecting to no/invalid attribute
        with self.assertRaises(og.OmniGraphError):
            (success, attr) = ogu.cmds.CreateCompoundSubgraphInput(
                compound_node=self._compound_node, input_name="added_input", connect_to=None
            )
        with ogts.ExpectedError():
            (success, attr) = ogu.cmds.CreateCompoundSubgraphInput(
                compound_node=self._compound_node, input_name="added_input", connect_to="/World/Path.inputs:not_valid"
            )
        self.assertTrue(success)
        self.assertIsNone(attr)

        # 3) connecting to an existing attribute name
        with ogts.ExpectedError():
            connect_to_attr = og.ObjectLookup.attribute(("inputs:x", self._compose_node))
            (success, attr) = ogu.cmds.CreateCompoundSubgraphInput(
                compound_node=self._compound_node, input_name="a", connect_to=connect_to_attr
            )
        self.assertTrue(success)
        self.assertIsNone(attr)

        # 4) connecting to a non-compound node
        with ogts.ExpectedError():
            (success, attr) = ogu.cmds.CreateCompoundSubgraphInput(
                compound_node=self._add3_node, input_name="other", connect_to=connect_to_attr
            )
        self.assertTrue(success)
        self.assertIsNone(attr)

        # 5) connecting to a connected attribute
        with ogts.ExpectedError():
            (success, attr) = ogu.cmds.CreateCompoundSubgraphInput(
                compound_node=self._compound_node,
                input_name="other",
                connect_to=og.ObjectLookup.attribute(("inputs:a", self._add1_node)),
            )
        self.assertTrue(success)
        self.assertIsNone(attr)

        # 6) connecting to output attribute
        with ogts.ExpectedError():
            (success, attr) = ogu.cmds.CreateCompoundSubgraphInput(
                compound_node=self._compound_node,
                input_name="other",
                connect_to=og.ObjectLookup.attribute(("outputs:sum", self._add1_node)),
            )
        self.assertTrue(success)
        self.assertIsNone(attr)

    # -------------------------------------------------------------------------
    def _validate_create_output_command_with_undo_redo(self, connect_from_attr: og.Attribute):
        """Helper to validate create output"""
        (success, attr) = ogu.cmds.CreateCompoundSubgraphOutput(
            compound_node=self._compound_node, output_name="added_output", connect_from=connect_from_attr
        )

        self.assertTrue(success)
        self.assertTrue(attr.is_valid())
        self.assertTrue(attr.get_upstream_connections()[0], connect_from_attr)

        omni.kit.undo.undo()
        self.assertTrue(self._compound_node.is_valid())
        with self.assertRaises(og.OmniGraphError):
            og.ObjectLookup.attribute(("outputs:added_output", self._compound_node))

        omni.kit.undo.redo()
        self.assertTrue(self._compound_node.is_valid())
        attr = og.ObjectLookup.attribute(("outputs:added_output", self._compound_node))
        self.assertTrue(attr.is_valid())
        self.assertTrue(attr.get_upstream_connections()[0], connect_from_attr)

    # -------------------------------------------------------------------------
    async def test_create_output_command(self):
        """Tests that output connections can be successfully added, including undo, redo"""

        # add a node to the subgraph, so there is an unconnected output
        new_node = og.GraphController.create_node(
            node_id=("Add4", self._subgraph),
            node_type_id="omni.graph.nodes.Add",
        )
        connect_from_attr = og.ObjectLookup.attribute(("outputs:sum", new_node))
        self._validate_create_output_command_with_undo_redo(connect_from_attr)

    # -------------------------------------------------------------------------
    async def test_create_output_command_multiple_outputs(self):
        """Tests that the output command connectes to an already connected output"""
        connect_from_attr = og.ObjectLookup.attribute(("outputs:sum", self._add2_node))
        self._validate_create_output_command_with_undo_redo(connect_from_attr)

    # -------------------------------------------------------------------------
    async def test_create_output_command_from_input(self):
        """Tests that the output command can create outputs from inputs as output nodes"""
        new_node = og.GraphController.create_node(
            node_id=("NewConstant", self._subgraph),
            node_type_id="omni.graph.nodes.ConstantDouble",
        )
        connect_from_attr = og.ObjectLookup.attribute(("inputs:value", new_node))
        self._validate_create_output_command_with_undo_redo(connect_from_attr)

    # -------------------------------------------------------------------------
    async def test_create_output_command_error_cases(self):
        """Tests command error cases of output commands"""

        # 1) connecting to an attribute not in the subgraph
        with ogts.ExpectedError():
            (success, attr) = ogu.cmds.CreateCompoundSubgraphOutput(
                compound_node=self._compound_node,
                output_name="added_output",
                connect_from=og.ObjectLookup.attribute("outputs:sum", self._add3_node),
            )
        self.assertTrue(success)
        self.assertIsNone(attr)

        # 2) connecting to no/invalid attribute
        with self.assertRaises(og.OmniGraphError):
            (success, attr) = ogu.cmds.CreateCompoundSubgraphOutput(
                compound_node=self._compound_node, output_name="added_output", connect_from=None
            )
        with ogts.ExpectedError():
            (success, attr) = ogu.cmds.CreateCompoundSubgraphOutput(
                compound_node=self._compound_node,
                output_name="added_output",
                connect_from="/World/Path.outputs:not_valid",
            )
        self.assertTrue(success)
        self.assertIsNone(attr)

        # 3) connecting to an existing attribute name
        with ogts.ExpectedError():
            connect_from_attr = og.ObjectLookup.attribute(("outputs:double3", self._compose_node))
            (success, attr) = ogu.cmds.CreateCompoundSubgraphOutput(
                compound_node=self._compound_node, output_name="sum", connect_from=connect_from_attr
            )
        self.assertTrue(success)
        self.assertIsNone(attr)

        # 4) connecting to a non-compound node
        with ogts.ExpectedError():
            (success, attr) = ogu.cmds.CreateCompoundSubgraphOutput(
                compound_node=self._add3_node, output_name="other", connect_from=connect_from_attr
            )
        self.assertTrue(success)
        self.assertIsNone(attr)

        # 5) try to promote a bundle output
        og.Controller.edit(
            self._compound_node.get_compound_graph_instance(),
            {og.Controller.Keys.CREATE_NODES: ("BundleCreator", "omni.graph.nodes.BundleConstructor")},
        )
        node = og.Controller.node(
            f"{self._compound_node.get_compound_graph_instance().get_path_to_graph()}/BundleCreator"
        )
        self.assertTrue(node.is_valid())
        bundle_attr = node.get_attribute("outputs_bundle")
        self.assertTrue(bundle_attr.is_valid())
        with ogts.ExpectedError():
            (success, attr) = ogu.cmds.CreateCompoundSubgraphOutput(
                compound_node=self._compound_node, output_name="exposed_bundle", connect_from=bundle_attr
            )
        self.assertTrue(success)
        self.assertIsNone(attr)

    # -------------------------------------------------------------------------
    async def test_remove_compound_input(self):
        """Tests the remove compound attribute command can remove an input, including undo and redo"""

        self._compound_input_1.set_resolved_type(og.Type(og.BaseDataType.DOUBLE))
        og.Controller.set(self._compound_input_1, 10.0, update_usd=True)

        connected_attr = self._compound_input_1.get_downstream_connections()[0]
        (success, result) = ogu.cmds.RemoveCompoundSubgraphAttribute(attribute=self._compound_input_1)

        self.assertTrue(success)
        self.assertTrue(result)
        self.assertFalse(self._compound_node.get_attribute_exists("inputs:a"))
        self.assertEqual(0, connected_attr.get_upstream_connection_count())

        omni.kit.undo.undo()
        self.assertTrue(self._compound_node.get_attribute_exists("inputs:a"))
        attr = og.ObjectLookup.attribute(("inputs:a", self._compound_node))
        self.assertGreater(attr.get_downstream_connection_count(), 0)
        self.assertEqual(attr.get_downstream_connections()[0], connected_attr)

        omni.kit.undo.redo()
        self.assertFalse(self._compound_node.get_attribute_exists("inputs:a"))
        self.assertEqual(0, connected_attr.get_upstream_connection_count())

    # -------------------------------------------------------------------------
    async def test_remove_compound_inputs_keeps_resolved_values(self):
        """Test the remove compound attribute restores resolved values"""

        # add the connection to the resolved node
        connect_to_attr = og.ObjectLookup.attribute(("inputs:x", self._compose_node))
        (success, attr) = ogu.cmds.CreateCompoundSubgraphInput(
            compound_node=self._compound_node, input_name="added_input", connect_to=connect_to_attr
        )

        self.assertTrue(success)
        self.assertTrue(attr.is_valid())

        (success, result) = ogu.cmds.RemoveCompoundSubgraphAttribute(attribute=attr)

        self.assertTrue(success)
        self.assertTrue(result)

        connect_to_attr = og.ObjectLookup.attribute(("inputs:x", self._compose_node))
        self.assertEqual(5.0, og.Controller.get(connect_to_attr))

    # -------------------------------------------------------------------------
    async def test_remove_compound_output(self):
        """Tests the remove compound attribute command can remove an output, including undo and redo"""
        connected_attr = self._compound_output_1.get_upstream_connections()[0]
        (success, result) = ogu.cmds.RemoveCompoundSubgraphAttribute(attribute=self._compound_output_1)

        self.assertTrue(success)
        self.assertTrue(result)
        self.assertFalse(self._compound_node.get_attribute_exists("outputs:sum"))
        self.assertEqual(0, connected_attr.get_downstream_connection_count())

        omni.kit.undo.undo()
        self.assertTrue(self._compound_node.get_attribute_exists("outputs:sum"))
        attr = og.ObjectLookup.attribute(("outputs:sum", self._compound_node))
        self.assertGreater(attr.get_upstream_connection_count(), 0)
        self.assertEqual(attr.get_upstream_connections()[0], connected_attr)

        omni.kit.undo.redo()
        self.assertFalse(self._compound_node.get_attribute_exists("outputs:sum"))
        self.assertEqual(0, connected_attr.get_downstream_connection_count())

    # -------------------------------------------------------------------------
    async def test_remove_compound_output_error_cases(self):
        """Validates known failure cases for remove attribute commands"""

        # 1) not a compound node
        with ogts.ExpectedError():
            (success, result) = ogu.cmds.RemoveCompoundSubgraphAttribute(
                attribute=og.ObjectLookup.attribute("inputs:b", self._add3_node)
            )
        self.assertTrue(success)
        self.assertFalse(result)

        # 2) not a valid attribute
        with ogts.ExpectedError():
            (success, result) = ogu.cmds.RemoveCompoundSubgraphAttribute(attribute="/World/Path.outputs:invalid_attr")
        self.assertTrue(success)
        self.assertFalse(result)

    # -------------------------------------------------------------------------
    async def test_rename_compound_input_output(self):
        """Tests the rename input/output functionality"""

        # Rename an input attribute
        upstream = self._compound_input_1.get_upstream_connections()
        downstream = self._compound_input_1.get_downstream_connections()
        old_attr_path = self._compound_input_1.get_path()

        (success, new_attr) = ogu.cmds.RenameCompoundSubgraphAttribute(
            attribute=self._compound_input_1, new_name="new_a"
        )

        # Verify that connections are now on new attr and old attr has been removed
        self.assertTrue(success)
        self.assertTrue(new_attr.is_valid())
        self.assertEqual(new_attr.get_upstream_connections(), upstream)
        self.assertEqual(new_attr.get_downstream_connections(), downstream)
        with self.assertRaises(og.OmniGraphError):
            og.ObjectLookup.attribute(old_attr_path)

        new_attr_path = new_attr.get_path()
        omni.kit.undo.undo()

        # Verify undo restores the old attr and connections
        with self.assertRaises(og.OmniGraphError):
            og.ObjectLookup.attribute(new_attr_path)
        attr = og.ObjectLookup.attribute(old_attr_path)
        self.assertEqual(attr.get_upstream_connections(), upstream)
        self.assertEqual(attr.get_downstream_connections(), downstream)

        # Verify redo
        omni.kit.undo.redo()
        attr = og.ObjectLookup.attribute(new_attr_path)
        self.assertEqual(attr.get_upstream_connections(), upstream)
        self.assertEqual(attr.get_downstream_connections(), downstream)
        with self.assertRaises(og.OmniGraphError):
            og.ObjectLookup.attribute(old_attr_path)

        # Rename an output attribute
        upstream = self._compound_output_1.get_upstream_connections()
        downstream = self._compound_output_1.get_downstream_connections()
        old_attr_path = self._compound_output_1.get_path()

        (success, new_attr) = ogu.cmds.RenameCompoundSubgraphAttribute(
            attribute=self._compound_output_1, new_name="new_sum"
        )

        self.assertTrue(success)
        self.assertTrue(new_attr.is_valid())
        self.assertEqual(new_attr.get_upstream_connections(), upstream)
        self.assertEqual(new_attr.get_downstream_connections(), downstream)

        new_attr_path = new_attr.get_path()
        omni.kit.undo.undo()

        with self.assertRaises(og.OmniGraphError):
            og.ObjectLookup.attribute(new_attr_path)
        attr = og.ObjectLookup.attribute(old_attr_path)
        self.assertEqual(attr.get_upstream_connections(), upstream)
        self.assertEqual(attr.get_downstream_connections(), downstream)

        omni.kit.undo.redo()
        attr = og.ObjectLookup.attribute(new_attr_path)
        self.assertEqual(attr.get_upstream_connections(), upstream)
        self.assertEqual(attr.get_downstream_connections(), downstream)

    # -------------------------------------------------------------------------
    async def test_rename_compound_input_output_error_cases(self):
        """Validates known failure cases for remove attribute commands"""
        # attempt to rename to an existing name
        with ogts.ExpectedError():
            (success, new_attr) = ogu.cmds.RenameCompoundSubgraphAttribute(
                attribute=self._compound_input_1, new_name="b"
            )
            self.assertFalse(success)
            self.assertEqual(new_attr, None)


# -----------------------------------------------------------------------------
class TestCompoundSubgraphWithVariables(ogts.OmniGraphTestCase):

    _test_graph_path = "/World/TestGraph"

    # ---------------------------------------------------------------------
    async def test_subgraph_with_read_variables_evaluates(self):
        """Tests that a subgraph with variables in it evaluates correctly"""

        # Create a subgraph/graph pair with the following setup
        #
        # |--------------------------------------|
        # | [ReadVariable]->|----------|         |
        # | [Constant]----->| Add      |-----> o | ---> [Magnitude]
        # |                 |----------|         |
        # | -------------------------------------|
        controller = og.Controller(update_usd=True)
        keys = controller.Keys
        (graph, (compound_node, abs_node), _, _) = controller.edit(
            self._test_graph_path,
            {
                keys.CREATE_NODES: [
                    (
                        "Compound",
                        {
                            keys.CREATE_NODES: [
                                ("ReadVariable", "omni.graph.core.ReadVariable"),
                                ("Constant", "omni.graph.nodes.ConstantDouble"),
                                ("Add", "omni.graph.nodes.Add"),
                            ],
                            keys.CONNECT: [
                                ("ReadVariable.outputs:value", "Add.inputs:a"),
                                ("Constant.inputs:value", "Add.inputs:b"),
                            ],
                            keys.PROMOTE_ATTRIBUTES: [("Add.outputs:sum", "value")],
                        },
                    ),
                    ("Abs", "omni.graph.nodes.Magnitude"),
                ],
                keys.CONNECT: [
                    ("Compound.outputs:value", "Abs.inputs:input"),
                ],
                keys.CREATE_VARIABLES: [
                    ("double_var", og.Type(og.BaseDataType.DOUBLE), 5.0),
                ],
                keys.SET_VALUES: [("Constant.inputs:value", 1.0), ("ReadVariable.inputs:variableName", "double_var")],
            },
        )

        await og.Controller.evaluate(graph)
        attr = og.Controller.attribute(("outputs:magnitude", abs_node))
        self.assertEqual(6.0, og.Controller.get(attr))

        # update the constant so we get  different result - because it 'moved' to the subgraph, the
        # existing variable is no longer valid
        constant = og.Controller.node(f"{compound_node.get_prim_path()}/Subgraph/Constant")
        constant_attr = og.Controller.attribute(("inputs:value", constant))
        og.Controller.set(constant_attr, 2.0)

        await og.Controller.evaluate(graph)
        abs_node = og.Controller.node(f"{self._test_graph_path}/Abs")
        attr = og.Controller.attribute(("outputs:magnitude", abs_node))
        self.assertEqual(7.0, og.Controller.get(attr))

    # ---------------------------------------------------------------------
    async def test_subgraph_read_write_variables_across_subnet(self):
        """Tests that a subgraph can read and write variables regardless of
        where they live
        """

        # A graph that doubles the variable value each time
        # A single read is in the main graph, one in the subgraph
        #                    |----------------------------------------|
        #                    |                                        |
        # [ReadVariable] --> | o-------------->|------|               |
        #                    | [ReadVariable]->| Add  |--->[WriteVar] |
        #                    |                 |------|               |
        #                    |----------------------------------------|
        controller = og.Controller(update_usd=True)
        keys = controller.Keys

        (graph, (_read1, _compound_node), _, _) = controller.edit(
            self._test_graph_path,
            {
                keys.CREATE_NODES: [
                    ("ReadVariable1", "omni.graph.core.ReadVariable"),
                    (
                        "Compound",
                        {
                            keys.CREATE_NODES: [
                                ("ReadVariable2", "omni.graph.core.ReadVariable"),
                                ("Add", "omni.graph.nodes.Add"),
                                ("WriteVariable", "omni.graph.core.WriteVariable"),
                            ],
                            keys.CONNECT: [
                                ("ReadVariable2.outputs:value", "Add.inputs:b"),
                                ("Add.outputs:sum", "WriteVariable.inputs:value"),
                            ],
                            keys.PROMOTE_ATTRIBUTES: [
                                ("Add.inputs:a", "inputs:a"),
                                ("WriteVariable.outputs:value", "outputs:value"),
                            ],
                            keys.SET_VALUES: [
                                ("ReadVariable2.inputs:variableName", "double_var"),
                                ("WriteVariable.inputs:variableName", "double_var"),
                            ],
                        },
                    ),
                ],
                keys.CREATE_VARIABLES: [
                    ("double_var", og.Type(og.BaseDataType.DOUBLE), 2.0),
                ],
                keys.CONNECT: [
                    ("ReadVariable1.outputs:value", "Compound.inputs:a"),
                ],
                keys.SET_VALUES: [
                    ("ReadVariable1.inputs:variableName", "double_var"),
                ],
            },
        )

        # get the variable value
        variable = graph.find_variable("double_var")
        var_value = variable.get(graph.get_default_graph_context())
        for _ in range(0, 3):
            await og.Controller.evaluate(graph)
            new_var_value = variable.get(graph.get_default_graph_context())
            self.assertEqual(var_value * 2, new_var_value)
            var_value = new_var_value

    async def test_promote_unconnected(self):
        """
        Test promoting all unconnected inputs/outputs on a node in a compound. Ensure that it also creates a unique name
        """
        controller = og.Controller(update_usd=True)
        keys = controller.Keys

        (graph, (_, _compound_node), _, _) = controller.edit(
            self._test_graph_path,
            {
                keys.CREATE_NODES: [
                    ("ConstDouble", "omni.graph.nodes.ConstantDouble"),
                    (
                        "Compound",
                        {
                            keys.CREATE_NODES: [
                                ("Add", "omni.graph.nodes.Add"),
                                ("ConstantFloat", "omni.graph.nodes.ConstantFloat"),
                                ("RandomBool", "omni.graph.nodes.RandomBoolean"),
                            ],
                            keys.PROMOTE_ATTRIBUTES: [
                                ("Add.inputs:a", "inputs:a"),
                            ],
                        },
                    ),
                ],
                keys.CONNECT: [
                    ("ConstDouble.inputs:value", "Compound.inputs:a"),
                ],
                keys.SET_VALUES: [("ConstDouble.inputs:value", 2.0)],
            },
        )

        await og.Controller.evaluate(graph)
        self.assertTrue(_compound_node.get_attribute("inputs:a").is_valid())
        self.assertFalse(_compound_node.get_attribute_exists("inputs:b"))
        add_node = og.Controller.node(f"{_compound_node.get_prim_path()}/Subgraph/Add")
        (success, ports) = ogu.cmds.PromoteUnconnectedToCompoundSubgraph(node=add_node, inputs=True, outputs=False)
        self.assertTrue(success)
        self.assertEqual(len(ports), 1)
        self.assertTrue(ports[0].is_valid())
        self.assertEqual(ports[0].get_name(), "inputs:b")
        self.assertTrue(_compound_node.get_attribute("inputs:b").is_valid())

        omni.kit.undo.undo()
        self.assertFalse(_compound_node.get_attribute_exists("inputs:b"))

        omni.kit.undo.redo()
        self.assertTrue(_compound_node.get_attribute("inputs:b").is_valid())

        # rename input "a" to "b" and promote unconnected again, should create b_01
        omni.kit.undo.undo()
        self.assertFalse(_compound_node.get_attribute_exists("inputs:b"))
        attr = _compound_node.get_attribute("inputs:a")
        (success, new_attr) = ogu.cmds.RenameCompoundSubgraphAttribute(attribute=attr, new_name="b")
        self.assertTrue(success)
        self.assertTrue(new_attr.is_valid())
        (success, ports) = ogu.cmds.PromoteUnconnectedToCompoundSubgraph(node=add_node, inputs=True, outputs=False)
        self.assertTrue(success)
        self.assertEqual(len(ports), 1)
        self.assertTrue(ports[0].is_valid())
        self.assertEqual(ports[0].get_name(), "inputs:b_01")

        omni.kit.undo.undo()
        c = og.Controller.create_attribute(add_node, "inputs:c", "double")
        self.assertTrue(c.is_valid())
        (success, ports) = ogu.cmds.PromoteUnconnectedToCompoundSubgraph(node=add_node, inputs=True, outputs=True)
        self.assertTrue(success)
        self.assertEqual(len(ports), 3)
        self.assertEqual(ports[0].get_name(), "inputs:b_01")
        self.assertEqual(ports[1].get_name(), "inputs:c")
        self.assertEqual(ports[2].get_name(), "outputs:sum")

        # verify that output-only inputs are promoted as outputs
        float_node = og.Controller.node(f"{_compound_node.get_prim_path()}/Subgraph/ConstantFloat")
        (success, ports) = ogu.cmds.PromoteUnconnectedToCompoundSubgraph(node=float_node, inputs=False, outputs=True)
        self.assertTrue(success)
        self.assertEqual(len(ports), 1)
        self.assertEqual(ports[0].get_name(), "outputs:value")

        # test that literal only inputs are not promoted
        rand_node = og.Controller.node(f"{_compound_node.get_prim_path()}/Subgraph/RandomBool")
        (success, ports) = ogu.cmds.PromoteUnconnectedToCompoundSubgraph(node=rand_node, inputs=True, outputs=False)
        self.assertTrue(success)
        self.assertEqual(len(ports), 3)
        self.assertFalse("inputs:isNoise" in {p.get_name() for p in ports})
