"""Tests related to the dynamic attribute Python interface generation feature"""
import carb
import omni.graph.core as og
import omni.graph.core.tests as ogts
import omni.kit
from omni.graph.tools._impl.node_generator.attributes.naming import CPP_KEYWORDS, PYTHON_KEYWORDS, SAFE_CPP_KEYWORDS
from pxr import UsdGeom


# ==============================================================================================================
class TestDynamicAttributes(ogts.OmniGraphTestCase):
    """Testing to ensure dynamic attributes works"""

    TEST_GRAPH_PATH = "/World/TestGraph"

    # --------------------------------------------------------------------------------------------------------------
    async def __test_dynamic_attributes(self, node_type: str):
        """Test basic creation and interaction with the dynamic attribute accessors, used in the test node compute"""
        controller = og.Controller()
        keys = og.Controller.Keys
        (test_graph, _, _, _) = controller.edit(self.TEST_GRAPH_PATH)

        async def test_helper(create_usd):
            node_path = f"{self.TEST_GRAPH_PATH}/TestNode{create_usd}"
            og.cmds.CreateNode(
                graph=test_graph,
                node_path=node_path,
                node_type=f"omni.graph.tutorials.{node_type}",
                create_usd=create_usd,
            )
            test_node = controller.node(node_path)
            self.assertTrue(test_node.is_valid())
            controller.edit(test_graph, {keys.SET_VALUES: (("inputs:value", test_node), 127)})
            output = controller.attribute("outputs:result", test_node)

            # Although this test only requires a single node it cannot be done via the .ogn test generator as it
            # needs to perform dynamic attribute manipulation between evaluations.

            # First test confirms the initial value is copied unchanged
            await controller.evaluate(test_graph)
            self.assertEqual(127, og.Controller.get(output))

            # Second test adds a bit manipulator and confirms the bit it defines is manipulated properly
            self.assertTrue(test_node.create_attribute("inputs:firstBit", og.Type(og.BaseDataType.UINT)))
            controller.edit(test_graph, {keys.SET_VALUES: (("inputs:firstBit", test_node), 3)})
            await controller.evaluate(test_graph)
            self.assertEqual(119, og.Controller.get(output))

            # Third test adds a second bit manipulator and confirms that both bits are manipulated properly
            self.assertTrue(test_node.create_attribute("inputs:secondBit", og.Type(og.BaseDataType.UINT)))
            controller.edit(test_graph, {keys.SET_VALUES: (("inputs:secondBit", test_node), 7)})
            await controller.evaluate(test_graph)
            self.assertEqual(247, og.Controller.get(output))

            # Fourth test removes the first bit manipulator and confirms the output adapts accordingly
            self.assertTrue(test_node.remove_attribute("inputs:firstBit"))
            await controller.evaluate(test_graph)
            self.assertEqual(255, og.Controller.get(output))

            # Fifth test alters the value of the modified bit and confirms the output adapts accordingly
            controller.edit(test_graph, {keys.SET_VALUES: (("inputs:secondBit", test_node), 4)})
            await controller.evaluate(test_graph)
            self.assertEqual(111, og.Controller.get(output))

            # Final test adds in the inversion attribute, with timecode role solely for the sake of checking role access
            self.assertTrue(
                test_node.create_attribute(
                    "inputs:invert", og.Type(og.BaseDataType.FLOAT, 1, 0, og.AttributeRole.TIMECODE)
                )
            )
            await controller.evaluate(test_graph)
            self.assertEqual(0xFFFFFFFF - 111, og.Controller.get(output))

        await test_helper(True)
        await test_helper(False)

    # --------------------------------------------------------------------------------------------------------------
    async def test_dynamic_attributes_py(self):
        """Test basic creation and interaction with the dynamic attribute accessors for the Python implementation"""
        await self.__test_dynamic_attributes("DynamicAttributesPy")

    # --------------------------------------------------------------------------------------------------------------
    async def test_dynamic_attributes_bundle(self):
        """Test basic creation and interaction with a bundle dynamic attribute"""
        controller = og.Controller()
        keys = og.Controller.Keys

        usd_context = omni.usd.get_context()
        stage = usd_context.get_stage()
        cube = ogts.create_cube(stage, "World/Cube", (1, 1, 1))
        UsdGeom.Xformable(cube).AddTranslateOp()

        (_, nodes, _, _) = controller.edit(
            "/TestGraph",
            {
                keys.CREATE_NODES: [
                    ("ReadPrimsBundle", "omni.graph.nodes.ReadPrimsBundle"),
                    ("ExtractPrim", "omni.graph.nodes.ExtractPrim"),
                    ("Script", "omni.graph.scriptnode.ScriptNode"),
                ],
                keys.SET_VALUES: [
                    ("ReadPrimsBundle.inputs:usePaths", True),
                    ("ReadPrimsBundle.inputs:primPaths", {"value": "/World/Cube", "type": "path"}),
                    ("ExtractPrim.inputs:primPath", "/World/Cube"),
                ],
                keys.CONNECT: [
                    ("ReadPrimsBundle.outputs_primsBundle", "ExtractPrim.inputs:prims"),
                ],
            },
        )

        # creates some attributes on the script node
        script = nodes[2]
        matrix_type = og.Type(og.BaseDataType.DOUBLE, 16, 0, og.AttributeRole.MATRIX)
        self.assertTrue(
            script.create_attribute(
                "inputs:bundle", og.Type(og.BaseDataType.RELATIONSHIP, 1, 0, og.AttributeRole.BUNDLE)
            )
        )
        self.assertTrue(
            script.create_attribute("outputs:matrix", matrix_type, og.AttributePortType.ATTRIBUTE_PORT_TYPE_OUTPUT)
        )

        # connect them
        controller.edit("/TestGraph", {keys.CONNECT: ("ExtractPrim.outputs_primBundle", "Script.inputs:bundle")})

        # set a script that extract the matrix
        s = 'db.outputs.matrix = db.inputs.bundle.get_attribute_by_name("worldMatrix").get()'
        controller.edit("/TestGraph", {keys.SET_VALUES: ("Script.inputs:script", s)})

        # evaluate
        await controller.evaluate()

        # check that we have a matrix
        matrix_attr = controller.attribute("outputs:matrix", script)
        matrix_val = matrix_attr.get()
        self.assertEqual(matrix_val[12], 0)
        self.assertEqual(matrix_val[13], 0)
        self.assertEqual(matrix_val[14], 0)

        attr = cube.GetAttribute("xformOp:translate")
        attr.Set((2, 3, 4))

        # evaluate
        await controller.evaluate()

        # check that we have a new matrix
        matrix_val = matrix_attr.get()
        self.assertEqual(matrix_val[12], 2)
        self.assertEqual(matrix_val[13], 3)
        self.assertEqual(matrix_val[14], 4)

    # ----------------------------------------------------------------------
    async def test_deprecated_dynamic_attributes_deprecated_bundle(self):
        """Test basic creation and interaction with a bundle dynamic attribute

        This is a backward compatibility test for deprecated ReadPrim.
        """

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

        usd_context = omni.usd.get_context()
        stage = usd_context.get_stage()
        cube = ogts.create_cube(stage, "World/Cube", (1, 1, 1))
        UsdGeom.Xformable(cube).AddTranslateOp()

        (_, nodes, _, _) = controller.edit(
            "/TestGraph",
            {
                keys.CREATE_NODES: [
                    ("ReadPrimBundle", "omni.graph.nodes.ReadPrimBundle"),
                    ("ExtractBundle", "omni.graph.nodes.ExtractBundle"),
                    ("Script", "omni.graph.scriptnode.ScriptNode"),
                ],
                keys.SET_VALUES: [
                    ("ReadPrimBundle.inputs:usePath", True),
                    ("ReadPrimBundle.inputs:primPath", "/World/Cube"),
                ],
                keys.CONNECT: [
                    ("ReadPrimBundle.outputs_primBundle", "ExtractBundle.inputs:bundle"),
                ],
            },
        )

        # creates some attributes on the script node
        script = nodes[2]
        matrix_type = og.Type(og.BaseDataType.DOUBLE, 16, 0, og.AttributeRole.MATRIX)
        self.assertTrue(
            script.create_attribute(
                "inputs:bundle", og.Type(og.BaseDataType.RELATIONSHIP, 1, 0, og.AttributeRole.BUNDLE)
            )
        )
        self.assertTrue(
            script.create_attribute("outputs:matrix", matrix_type, og.AttributePortType.ATTRIBUTE_PORT_TYPE_OUTPUT)
        )

        # connect them
        controller.edit("/TestGraph", {keys.CONNECT: ("ExtractBundle.outputs_passThrough", "Script.inputs:bundle")})

        # set a script that extract the matrix
        s = 'db.outputs.matrix = db.inputs.bundle.get_attribute_by_name("worldMatrix").get()'
        controller.edit("/TestGraph", {keys.SET_VALUES: ("Script.inputs:script", s)})

        # evaluate
        await controller.evaluate()

        # check that we have a matrix
        matrix_attr = controller.attribute("outputs:matrix", script)
        matrix_val = matrix_attr.get()
        self.assertEqual(matrix_val[12], 0)
        self.assertEqual(matrix_val[13], 0)
        self.assertEqual(matrix_val[14], 0)

        attr = cube.GetAttribute("xformOp:translate")
        attr.Set((2, 3, 4))

        # evaluate
        await controller.evaluate()

        # check that we have a new matrix
        matrix_val = matrix_attr.get()
        self.assertEqual(matrix_val[12], 2)
        self.assertEqual(matrix_val[13], 3)
        self.assertEqual(matrix_val[14], 4)

    # --------------------------------------------------------------------------------------------------------------
    async def test_dynamic_attributes_cpp(self):
        """Test basic creation and interaction with the dynamic attribute accessors for the C++ implementation"""
        await self.__test_dynamic_attributes("DynamicAttributes")

    # --------------------------------------------------------------------------------------------------------------
    async def test_dynamic_attribute_load(self):
        """Test basic creation and interaction with the dynamic attribute accessors, used in the test node compute"""
        (result, error) = await ogts.load_test_file("TestDynamicAttributes.usda", use_caller_subdirectory=True)
        self.assertTrue(result, error)
        controller = og.Controller()
        keys = og.Controller.Keys

        # The files each contain a set of test nodes in the different configurations of their dynamic attributes.
        # The name indicates which dynamic attributes exist on the node, similar to what the above test uses.
        # Test data is a list of five elements:
        #    Node to test
        #    Should the firstBit attribute be present?
        #    Should the secondBit attribute be present?
        #    Should the invert attribute be present?
        #    Expected output value when the input is set to 127
        test_data = [
            ["/graph/NoDynamicAttributes", False, False, False, 127],
            ["/graph/OnlyFirst", True, False, False, 119],
            ["/graph/FirstAndSecond", True, True, False, 103],
            ["/graph/SecondInvert", False, True, True, 0xFFFFFFFF - 111],
            ["/graph/NoDynamicAttributesPy", False, False, False, 127],
            ["/graph/OnlyFirstPy", True, False, False, 119],
            ["/graph/FirstAndSecondPy", True, True, False, 103],
            ["/graph/SecondInvertPy", False, True, True, 0xFFFFFFFF - 111],
        ]

        for node_name, first_expected, second_expected, invert_expected, result_expected in test_data:
            node = controller.node(node_name)
            controller.edit("/graph", {keys.SET_VALUES: (("inputs:value", node), 127)})
            await controller.evaluate()
            output = controller.attribute("outputs:result", node)
            self.assertIsNotNone(output)

            # Test for the presence of the expected dynamic attributes
            attributes_actual = [a.get_name() for a in node.get_attributes()]
            first_actual = "inputs:firstBit" in attributes_actual
            self.assertEqual(first_actual, first_expected, f"Dynamic attribute firstBit presence in {node_name}")
            second_actual = "inputs:secondBit" in attributes_actual
            self.assertEqual(second_actual, second_expected, f"Dynamic attribute secondBit presence in {node_name}")
            invert_actual = "inputs:invert" in attributes_actual
            self.assertEqual(invert_actual, invert_expected, f"Dynamic attribute invert presence in {node_name}")

            # Test the correct output value
            self.assertEqual(result_expected, og.Controller.get(output))

    # --------------------------------------------------------------------------------------------------------------
    async def __remove_connected_attribute(self, use_controller: bool):
        """
        Test what happens when removing dynamic attributes that have connections, both incoming and outgoing.

        Args:
            use_controller: If True then use the og.Controller methods rather than the direct commands
                        (This is mostly possible due to the fact that the controller and the command take the
                        same argument list.)
        """
        controller = og.Controller()
        keys = og.Controller.Keys

        (_, (source_node, sink_node), _, _) = controller.edit(
            self.TEST_GRAPH_PATH,
            {
                keys.CREATE_NODES: [
                    ("Source", "omni.graph.tutorials.DynamicAttributesPy"),
                    ("Sink", "omni.graph.tutorials.DynamicAttributesPy"),
                ]
            },
        )

        def run_create(**kwargs):
            """Select and run the appropriate create attribute command"""
            if use_controller:
                return og.NodeController.create_attribute(**kwargs) is not None

            if not og.cmds.CreateAttr(**kwargs)[1]:
                raise og.OmniGraphError(f"Create attribute using {kwargs} failed")
            return True

        def run_remove(**kwargs):
            """Select and run the appropriate remove attribute command"""
            if use_controller:
                return og.NodeController.remove_attribute(**kwargs)

            if not og.cmds.RemoveAttr(**kwargs)[1]:
                raise og.OmniGraphError(f"Remove attribute using {kwargs} failed")
            return True

        # Set up dynamic attributes on both source and sink and connect them
        self.assertTrue(
            run_create(
                node=source_node,
                attr_name="source",
                attr_type="float",
                attr_port=og.AttributePortType.ATTRIBUTE_PORT_TYPE_OUTPUT,
            )
        )
        source_attr = source_node.get_attribute("outputs:source")
        self.assertIsNotNone(source_attr)

        self.assertTrue(
            run_create(
                node=sink_node,
                attr_name="sink",
                attr_type="float",
                attr_port=og.AttributePortType.ATTRIBUTE_PORT_TYPE_INPUT,
            )
        )
        sink_attr = sink_node.get_attribute("inputs:sink")
        self.assertIsNotNone(sink_attr)

        controller.edit(self.TEST_GRAPH_PATH, {keys.CONNECT: (source_attr, sink_attr)})
        self.assertEqual([source_attr], sink_attr.get_upstream_connections())

        # Try to remove the source attribute, failing due to the connection existing
        with self.assertRaises(og.OmniGraphError):
            run_remove(attribute=source_attr)

        # Remove the connections, then remove the source attribute, which should now succeed
        (success, result) = og.cmds.DisconnectAllAttrs(attr=source_attr, modify_usd=True)
        self.assertTrue(success, f"Disconnect all from source - {result}")
        self.assertTrue(run_remove(attribute=source_attr), "Remove source")
        self.assertEqual(0, sink_attr.get_upstream_connection_count())

        # Restore the attribute and confirm the connection has returned
        omni.kit.undo.undo()
        omni.kit.undo.undo()
        source_attr = source_node.get_attribute("outputs:source")
        self.assertEqual(1, sink_attr.get_upstream_connection_count())
        self.assertEqual(1, source_attr.get_downstream_connection_count())

        # Try to remove the source attribute, failing due to the connection existing
        with self.assertRaises(og.OmniGraphError):
            run_remove(attribute=sink_attr)

        # Remove the connection, then remove the sink attribute, which should now succeed
        (success, result) = og.cmds.DisconnectAllAttrs(attr=sink_attr, modify_usd=True)
        self.assertTrue(success, "Disconnect all from sink")
        self.assertTrue(run_remove(attribute=sink_attr), "Remove sink")

        # Restore the attribute and confirm the connection has returned
        omni.kit.undo.undo()
        omni.kit.undo.undo()
        sink_attr = sink_node.get_attribute("inputs:sink")
        self.assertEqual(1, sink_attr.get_upstream_connection_count())
        self.assertEqual(1, source_attr.get_downstream_connection_count())

        # Confirm that the attribute configuration hasn't changed
        self.assertEqual("outputs:source", source_attr.get_name())
        self.assertEqual("float", source_attr.get_type_name())
        self.assertEqual(og.AttributePortType.ATTRIBUTE_PORT_TYPE_OUTPUT, source_attr.get_port_type())
        self.assertEqual(og.ExtendedAttributeType.EXTENDED_ATTR_TYPE_REGULAR, source_attr.get_extended_type())
        # TODO: Check on the default once it has been added
        self.assertEqual(0.0, source_attr.get())

        self.assertEqual("inputs:sink", sink_attr.get_name())
        self.assertEqual("float", sink_attr.get_type_name())
        self.assertEqual(og.AttributePortType.ATTRIBUTE_PORT_TYPE_INPUT, sink_attr.get_port_type())
        self.assertEqual(og.ExtendedAttributeType.EXTENDED_ATTR_TYPE_REGULAR, sink_attr.get_extended_type())
        # TODO: Check on the default once it has been added
        self.assertEqual(0.0, sink_attr.get())

    # --------------------------------------------------------------------------------------------------------------
    async def test_remove_connected_attribute_by_command(self):
        """Run the dynamic attribute removal tests by direct commands"""
        await self.__remove_connected_attribute(False)

    # --------------------------------------------------------------------------------------------------------------
    async def test_remove_connected_attribute_by_controller(self):
        """Run the dynamic attribute removal tests by using the Controller class"""
        await self.__remove_connected_attribute(True)

    # --------------------------------------------------------------------------------------------------------------
    async def test_dynamic_extended_attribute_customdata(self):
        """Test that creating dynamic extended attribute types sets the extended types correctly"""
        controller = og.Controller()
        keys = og.Controller.Keys
        (_, (node,), _, _) = controller.edit(
            self.TEST_GRAPH_PATH, {keys.CREATE_NODES: ("Node", "omni.graph.tutorials.DynamicAttributesPy")}
        )
        exec_attr = controller.create_attribute(
            node,
            "outputs:exec",
            og.Type(og.BaseDataType.UINT, 1, 0, og.AttributeRole.EXECUTION),
            og.AttributePortType.ATTRIBUTE_PORT_TYPE_OUTPUT,
        )

        any_attr = controller.create_attribute(
            node,
            "outputs:any",
            "any",
            og.AttributePortType.ATTRIBUTE_PORT_TYPE_OUTPUT,
            None,
            og.ExtendedAttributeType.EXTENDED_ATTR_TYPE_ANY,
        )

        union_attr = controller.create_attribute(
            node,
            "outputs:union",
            "[int, double[3], float[3]]",
            og.AttributePortType.ATTRIBUTE_PORT_TYPE_OUTPUT,
            None,
            (og.ExtendedAttributeType.EXTENDED_ATTR_TYPE_UNION, ["int", "double[3]", "float[3]"]),
        )

        int_attr = controller.create_attribute(node, "inputs:int", "int")

        stage = omni.usd.get_context().get_stage()
        node_prim = stage.GetPrimAtPath(node.get_prim_path())

        exec_metadata = node_prim.GetAttribute(exec_attr.get_name()).GetCustomData()
        any_metadata = node_prim.GetAttribute(any_attr.get_name()).GetCustomData()
        union_metadata = node_prim.GetAttribute(union_attr.get_name()).GetCustomData()
        int_metadata = node_prim.GetAttribute(int_attr.get_name()).GetCustomData()

        self.assertEqual(exec_metadata, {"ExecutionType": True})
        self.assertEqual(any_metadata, {"ExtendedAttributeType": "Any", "omni": {"graph": {"attrType": "Any"}}})
        self.assertEqual(
            union_metadata,
            {
                "ExtendedAttributeType": "Union-->int,double[3],float[3]",
                "omni": {"graph": {"attrType": "Union-->int,double[3],float[3]"}},
            },
        )
        self.assertEqual(int_metadata, {})

    # --------------------------------------------------------------------------------------------------------------
    async def test_load_dynamic_extended_attribute_customdata(self):
        """The test file uses the deprecated global implicit graph so this also tests automatic conversion"""
        (result, error) = await ogts.load_test_file("TestDynamicExtendedAttributes.usda", use_caller_subdirectory=True)
        self.assertTrue(result, error)
        controller = og.Controller()
        exec_attr = controller.attribute("outputs:exec", "/__graphUsingSchemas/node")
        any_attr = controller.attribute("outputs:any", "/__graphUsingSchemas/node")
        union_attr = controller.attribute("outputs:union", "/__graphUsingSchemas/node")

        self.assertEqual(exec_attr.get_resolved_type(), og.Type(og.BaseDataType.UINT, 1, 0, og.AttributeRole.EXECUTION))
        self.assertEqual(any_attr.get_resolved_type(), og.Type(og.BaseDataType.UNKNOWN))
        self.assertEqual(union_attr.get_resolved_type(), og.Type(og.BaseDataType.UNKNOWN))

        self.assertEqual(exec_attr.get_extended_type(), og.ExtendedAttributeType.EXTENDED_ATTR_TYPE_REGULAR)
        self.assertEqual(any_attr.get_extended_type(), og.ExtendedAttributeType.EXTENDED_ATTR_TYPE_ANY)
        self.assertEqual(union_attr.get_extended_type(), og.ExtendedAttributeType.EXTENDED_ATTR_TYPE_UNION)

        self.assertEqual(exec_attr.get_union_types(), None)
        self.assertEqual(any_attr.get_union_types(), None)
        self.assertEqual(union_attr.get_union_types(), ["int", "double[3]", "float[3]"])

    # --------------------------------------------------------------------------------------------------------------
    async def test_dynamic_attributes_memory_location(self):
        """Test that modifying the memory location of dynamic attributes changes how the data is set or retrieved"""
        controller = og.Controller(undoable=False)
        keys = og.Controller.Keys
        (graph, (test_node, _, _, _), _, _) = controller.edit(
            self.TEST_GRAPH_PATH,
            {
                keys.CREATE_NODES: [
                    ("DynamicNode", "omni.graph.test.TestDynamicAttributeMemory"),
                    ("ArrayDefault", "omni.graph.nodes.ConstantInt"),
                    ("ArrayNode", "omni.graph.nodes.ConstructArray"),
                    ("SimpleDefault", "omni.graph.nodes.ConstantDouble"),
                ],
                keys.SET_VALUES: [
                    ("ArrayDefault.inputs:value", 1),
                    ("ArrayNode.inputs:arraySize", 5),
                    ("ArrayNode.inputs:arrayType", "int[]"),
                    ("SimpleDefault.inputs:value", 4.25),
                ],
                keys.CONNECT: [
                    ("ArrayDefault.inputs:value", "ArrayNode.inputs:input0"),
                ],
            },
        )
        await controller.evaluate()
        on_gpu_attr = controller.attribute("inputs:onGpu", test_node)
        use_gpu_attr = controller.attribute("inputs:gpuPtrsOnGpu", test_node)
        in_verify_attr = controller.attribute("outputs:inputMemoryVerified", test_node)
        out_verify_attr = controller.attribute("outputs:outputMemoryVerified", test_node)
        self.assertTrue(on_gpu_attr.is_valid())
        self.assertTrue(use_gpu_attr.is_valid())
        self.assertTrue(in_verify_attr.is_valid())
        self.assertTrue(out_verify_attr.is_valid())

        # Meta-test that the test node does not give false positives when no dynamic attributes exist
        self.assertFalse(controller.get(in_verify_attr))
        self.assertFalse(controller.get(out_verify_attr))

        # Add the dynamic attributes the test node will be looking for
        in_attr = controller.create_attribute(test_node, "inputs:simple", "double")
        self.assertTrue(in_attr is not None and in_attr.is_valid())
        in_arr_attr = controller.create_attribute(test_node, "inputs:array", "int[]")
        self.assertTrue(in_arr_attr is not None and in_arr_attr.is_valid())
        out_attr = controller.create_attribute(test_node, "outputs:simple", "double")
        self.assertTrue(out_attr is not None and out_attr.is_valid())
        out_arr_attr = controller.create_attribute(test_node, "outputs:array", "int[]")
        self.assertTrue(out_arr_attr is not None and out_arr_attr.is_valid())
        controller.edit(graph, {keys.CONNECT: ("ArrayNode.outputs:array", "DynamicNode.inputs:array")})

        # Run through all of the legal memory locations, verifying the input and output memory locations
        for on_gpu in [False, True, True]:
            for use_gpu_ptr in [False, False, True]:
                controller.set(on_gpu_attr, on_gpu)
                controller.set(use_gpu_attr, use_gpu_ptr)
                await controller.evaluate()
                self.assertTrue(controller.get(in_verify_attr), f"on_gpu={on_gpu}, use_gpu_ptr={use_gpu_ptr}")
                self.assertTrue(controller.get(out_verify_attr), f"on_gpu={on_gpu}, use_gpu_ptr={use_gpu_ptr}")

    # ----------------------------------------------------------------------
    async def test_illegal_dynamic_attributes(self):
        """Test that attempting to create illegal attributes correctly raises an exception"""
        keys = og.Controller.Keys
        (_, (test_node,), _, _) = og.Controller.edit(
            self.TEST_GRAPH_PATH,
            {
                keys.CREATE_NODES: [
                    ("SimpleDefault", "omni.graph.nodes.ConstantDouble"),
                ],
            },
        )

        for undoable in [False, True]:
            controller = og.Controller(undoable=undoable)
            # First addition works
            in_attr = controller.create_attribute(test_node, "inputs:simple", "double")
            self.assertTrue(in_attr is not None and in_attr.is_valid())

            # Second addition raises an exception
            with self.assertRaises(og.OmniGraphError):
                with ogts.ExpectedError():
                    controller.create_attribute(test_node, "inputs:simple", "double")

            # First removal works
            controller.remove_attribute(in_attr)

            # Second removal raises an exception
            with self.assertRaises(og.OmniGraphError):
                with ogts.ExpectedError():
                    controller.remove_attribute("inputs:simple", test_node)

    # --------------------------------------------------------------------------------------------------------------
    async def test_adding_keyword_attributes(self):
        """Confirm that attempting to add attributes that are language keywords is disallowed"""
        controller = og.Controller()
        keys = og.Controller.Keys

        # The exact node type doesn't matter for this test, only that one is implemented in Python and one is C++
        (_, (py_node, cpp_node), _, _) = controller.edit(
            "/TestGraph",
            {
                keys.CREATE_NODES: [
                    ("PyNode", "omni.graph.test.TestSubtract"),
                    ("CppNode", "omni.graph.test.Add2IntegerArrays"),
                ]
            },
        )

        # First check that the Python node types rejects attributes named for Python keywords
        for kw in PYTHON_KEYWORDS:
            with ogts.ExpectedError():
                with self.assertRaises(ValueError):
                    py_node.create_attribute(f"inputs:{kw}", og.Type(og.BaseDataType.UINT))
                    carb.log_error(f"Incorrectly allowed addition of attribute named for Python keyword '{kw}'")

        # Now do the same for the C++ nodes, which have some "safe" keywords, though they do issue a warning
        for kw in CPP_KEYWORDS:
            if kw in SAFE_CPP_KEYWORDS:
                self.assertTrue(
                    cpp_node.create_attribute(f"inputs:{kw}", og.Type(og.BaseDataType.UINT)),
                    f"Trying to add attribute using C++ keyword '{kw}'",
                )
            else:
                with ogts.ExpectedError():
                    with self.assertRaises(ValueError):
                        cpp_node.create_attribute(f"inputs:{kw}", og.Type(og.BaseDataType.UINT))
                        carb.log_error(f"Incorrectly allowed addition of attribute named for C++ keyword '{kw}'")

    # --------------------------------------------------------------------------------------------------------------
    async def test_output_raw_data_works(self):
        """Validate that dynamic output and state attributes accessed from the node DB allow writing the raw data pointer"""

        controller = og.Controller(undoable=False)
        keys = og.Controller.Keys
        (graph, nodes, _, _) = controller.edit(
            self.TEST_GRAPH_PATH,
            {
                keys.CREATE_NODES: [
                    ("DynamicNode", "omni.graph.test.TestDynamicAttributeRawData"),
                ],
            },
        )
        # Evaluate the graph once to ensure the node has a Database created
        await controller.evaluate(graph)

        test_node = nodes[0]

        for i in range(10):
            controller.create_attribute(
                test_node,
                f"inputs:input{i}",
                og.Type(og.BaseDataType.INT),
                og.AttributePortType.ATTRIBUTE_PORT_TYPE_INPUT,
                i,
            )

        controller.create_attribute(
            test_node,
            "outputs:output0",
            og.Type(og.BaseDataType.INT),
            og.AttributePortType.ATTRIBUTE_PORT_TYPE_OUTPUT,
            0,
        )

        controller.create_attribute(
            test_node,
            "state:state0",
            og.Type(og.BaseDataType.INT),
            og.AttributePortType.ATTRIBUTE_PORT_TYPE_STATE,
            0,
        )

        # Act: evaluate the graph
        await controller.evaluate(graph)

        # Assert: read the sum output and test that the dynamic attributes were included in the computation
        output_attr = controller.attribute("outputs:output0", test_node)
        self.assertEqual(45, controller.get(output_attr), "The dynamic output does not have the expected value")

        state_attr = controller.attribute("state:state0", test_node)
        self.assertEqual(45, controller.get(state_attr), "The dynamic state does not have the expected value")
