# pylint: disable=too-many-lines
""" Tests for omnigraph variables """

import os
import string
import tempfile
from typing import List, Optional, Set

import omni.graph.core as og
import omni.graph.core.tests as ogts
import omni.kit.test
import omni.usd
from omni.kit.usd.layers import LayerUtils
from pxr import OmniGraphSchemaTools

ALL_USD_TYPES = ogts.DataTypeHelper.all_attribute_types()


# =====================================================================
# Helper method to create a flat list item, list, tuples, for easy comparison
def flatten(item):
    def _internal_flatten(item, flattened_list):
        if hasattr(item, "__iter__") and not isinstance(item, str):
            for i in item:
                flattened_list = _internal_flatten(i, flattened_list)
        else:
            flattened_list.append(item)
        return flattened_list

    return _internal_flatten(item, [])


# ======================================================================
class TestVariables(ogts.OmniGraphTestCase):

    _graph_path = "/World/TestGraph"

    async def setUp(self):
        self._var_index = 0
        await super().setUp()

    # ----------------------------------------------------------------------
    def create_variable(
        self,
        graph,
        name,
        var_type,
        display_name: Optional[str] = None,
        category: Optional[str] = None,
        tooltip: Optional[str] = None,
        scope: Optional[og.eVariableScope] = None,
    ):
        """Helper method to create and validate a variable in a graph"""
        var = graph.create_variable(name, var_type)
        self.assertNotEquals(var, None)
        self.assertEquals(var.source_path, self._graph_path + ".graph:variable:" + name)
        self.assertEquals(var.name, name)
        self.assertEquals(var.type, var_type)

        if display_name:
            var.display_name = display_name
            self.assertEquals(var.display_name, display_name)
        if category:
            var.category = category
            self.assertEquals(var.category, category)
        if tooltip:
            var.tooltip = tooltip
            self.assertEquals(var.tooltip, tooltip)
        if scope:
            var.scope = scope
            self.assertEquals(var.scope, scope)

        return var

    # ----------------------------------------------------------------------
    def create_simple_graph(self):
        (graph, _, _, _) = og.Controller.edit(
            self._graph_path,
            {
                og.Controller.Keys.CREATE_NODES: [("SimpleNode", "omni.graph.examples.cpp.Simple")],
            },
        )
        return graph

    # ----------------------------------------------------------------------
    def next_variable_name(self):
        """Returns a unique variable name each time it is called"""
        variable_name = "variable"
        index = self._var_index
        while index >= 0:
            variable_name += chr(ord("a") + index % 26)
            index -= 26
        self._var_index += 1
        return variable_name

    # ----------------------------------------------------------------------
    def get_variables_from_types(self):
        """Returns a list of variables of all supported types in the form
        (name, Og.Type, data)
        """
        return [
            (
                self.next_variable_name(),
                og.AttributeType.type_from_sdf_type_name(entry),
                ogts.DataTypeHelper.test_input_value(entry),
            )
            for entry in ALL_USD_TYPES
        ]

    # -------------------------------------------------------------------------
    async def test_graph_variables_return_expected_value(self):
        """
        Simple test that checks the API returns expected values from a loaded file
        """
        # load the graph
        (result, error) = await ogts.load_test_file("TestGraphVariables.usda", use_caller_subdirectory=True)
        self.assertTrue(result, error)
        graph = omni.graph.core.get_graph_by_path("/World/ComputeGraph")

        # put the variables in a dictionary
        variables = {var.source_path: var for var in graph.get_variables()}
        self.assertEquals(len(variables), 103)

        # walk through the usd types and validate they all load as variables
        # and return expected values
        for usd_type in ALL_USD_TYPES:
            var_name = usd_type.lower().replace("[]", "_array")
            key = "/World/ComputeGraph.graph:variable:var_" + var_name
            self.assertTrue(key in variables, key)
            self.assertEquals(variables[key].name, "var_" + var_name)

            is_array = "_array" in key or usd_type == "string"
            var_type = variables[key].type
            if is_array:
                self.assertNotEqual(var_type.array_depth, 0, key)
            else:
                self.assertEqual(var_type.array_depth, 0, key)

    async def test_variables_get_and_set_oncontext(self):
        """
        Validates variable values can be written and retrieved through the API
        """
        # turn off global graphs, as the test file doesn't use them
        (result, error) = await ogts.load_test_file("TestGraphVariables.usda", use_caller_subdirectory=True)
        self.assertTrue(result, error)

        graph = omni.graph.core.get_graph_by_path("/World/ComputeGraph")
        context = graph.get_default_graph_context()

        variables = {var.source_path: var for var in graph.get_variables()}

        for entry in ogts.DataTypeHelper.all_attribute_types():
            var_name = entry.lower().replace("[]", "_array")
            key = "/World/ComputeGraph.graph:variable:var_" + var_name

            variable = variables[key]

            # grab the reference value, set it, read it back and validate they are the same
            ref_value = ogts.DataTypeHelper.test_input_value(entry)
            variable.set(context, ref_value)

            if "_array" in key:
                value = variable.get_array(context, False, 0)
            else:
                value = variable.get(context)

            # use almostEqual to deal with numerical imprecision
            # that occurs with some floating point precisions
            for a, b in zip(flatten(value), flatten(ref_value)):
                self.assertAlmostEqual(a, b, places=1)

    async def test_variables_can_be_created_and_removed(self):
        """Validates variables are correctly created and removed"""
        graph = self.create_simple_graph()

        # validate variables can be created
        var1 = self.create_variable(graph, "var_float", og.Type(og.BaseDataType.FLOAT))
        var2 = self.create_variable(graph, "var_int", og.Type(og.BaseDataType.INT))
        var3 = self.create_variable(graph, "var_float_array", og.Type(og.BaseDataType.FLOAT, 3, 0))
        var4 = self.create_variable(graph, "var_int_tuple", og.Type(og.BaseDataType.INT, 2, 0))

        # check they are all valid, using both valid property and __bool__ override
        self.assertTrue(var1.valid)
        self.assertTrue(var2.valid)
        self.assertTrue(var3.valid)
        self.assertTrue(var4.valid)
        self.assertTrue(bool(var1))
        self.assertTrue(bool(var2))
        self.assertTrue(bool(var3))
        self.assertTrue(bool(var4))

        # check that the list is created as expected
        variables = graph.get_variables()
        self.assertEquals(len(variables), 4)
        self.assertIn(var1, variables)
        self.assertIn(var2, variables)
        self.assertIn(var3, variables)
        self.assertIn(var4, variables)

        # validate nodes that already exist (same type, different type) are ignored
        with ogts.ExpectedError():
            var_x = graph.create_variable("var_float", og.Type(og.BaseDataType.FLOAT))
            self.assertEquals(var_x, None)
        with ogts.ExpectedError():
            var_x = graph.create_variable("var_float", og.Type(og.BaseDataType.INT))
            self.assertEquals(var_x, None)

        # remove a variable, then again to verify it fails
        self.assertTrue(graph.remove_variable(var3))
        with ogts.ExpectedError():
            self.assertFalse(graph.remove_variable(var3))
        self.assertEquals(len(graph.get_variables()), 3)
        self.assertNotIn(var3, graph.get_variables())
        self.assertFalse(var3.valid)
        self.assertFalse(bool(var3))

    async def test_find_variable(self):
        """Tests that variables can be correctly searched for by name"""
        graph = self.create_simple_graph()

        # create a set of variables
        for a in string.ascii_lowercase:
            self.create_variable(graph, "var_" + a, og.Type(og.BaseDataType.FLOAT))

        # validate they are correctly found
        for a in string.ascii_lowercase:
            var = graph.find_variable("var_" + a)
            self.assertIsNotNone(var)
            self.assertEquals(var.name, "var_" + a)
            self.assertIsNone(graph.find_variable("not_var_" + a))

    async def test_variable_metadata(self):
        """Validates the metadata for the variables can and retreived as expected."""
        graph = self.create_simple_graph()

        var = graph.create_variable("var", og.Type(og.BaseDataType.FLOAT))
        self.assertTrue(var.valid)

        # validate default values
        self.assertEquals(var.display_name, "var")
        self.assertEquals(var.category, "")
        self.assertEquals(var.tooltip, "")
        self.assertEquals(var.scope, og.eVariableScope.E_PRIVATE)

        # set valid values
        var.display_name = "VAR1"
        var.category = "New Category"
        var.tooltip = "Tooltip"
        var.scope = og.eVariableScope.E_READ_ONLY

        self.assertEquals(var.display_name, "VAR1")
        self.assertEquals(var.category, "New Category")
        self.assertEquals(var.tooltip, "Tooltip")
        self.assertEquals(var.scope, og.eVariableScope.E_READ_ONLY)

        # set special and/or secondary values
        var.display_name = ""
        var.category = ""
        var.tooltip = ""
        var.scope = og.eVariableScope.E_PUBLIC

        self.assertEquals(var.display_name, "var")  # "" resets to default value
        self.assertEquals(var.category, "")
        self.assertEquals(var.tooltip, "")
        self.assertEquals(var.scope, og.eVariableScope.E_PUBLIC)

        var.category = "New Category"
        var.tooltip = "Tooltip"

        # validate that removed attribute specify empty/default values
        graph.remove_variable(var)
        self.assertFalse(var.valid)
        self.assertEquals(var.display_name, "")
        self.assertEquals(var.category, "")
        self.assertEquals(var.tooltip, "")
        self.assertEquals(var.scope, og.eVariableScope.E_PRIVATE)

    async def test_variables_save_and_load(self):
        """Verifies that variables are correctly saved and restored from serialization"""
        type_list = [og.BaseDataType.FLOAT, og.BaseDataType.INT, og.BaseDataType.HALF]
        graph = self.create_simple_graph()

        for x in type_list:
            type_name = str(x).replace(".", "_")
            self.create_variable(
                graph,
                str(type_name).lower(),
                og.Type(x),
                display_name=str(type_name).upper(),
                category=str(type_name).swapcase(),
                scope=og.eVariableScope.E_PUBLIC,
                tooltip=str(type_name).title(),
            )

        with tempfile.TemporaryDirectory() as tmpdirname:
            # save the file
            tmp_file_path = os.path.join(tmpdirname, "tmp.usda")
            result = omni.usd.get_context().save_as_stage(tmp_file_path)
            self.assertTrue(result)

            # refresh the stage
            await omni.usd.get_context().new_stage_async()

            # reload the file back
            (result, error) = await ogts.load_test_file(tmp_file_path)
            self.assertTrue(result, error)

        graph = omni.graph.core.get_graph_by_path(self._graph_path)
        self.assertTrue(graph.is_valid)

        self.assertEquals(len(graph.get_variables()), len(type_list))
        for x in type_list:
            type_name = str(x).replace(".", "_")
            var = graph.find_variable(str(type_name).lower())
            self.assertEquals(var.type, og.Type(x))
            self.assertEquals(var.scope, og.eVariableScope.E_PUBLIC)
            self.assertEquals(var.display_name, str(type_name).upper())
            self.assertEquals(var.category, str(type_name).swapcase())
            self.assertEquals(var.tooltip, str(type_name).title())

    # ----------------------------------------------------------------------
    async def test_write_variable_node(self):
        """Tests WriteVariable Nodes that variable data is correctly written for all variable types"""
        # +-------+   +-------------+
        # |OnTick |-->|WriteVariable|
        # +-------+   +-------------+

        # C++ node, Python equivalent
        for write_node in ["omni.graph.core.WriteVariable", "omni.graph.test.TestWriteVariablePy"]:
            with self.subTest(write_node=write_node):
                controller = og.Controller()
                keys = og.Controller.Keys
                variables = self.get_variables_from_types()

                # Create a graph that writes the give variable
                (graph, (_, write_variable_node), _, _) = controller.edit(
                    self._graph_path,
                    {
                        keys.CREATE_NODES: [
                            ("OnTick", "omni.graph.action.OnTick"),
                            ("WriteVariable", write_node),
                        ],
                        keys.CONNECT: [
                            ("OnTick.outputs:tick", "WriteVariable.inputs:execIn"),
                        ],
                        keys.CREATE_VARIABLES: [(var_name, var_type) for (var_name, var_type, _) in variables],
                        keys.SET_VALUES: [
                            ("OnTick.inputs:onlyPlayback", False),
                        ],
                    },
                )

                for (var_name, og_type, ref_value) in variables:
                    variable = graph.find_variable(var_name)

                    # First write the variable type, then the value. The value field needs to resolve first,
                    # based on the type of variable that is set
                    variable_name_attr = write_variable_node.get_attribute("inputs:variableName")
                    variable_value_attr = write_variable_node.get_attribute("inputs:value")
                    og.Controller.set(variable_name_attr, var_name)
                    og.Controller.set(variable_value_attr, ref_value)

                    # Evaluate the graph, which writes the value
                    await controller.evaluate(graph)

                    # fetch the data from variable node
                    variable_output_attr = write_variable_node.get_attribute("outputs:value")

                    # retrieve the value from the graph variable, and the output port
                    if og_type.array_depth >= 1:
                        value = variable.get_array(graph.get_default_graph_context(), False, 0)
                        output_value = variable_output_attr.get_array(False, False, 0)
                    else:
                        value = variable.get(graph.get_default_graph_context())
                        output_value = variable_output_attr.get()

                    # unpack and compare the elements
                    for a, b, c in zip(flatten(value), flatten(ref_value), flatten(output_value)):
                        self.assertAlmostEqual(a, b, places=1, msg=f"{var_name} {og_type}")
                        self.assertAlmostEqual(a, c, places=1, msg=f"{var_name} {og_type}")

            # reset the stage, so the next iteration starts clean
            await omni.usd.get_context().new_stage_async()
            await omni.kit.app.get_app().next_update_async()

    # ----------------------------------------------------------------------
    async def test_variable_graph_type_resolution(self):
        """Tests output port resolution for built"""
        controller = og.Controller()
        keys = og.Controller.Keys

        #                    [ReadVariable]->+------------+
        # [ConstantString]->[WriteVariable]->|AppendString|->[WriteVariable]
        #                                    +------------+
        (graph, _, _, _) = controller.edit(
            self._graph_path,
            {
                keys.CREATE_NODES: [
                    ("ReadVariable", "omni.graph.core.ReadVariable"),
                    ("WriteVariable", "omni.graph.core.WriteVariable"),
                    ("AppendString", "omni.graph.nodes.AppendString"),
                    ("WriteVariable2", "omni.graph.core.WriteVariable"),
                    ("ConstantToken", "omni.graph.nodes.ConstantToken"),
                ],
                keys.CREATE_VARIABLES: [
                    ("var_float", og.Type(og.BaseDataType.FLOAT)),
                    ("var_str", og.Type(og.BaseDataType.TOKEN)),
                ],
                keys.CONNECT: [
                    ("ReadVariable.outputs:value", "AppendString.inputs:value"),
                    ("WriteVariable.outputs:value", "AppendString.inputs:suffix"),
                    ("AppendString.outputs:value", "WriteVariable2.inputs:value"),
                    ("ConstantToken.inputs:value", "WriteVariable.inputs:value"),
                ],
                keys.SET_VALUES: [
                    ("ReadVariable.inputs:variableName", "var_str"),
                    ("WriteVariable.inputs:variableName", "var_str"),
                    ("WriteVariable2.inputs:variableName", "var_str"),
                    ("ConstantToken.inputs:value", "_A_"),
                ],
            },
        )

        await omni.kit.app.get_app().next_update_async()

        # Helper to verify the base data type of an attribute
        def assert_attrib_is(attrib, node, base_type):
            attrib = controller.attribute(attrib, node, graph)
            self.assertEqual(attrib.get_resolved_type().base_type, base_type)

        # validate the output types
        assert_attrib_is("outputs:value", "ReadVariable", og.BaseDataType.TOKEN)
        assert_attrib_is("outputs:value", "WriteVariable", og.BaseDataType.TOKEN)
        assert_attrib_is("inputs:value", "WriteVariable", og.BaseDataType.TOKEN)
        assert_attrib_is("inputs:value", "WriteVariable2", og.BaseDataType.TOKEN)

        # disconnect the input
        controller.edit(
            graph,
            {
                keys.DISCONNECT: [
                    ("ReadVariable.outputs:value", "AppendString.inputs:value"),
                    ("WriteVariable.outputs:value", "AppendString.inputs:suffix"),
                    ("ConstantToken.inputs:value", "WriteVariable.inputs:value"),
                ]
            },
        )
        await omni.kit.app.get_app().next_update_async()

        # resolution types
        assert_attrib_is("outputs:value", "ReadVariable", og.BaseDataType.TOKEN)
        assert_attrib_is("outputs:value", "WriteVariable", og.BaseDataType.TOKEN)
        assert_attrib_is("inputs:value", "WriteVariable", og.BaseDataType.TOKEN)
        assert_attrib_is("inputs:value", "AppendString", og.BaseDataType.UNKNOWN)
        assert_attrib_is("inputs:suffix", "AppendString", og.BaseDataType.UNKNOWN)

        # change the variable type one at a time
        controller.edit(
            graph,
            {
                keys.SET_VALUES: [
                    ("WriteVariable.inputs:variableName", "var_float"),
                    ("ReadVariable.inputs:variableName", "var_float"),
                ],
            },
        )
        await omni.kit.app.get_app().next_update_async()

        # validate the updated resolution - outputs change, inputs unresolve
        assert_attrib_is("outputs:value", "ReadVariable", og.BaseDataType.FLOAT)
        assert_attrib_is("outputs:value", "WriteVariable", og.BaseDataType.FLOAT)
        assert_attrib_is("inputs:value", "WriteVariable", og.BaseDataType.FLOAT)

        # 'unset' the variable name
        controller.edit(
            graph,
            {
                keys.SET_VALUES: [
                    ("ReadVariable.inputs:variableName", ""),
                    ("WriteVariable.inputs:variableName", ""),
                ]
            },
        )
        await omni.kit.app.get_app().next_update_async()

        assert_attrib_is("outputs:value", "ReadVariable", og.BaseDataType.UNKNOWN)
        assert_attrib_is("outputs:value", "WriteVariable", og.BaseDataType.UNKNOWN)
        assert_attrib_is("inputs:value", "WriteVariable", og.BaseDataType.UNKNOWN)

    # ----------------------------------------------------------------------
    async def test_read_variable_node(self):
        """Tests ReadVariable Nodes that variable data is properly read for all variable types"""

        # C++ node, Python equivalent
        for read_node in ["omni.graph.core.ReadVariable", "omni.graph.test.TestReadVariablePy"]:
            with self.subTest(read_node=read_node):
                controller = og.Controller()
                keys = og.Controller.Keys

                variables = self.get_variables_from_types()

                # Create a graph that writes the give variable
                (graph, nodes, _, _) = controller.edit(
                    self._graph_path,
                    {
                        keys.CREATE_NODES: [
                            ("ReadVariable", read_node),
                        ],
                        keys.CREATE_VARIABLES: [(var_name, var_type) for (var_name, var_type, _) in variables],
                    },
                )
                context = graph.get_default_graph_context()

                for (var_name, var_type, ref_value) in variables:
                    # set the value on the variable
                    variable = graph.find_variable(var_name)
                    variable.set(context, ref_value)

                    # set the node to use the appropriate variable
                    variable_name_attr = nodes[0].get_attribute("inputs:variableName")
                    og.Controller.set(variable_name_attr, var_name)

                    # Evaluate the graph and read the output of the node
                    await controller.evaluate(graph)
                    value = og.Controller.get(nodes[0].get_attribute("outputs:value"))

                    # unpack and compare the elements
                    for a, b in zip(flatten(value), flatten(ref_value)):
                        self.assertAlmostEqual(a, b, places=1, msg=f"{var_name} : {var_type}")

            # reset the stage, so the next iteration starts clean
            await omni.usd.get_context().new_stage_async()
            await omni.kit.app.get_app().next_update_async()

    # ----------------------------------------------------------------------
    async def test_variable_graph_events(self):
        """Tests that variable related graph events trigger and are caught by built-in variable nodes"""
        controller = og.Controller()
        keys = og.Controller.Keys

        (graph, (read_var_node, write_var_node), _, _) = controller.edit(
            self._graph_path,
            {
                keys.CREATE_NODES: [
                    ("ReadVariable", "omni.graph.core.ReadVariable"),
                    ("WriteVariable", "omni.graph.core.WriteVariable"),
                ],
                keys.SET_VALUES: [
                    ("ReadVariable.inputs:variableName", "test_var"),
                    ("WriteVariable.inputs:variableName", "test_var"),
                ],
            },
        )

        create_counter = 0
        remove_counter = 0
        last_var_name = ""

        def on_graph_event(event):
            nonlocal create_counter
            nonlocal remove_counter
            nonlocal last_var_name

            if event.type == int(og.GraphEvent.CREATE_VARIABLE):
                create_counter += 1
            if event.type == int(og.GraphEvent.REMOVE_VARIABLE):
                remove_counter += 1
            last_var_name = event.payload["variable"]

        sub = graph.get_event_stream().create_subscription_to_pop(on_graph_event)

        var = graph.create_variable("test_var", og.Type(og.BaseDataType.FLOAT))

        # check the callback has been hit, and that the resolved type has been updated
        self.assertEquals(create_counter, 1)
        self.assertEquals(
            read_var_node.get_attribute("outputs:value").get_resolved_type(), og.Type(og.BaseDataType.FLOAT)
        )
        self.assertEquals(
            write_var_node.get_attribute("inputs:value").get_resolved_type(), og.Type(og.BaseDataType.FLOAT)
        )
        self.assertEquals(last_var_name, "test_var")
        last_var_name = ""

        # remove the variable, it should unresolve
        graph.remove_variable(var)
        self.assertEquals(remove_counter, 1)
        self.assertEquals(
            read_var_node.get_attribute("outputs:value").get_resolved_type(), og.Type(og.BaseDataType.UNKNOWN)
        )
        self.assertEquals(
            write_var_node.get_attribute("inputs:value").get_resolved_type(), og.Type(og.BaseDataType.UNKNOWN)
        )
        self.assertEquals(last_var_name, "test_var")

        # add, remove a variety of variables, including the listened one (with a different type)
        graph.create_variable("test_var", og.Type(og.BaseDataType.BOOL))
        for i in range(1, 10):
            var_n = graph.create_variable(f"test_var{i}", og.Type(og.BaseDataType.INT))
            self.assertEquals(last_var_name, f"test_var{i}")
            if i % 2 == 0:
                graph.remove_variable(var_n)

        self.assertEquals(create_counter, 11)
        self.assertEquals(remove_counter, 5)
        self.assertEquals(
            read_var_node.get_attribute("outputs:value").get_resolved_type(), og.Type(og.BaseDataType.BOOL)
        )
        self.assertEquals(
            write_var_node.get_attribute("inputs:value").get_resolved_type(), og.Type(og.BaseDataType.BOOL)
        )

        del sub

    def assert_almost_equal(self, a_val, b_val):
        flat_a = flatten(a_val)
        flat_b = flatten(b_val)
        self.assertEqual(len(flat_a), len(flat_b))
        for a, b in zip(flatten(a_val), flatten(b_val)):
            self.assertAlmostEqual(a, b, places=1, msg=f"{a_val} != {b_val}")

    # ----------------------------------------------------------------------
    async def test_variables_load_default_values(self):
        """Tests variables are initialized with attribute default values"""
        (result, error) = await ogts.load_test_file("TestGraphVariables.usda", use_caller_subdirectory=True)
        self.assertTrue(result, error)

        graph = omni.graph.core.get_graph_by_path("/World/ComputeGraph")
        ctxt = graph.get_default_graph_context()

        self.assert_almost_equal(graph.find_variable("var_bool").get(ctxt), 1)
        self.assert_almost_equal(graph.find_variable("var_bool_array").get_array(ctxt), [1, 0, 1])
        self.assert_almost_equal(graph.find_variable("var_color3d").get(ctxt), (1, 2, 3))
        self.assert_almost_equal(graph.find_variable("var_color3d_array").get_array(ctxt), [(1, 2, 3), (11, 12, 13)])
        self.assert_almost_equal(graph.find_variable("var_color3f").get(ctxt), (1, 2, 3))
        self.assert_almost_equal(graph.find_variable("var_color3f_array").get_array(ctxt), [(1, 2, 3), (11, 12, 13)])
        self.assert_almost_equal(graph.find_variable("var_color3h").get(ctxt), (1, 2, 3))
        self.assert_almost_equal(graph.find_variable("var_color3h_array").get_array(ctxt), [(1, 2, 3), (11, 12, 13)])
        self.assert_almost_equal(graph.find_variable("var_color4d").get(ctxt), (1, 2, 3, 4))
        self.assert_almost_equal(
            graph.find_variable("var_color4d_array").get_array(ctxt), [(1, 2, 3, 4), (11, 12, 13, 14)]
        )
        self.assert_almost_equal(graph.find_variable("var_color4f").get(ctxt), (1, 2, 3, 4))
        self.assert_almost_equal(
            graph.find_variable("var_color4f_array").get_array(ctxt), [(1, 2, 3, 4), (11, 12, 13, 14)]
        )
        self.assert_almost_equal(graph.find_variable("var_color4h").get(ctxt), (1, 2, 3, 4))
        self.assert_almost_equal(
            graph.find_variable("var_color4h_array").get_array(ctxt), [(1, 2, 3, 4), (11, 12, 13, 14)]
        )
        self.assert_almost_equal(graph.find_variable("var_double").get(ctxt), 1)
        self.assert_almost_equal(graph.find_variable("var_double2").get(ctxt), (1, 2))
        self.assert_almost_equal(graph.find_variable("var_double2_array").get_array(ctxt), [(1, 2), (11, 12)])
        self.assert_almost_equal(graph.find_variable("var_double3").get(ctxt), (1, 2, 3))
        self.assert_almost_equal(graph.find_variable("var_double3_array").get_array(ctxt), [(1, 2, 3), (11, 12, 13)])
        self.assert_almost_equal(graph.find_variable("var_double4").get(ctxt), (1, 2, 3, 4))
        self.assert_almost_equal(
            graph.find_variable("var_double4_array").get_array(ctxt), [(1, 2, 3, 4), (11, 12, 13, 14)]
        )
        self.assert_almost_equal(graph.find_variable("var_double_array").get_array(ctxt), [1, 2])
        self.assert_almost_equal(graph.find_variable("var_float").get(ctxt), 1)
        self.assert_almost_equal(graph.find_variable("var_float2").get(ctxt), (1, 2))
        self.assert_almost_equal(graph.find_variable("var_float2_array").get_array(ctxt), [(1, 2), (11, 12)])
        self.assert_almost_equal(graph.find_variable("var_float3").get(ctxt), (1, 2, 3))
        self.assert_almost_equal(graph.find_variable("var_float3_array").get_array(ctxt), [(1, 2, 3), (11, 12, 13)])
        self.assert_almost_equal(graph.find_variable("var_float4").get(ctxt), (1, 2, 3, 4))
        self.assert_almost_equal(
            graph.find_variable("var_float4_array").get_array(ctxt), [(1, 2, 3, 4), (11, 12, 13, 14)]
        )
        self.assert_almost_equal(graph.find_variable("var_float_array").get_array(ctxt), [1, 2])
        self.assert_almost_equal(
            graph.find_variable("var_frame4d").get(ctxt),
            ((1, 2, 3, 4), (5, 6, 7, 8), (9, 10, 11, 12), (13, 14, 15, 16)),
        )
        self.assert_almost_equal(
            graph.find_variable("var_frame4d_array").get_array(ctxt),
            [
                ((1, 2, 3, 4), (5, 6, 7, 8), (9, 10, 11, 12), (13, 14, 15, 16)),
                ((11, 12, 13, 14), (15, 16, 17, 18), (19, 20, 21, 22), (23, 24, 25, 26)),
            ],
        )
        self.assert_almost_equal(graph.find_variable("var_half").get(ctxt), 1)
        self.assert_almost_equal(graph.find_variable("var_half2").get(ctxt), (1, 2))
        self.assert_almost_equal(graph.find_variable("var_half2_array").get_array(ctxt), [(1, 2), (11, 12)])
        self.assert_almost_equal(graph.find_variable("var_half3").get(ctxt), (1, 2, 3))
        self.assert_almost_equal(graph.find_variable("var_half3_array").get_array(ctxt), [(1, 2, 3), (11, 12, 13)])
        self.assert_almost_equal(graph.find_variable("var_half4").get(ctxt), (1, 2, 3, 4))
        self.assert_almost_equal(
            graph.find_variable("var_half4_array").get_array(ctxt), [(1, 2, 3, 4), (11, 12, 13, 14)]
        )
        self.assert_almost_equal(graph.find_variable("var_half_array").get_array(ctxt), [1, 2])
        self.assert_almost_equal(graph.find_variable("var_int").get(ctxt), 1)
        self.assert_almost_equal(graph.find_variable("var_int2").get(ctxt), (1, 2))
        self.assert_almost_equal(graph.find_variable("var_int2_array").get_array(ctxt), [(1, 2), (3, 4)])
        self.assert_almost_equal(graph.find_variable("var_int3").get(ctxt), (1, 2, 3))
        self.assert_almost_equal(graph.find_variable("var_int3_array").get_array(ctxt), [(1, 2, 3), (4, 5, 6)])
        self.assert_almost_equal(graph.find_variable("var_int4").get(ctxt), (1, 2, 3, 4))
        self.assert_almost_equal(graph.find_variable("var_int4_array").get_array(ctxt), [(1, 2, 3, 4), (5, 6, 7, 8)])
        self.assert_almost_equal(graph.find_variable("var_int64").get(ctxt), 12345)
        self.assert_almost_equal(graph.find_variable("var_int64_array").get_array(ctxt), [12345, 23456])
        self.assert_almost_equal(graph.find_variable("var_int_array").get_array(ctxt), [1, 2])
        self.assert_almost_equal(graph.find_variable("var_matrix2d").get(ctxt), ((1, 2), (3, 4)))
        self.assert_almost_equal(
            graph.find_variable("var_matrix2d_array").get_array(ctxt), [((1, 2), (3, 4)), ((11, 12), (13, 14))]
        )
        self.assert_almost_equal(graph.find_variable("var_matrix3d").get(ctxt), ((1, 2, 3), (4, 5, 6), (7, 8, 9)))
        self.assert_almost_equal(
            graph.find_variable("var_matrix3d_array").get_array(ctxt),
            [((1, 2, 3), (4, 5, 6), (7, 8, 9)), ((11, 12, 13), (14, 15, 16), (17, 18, 19))],
        )
        self.assert_almost_equal(
            graph.find_variable("var_matrix4d").get(ctxt),
            ((1, 2, 3, 4), (5, 6, 7, 8), (9, 10, 11, 12), (13, 14, 15, 16)),
        )
        self.assert_almost_equal(
            graph.find_variable("var_matrix4d_array").get_array(ctxt),
            [
                ((1, 2, 3, 4), (5, 6, 7, 8), (9, 10, 11, 12), (13, 14, 15, 16)),
                ((11, 12, 13, 14), (15, 16, 17, 18), (19, 20, 21, 22), (23, 24, 25, 26)),
            ],
        )
        self.assert_almost_equal(graph.find_variable("var_normal3d").get(ctxt), (1, 2, 3))
        self.assert_almost_equal(graph.find_variable("var_normal3d_array").get_array(ctxt), [(1, 2, 3), (11, 12, 13)])
        self.assert_almost_equal(graph.find_variable("var_normal3f").get(ctxt), (1, 2, 3))
        self.assert_almost_equal(graph.find_variable("var_normal3f_array").get_array(ctxt), [(1, 2, 3), (11, 12, 13)])
        self.assert_almost_equal(graph.find_variable("var_normal3h").get(ctxt), (1, 2, 3))
        self.assert_almost_equal(graph.find_variable("var_normal3h_array").get_array(ctxt), [(1, 2, 3), (11, 12, 13)])
        self.assert_almost_equal(graph.find_variable("var_point3d").get(ctxt), (1, 2, 3))
        self.assert_almost_equal(graph.find_variable("var_point3d_array").get_array(ctxt), [(1, 2, 3), (11, 12, 13)])
        self.assert_almost_equal(graph.find_variable("var_point3f").get(ctxt), (1, 2, 3))
        self.assert_almost_equal(graph.find_variable("var_point3f_array").get_array(ctxt), [(1, 2, 3), (11, 12, 13)])
        self.assert_almost_equal(graph.find_variable("var_point3h").get(ctxt), (1, 2, 3))
        self.assert_almost_equal(graph.find_variable("var_point3h_array").get_array(ctxt), [(1, 2, 3), (11, 12, 13)])
        # note the reordering of the quaternion layout
        self.assert_almost_equal(graph.find_variable("var_quatd").get(ctxt), (2, 3, 4, 1))
        self.assert_almost_equal(
            graph.find_variable("var_quatd_array").get_array(ctxt), [(2, 3, 4, 1), (12, 13, 14, 11)]
        )
        self.assert_almost_equal(graph.find_variable("var_quatf").get(ctxt), (2, 3, 4, 1))
        self.assert_almost_equal(
            graph.find_variable("var_quatf_array").get_array(ctxt), [(2, 3, 4, 1), (12, 13, 14, 11)]
        )
        self.assert_almost_equal(graph.find_variable("var_quath").get(ctxt), (2, 3, 4, 1))
        self.assert_almost_equal(
            graph.find_variable("var_quath_array").get_array(ctxt), [(2, 3, 4, 1), (12, 13, 14, 11)]
        )
        # strings use get_array, not get
        self.assertEqual(graph.find_variable("var_string").get_array(ctxt), "Rey")
        self.assert_almost_equal(graph.find_variable("var_texcoord2d").get(ctxt), (1, 2))
        self.assert_almost_equal(graph.find_variable("var_texcoord2d_array").get_array(ctxt), [(1, 2), (11, 12)])
        self.assert_almost_equal(graph.find_variable("var_texcoord2f").get(ctxt), (1, 2))
        self.assert_almost_equal(graph.find_variable("var_texcoord2f_array").get_array(ctxt), [(1, 2), (11, 12)])
        self.assert_almost_equal(graph.find_variable("var_texcoord2h").get(ctxt), (1, 2))
        self.assert_almost_equal(graph.find_variable("var_texcoord2h_array").get_array(ctxt), [(1, 2), (11, 12)])
        self.assert_almost_equal(graph.find_variable("var_texcoord3d").get(ctxt), (1, 2, 3))
        self.assert_almost_equal(graph.find_variable("var_texcoord3d_array").get_array(ctxt), [(1, 2, 3), (11, 12, 13)])
        self.assert_almost_equal(graph.find_variable("var_texcoord3f").get(ctxt), (1, 2, 3))
        self.assert_almost_equal(graph.find_variable("var_texcoord3f_array").get_array(ctxt), [(1, 2, 3), (11, 12, 13)])
        self.assert_almost_equal(graph.find_variable("var_texcoord3h").get(ctxt), (1, 2, 3))
        self.assert_almost_equal(graph.find_variable("var_texcoord3h_array").get_array(ctxt), [(1, 2, 3), (11, 12, 13)])
        self.assert_almost_equal(graph.find_variable("var_timecode").get(ctxt), 1)
        self.assert_almost_equal(graph.find_variable("var_timecode_array").get_array(ctxt), [1, 2])
        self.assert_almost_equal(graph.find_variable("var_token").get(ctxt), "Sith")
        self.assert_almost_equal(graph.find_variable("var_token_array").get_array(ctxt), ["Kylo", "Ren"])
        self.assert_almost_equal(graph.find_variable("var_uchar").get(ctxt), 1)
        self.assert_almost_equal(graph.find_variable("var_uchar_array").get_array(ctxt), [1, 2])
        self.assert_almost_equal(graph.find_variable("var_uint").get(ctxt), 1)
        self.assert_almost_equal(graph.find_variable("var_uint64").get(ctxt), 1)
        self.assert_almost_equal(graph.find_variable("var_uint64_array").get_array(ctxt), [1, 2])
        self.assert_almost_equal(graph.find_variable("var_uint_array").get_array(ctxt), [1, 2])
        self.assert_almost_equal(graph.find_variable("var_vector3d").get(ctxt), (1, 2, 3))
        self.assert_almost_equal(graph.find_variable("var_vector3d_array").get_array(ctxt), [(1, 2, 3), (11, 12, 13)])
        self.assert_almost_equal(graph.find_variable("var_vector3f").get(ctxt), (1, 2, 3))
        self.assert_almost_equal(graph.find_variable("var_vector3f_array").get_array(ctxt), [(1, 2, 3), (11, 12, 13)])
        self.assert_almost_equal(graph.find_variable("var_vector3h").get(ctxt), (1, 2, 3))
        self.assert_almost_equal(graph.find_variable("var_vector3h_array").get_array(ctxt), [(1, 2, 3), (11, 12, 13)])

    # -------------------------------------------------------------------------
    async def test_variables_create_remove_stress_test(self):
        """Stress test the creation, removal and query of variables"""

        controller = og.Controller()
        keys = og.Controller.Keys
        (graph, _, _, _) = controller.edit(
            self._graph_path,
            {
                keys.CREATE_NODES: [
                    ("ReadVariable", "omni.graph.core.ReadVariable"),
                ],
            },
        )
        context = graph.get_default_graph_context()

        # get/set the active variable (FC)
        def active_set(var, val):
            nonlocal context
            var.set(context, val)

        def active_get(var):
            nonlocal context
            return var.get(context)

        # get/set the default value
        def default_set(var, val):
            og.Controller.set_variable_default_value(var, val)

        def default_get(var):
            return og.Controller.get_variable_default_value(var)

        # include setting the default value and getting the active value
        func_list = [(active_get, active_set), (default_get, default_set), (active_get, default_set)]

        # loop through default and active get, set values methods
        for (getter, setter) in func_list:
            for i in range(1, 10):
                variable = graph.create_variable("TestVariable", og.Type(og.BaseDataType.FLOAT))
                setter(variable, float(i))
                self.assertEqual(getter(variable), float(i))
                graph.remove_variable(variable)
                self.assertFalse(bool(variable))
                self.assertIsNone(graph.find_variable("TestVariable"))

    # -------------------------------------------------------------------------
    async def test_variables_write_update_on_graph(self):
        """Tests that the updating of variables occurs as expected"""

        # Create a graph that writes the give variable
        controller = og.Controller()
        keys = og.Controller.Keys

        # [OnTick] --> [WriteVariable(2)]
        (graph, _, _, _) = controller.edit(
            self._graph_path,
            {
                keys.CREATE_NODES: [
                    ("OnTick", "omni.graph.action.OnTick"),
                    ("WriteVariable", "omni.graph.core.WriteVariable"),
                    ("ConstInt", "omni.graph.nodes.ConstantInt"),
                ],
                keys.CONNECT: [
                    ("OnTick.outputs:tick", "WriteVariable.inputs:execIn"),
                    ("ConstInt.inputs:value", "WriteVariable.inputs:value"),
                ],
                keys.CREATE_VARIABLES: [("TestVariable", og.Type(og.BaseDataType.INT))],
                keys.SET_VALUES: [
                    ("OnTick.inputs:onlyPlayback", False),
                    ("ConstInt.inputs:value", 2),
                    ("WriteVariable.inputs:variableName", "TestVariable"),
                ],
            },
        )

        variable = graph.find_variable("TestVariable")
        og.Controller.set_variable_default_value(variable, 1)

        await controller.evaluate(graph)

        # USD value remains at 1
        self.assertEqual(og.Controller.get_variable_default_value(variable), 1)
        # Fabric value remains at 2
        self.assertEqual(variable.get(graph.get_default_graph_context()), 2)

        # change default value - variable value should change
        og.Controller.set_variable_default_value(variable, 3)
        self.assertEqual(og.Controller.get_variable_default_value(variable), 3)
        self.assertEqual(variable.get(graph.get_default_graph_context()), 3)

        await controller.evaluate(graph)

        # FC value should change again
        self.assertEqual(og.Controller.get_variable_default_value(variable), 3)
        self.assertEqual(variable.get(graph.get_default_graph_context()), 2)

    # -------------------------------------------------------------------------
    async def __create_instances(self, graph_path: str, prim_names: List[str]):
        """Helper method to create graph instances"""
        stage = omni.usd.get_context().get_stage()
        for prim_name in prim_names:
            prim = stage.DefinePrim(prim_name)
            OmniGraphSchemaTools.applyOmniGraphAPI(stage, prim_name, graph_path)
            self.assertTrue(prim.GetRelationship("omniGraphs").IsValid())

        # need to wait a frame for USD changes to apply. This could use an immediate
        # python API.
        await omni.kit.app.get_app().next_update_async()

    # -------------------------------------------------------------------------
    async def test_get_set_variable_instance_values(self):
        """Tests for retrieving and setting variable instance values"""

        # create the graph
        controller = og.Controller()
        keys = og.Controller.Keys
        (graph, _, _, _) = controller.edit(
            self._graph_path,
            {
                keys.CREATE_VARIABLES: [
                    ("float_var", og.Type(og.BaseDataType.FLOAT)),
                    ("token_var", og.Type(og.BaseDataType.TOKEN)),
                    ("array_var", og.Type(og.BaseDataType.INT, array_depth=1)),
                    ("tuple_var", og.Type(og.BaseDataType.FLOAT, 3, 0)),
                ],
            },
        )

        # create the instances
        prim_names = [f"/World/Prim_{i}" for i in range(1, 10)]
        await self.__create_instances(self._graph_path, prim_names)

        context = graph.get_default_graph_context()

        float_var = graph.find_variable("float_var")
        token_var = graph.find_variable("token_var")
        array_var = graph.find_variable("array_var")
        tuple_var = graph.find_variable("tuple_var")

        # set the default variable values. This should be promoted to each instance
        og.Controller.set_variable_default_value(float_var, -1.0)
        og.Controller.set_variable_default_value(token_var, "token_var")
        og.Controller.set_variable_default_value(array_var, [1.0, 1.0, 1.0, 1.0])
        og.Controller.set_variable_default_value(tuple_var, (1.0, 1.0, 1.0))

        # validate all the instances have the default value
        for prim_name in prim_names:
            self.assertEquals(float_var.get(context, instance_path=prim_name), -1.0)
            self.assertEquals(token_var.get(context, instance_path=prim_name), "token_var")
            self.assert_almost_equal(array_var.get_array(context, instance_path=prim_name), [1.0, 1.0, 1.0, 1.0])
            self.assert_almost_equal(tuple_var.get(context, instance_path=prim_name), (1.0, 1.0, 1.0))

        # set the variables
        i = 0
        for prim_name in prim_names:
            float_var.set(context, instance_path=prim_name, value=float(i))
            token_var.set(context, instance_path=prim_name, value=prim_name)
            array_var.set(context, instance_path=prim_name, value=[i, i + 1, i + 2, i + 3])
            tuple_var.set(context, instance_path=prim_name, value=(float(i), float(i + 1), float(i + 2)))
            i = i + 1

        # retrieve the variables
        i = 0
        for prim_name in prim_names:
            self.assertEquals(float_var.get(context, instance_path=prim_name), float(i))
            self.assertEquals(token_var.get(context, instance_path=prim_name), prim_name)
            self.assert_almost_equal(array_var.get_array(context, instance_path=prim_name), [i, i + 1, i + 2, i + 3])
            self.assert_almost_equal(
                tuple_var.get(context, instance_path=prim_name), (float(i), float(i + 1), float(i + 2))
            )
            i = i + 1

    # -------------------------------------------------------------------------
    async def test_invalid_variable_access_throws(self):
        """Tests that invalid variable access throws errors"""

        # create the graph
        controller = og.Controller()
        keys = og.Controller.Keys
        (graph, _, _, _) = controller.edit(
            self._graph_path,
            {
                keys.CREATE_VARIABLES: [
                    ("float_var", og.Type(og.BaseDataType.FLOAT)),
                    ("array_var", og.Type(og.BaseDataType.FLOAT, array_depth=1)),
                    ("invalid_var", og.Type(og.BaseDataType.FLOAT)),
                ]
            },
        )

        # create an instance
        context = graph.get_default_graph_context()
        valid_prim_name = "/World/ValidInstance"
        invalid_instance_name = "/World/InvalidInstance"
        await self.__create_instances(self._graph_path, [valid_prim_name])

        float_var = graph.find_variable("float_var")
        array_var = graph.find_variable("array_var")
        invalid_var = graph.find_variable("invalid_var")

        graph.remove_variable(invalid_var)

        og.Controller.set_variable_default_value(float_var, 1.0)

        self.assertEquals(float_var.get(context), 1.0)
        self.assertEquals(float_var.get(context, instance_path=valid_prim_name), 1.0)

        # invalid variable
        with self.assertRaises(TypeError):
            invalid_var.get(context)

        # invalid variable and instance
        with self.assertRaises(TypeError):
            invalid_var.get(context, instance_path=invalid_instance_name)

        # invalid instance, valid variable
        with self.assertRaises(TypeError):
            float_var.get(context, instance_path=invalid_instance_name)

        # invalid instance with array variable
        with self.assertRaises(TypeError):
            array_var.get(context, instance_path=invalid_instance_name)

    # -------------------------------------------------------------------------
    def _get_ogtypes_for_testing(self) -> Set[og.Type]:
        """Get a list of types to be used for testing."""
        numerics = og.AttributeType.get_unions()["numerics"]
        type_set = {og.AttributeType.type_from_ogn_type_name(n_type) for n_type in numerics}.union(
            {
                og.Type(og.BaseDataType.UCHAR, 1, 1, og.AttributeRole.TEXT),
                og.Type(og.BaseDataType.TOKEN, 1, 0, og.AttributeRole.NONE),
            }
        )
        # transform converts to matrix when used in a variable
        type_set.remove(og.Type(og.BaseDataType.DOUBLE, 16, 1, og.AttributeRole.TRANSFORM))
        type_set.remove(og.Type(og.BaseDataType.DOUBLE, 16, 0, og.AttributeRole.TRANSFORM))
        return type_set

    # -------------------------------------------------------------------------
    async def test_changevariabletype_command(self):
        """
        Validates the ChangeVariableType Command property changes the type of the variable
        """
        controller = og.Controller()
        keys = og.Controller.Keys
        (graph, _, _, _) = controller.edit(
            self._graph_path,
            {
                keys.CREATE_VARIABLES: [
                    ("var", og.Type(og.BaseDataType.FLOAT)),
                ]
            },
        )

        types = self._get_ogtypes_for_testing()
        variable = graph.find_variable("var")

        for og_type in types:
            og.cmds.ChangeVariableType(variable=variable, variable_type=og_type)
            self.assertEqual(variable.type, og_type)

    # -------------------------------------------------------------------------
    async def test_changevariabletype_command_undo_redo(self):
        """
        Validates the change variable type command applies and restores default values
        as expected with undo and redo
        """
        controller = og.Controller()
        keys = og.Controller.Keys
        original_type = og.Type(og.BaseDataType.FLOAT)
        (graph, _, _, _) = controller.edit(
            self._graph_path,
            {
                keys.CREATE_VARIABLES: [
                    ("var", original_type),
                ]
            },
        )

        types = self._get_ogtypes_for_testing()
        types.remove(original_type)

        variable = graph.find_variable("var")
        default_value = 5.0
        og.Controller.set_variable_default_value(variable, default_value)

        # validate not changing type doesn't erase the default value
        og.cmds.ChangeVariableType(variable=variable, variable_type=original_type)
        self.assertEqual(variable.type, original_type)
        self.assertEqual(og.Controller.get_variable_default_value(variable), default_value)

        for og_type in types:
            og.cmds.ChangeVariableType(variable=variable, variable_type=og_type)
            self.assertEqual(variable.type, og_type)
            self.assertIsNone(og.Controller.get_variable_default_value(variable))

            # on undo, when the type changes, the default value is removed
            omni.kit.undo.undo()
            self.assertEqual(variable.type, original_type)
            self.assertEqual(og.Controller.get_variable_default_value(variable), default_value)

            # on redo, the type and default value are restored
            omni.kit.undo.redo()
            self.assertEqual(variable.type, og_type)
            self.assertIsNone(og.Controller.get_variable_default_value(variable))

            omni.kit.undo.undo()

    # -------------------------------------------------------------------------
    async def test_changevariabletype_command_on_separate_layer(self):
        """
        Validates the behaviour of change variable type command on different
        layers
        """
        usd_context = omni.usd.get_context()
        stage = usd_context.get_stage()
        root_layer = stage.GetRootLayer()
        identifier1 = LayerUtils.create_sublayer(root_layer, 0, "").identifier
        identifier2 = LayerUtils.create_sublayer(root_layer, 1, "").identifier
        identifier3 = LayerUtils.create_sublayer(root_layer, 2, "").identifier
        LayerUtils.set_edit_target(stage, identifier2)

        # create a graph on the middle layer
        controller = og.Controller()
        keys = og.Controller.Keys
        original_type = og.Type(og.BaseDataType.FLOAT)
        (graph, _, _, _) = controller.edit(
            self._graph_path,
            {
                keys.CREATE_VARIABLES: [
                    ("var", original_type),
                ]
            },
        )

        # validate the type can be changed
        variable = graph.find_variable("var")
        og_type = og.Type(og.BaseDataType.INT)
        og.cmds.ChangeVariableType(variable=variable, variable_type=og_type)
        self.assertEqual(variable.type, og_type)

        # change the edit layer to a type with a stronger opinion and validate the change still works
        LayerUtils.set_edit_target(stage, identifier1)
        variable = graph.find_variable("var")
        og_type = og.Type(og.BaseDataType.DOUBLE)
        og.cmds.ChangeVariableType(variable=variable, variable_type=og_type)
        self.assertEqual(variable.type, og_type)

        # change the edit layer to a type with a weaker opinion and validate the change fails
        LayerUtils.set_edit_target(stage, identifier3)
        variable = graph.find_variable("var")
        og_type = og.Type(og.BaseDataType.INT, 2)
        omni.kit.commands.set_logging_enabled(False)
        (_, succeeded) = og.cmds.ChangeVariableType(variable=variable, variable_type=og_type)
        omni.kit.commands.set_logging_enabled(True)
        self.assertFalse(succeeded)

    # -----------------------------------------------------------------------------------------
    async def test_changevariabletype_changes_resolution(self):
        """
        Validate that changing a variable type changes the resolution of node types
        """
        controller = og.Controller()
        keys = og.Controller.Keys
        original_type = og.Type(og.BaseDataType.FLOAT)
        (graph, nodes, _, _) = controller.edit(
            self._graph_path,
            {
                keys.CREATE_NODES: [
                    ("ReadVariable", "omni.graph.core.ReadVariable"),
                ],
                keys.CREATE_VARIABLES: [
                    ("var", original_type),
                ],
                keys.SET_VALUES: [
                    ("ReadVariable.inputs:variableName", "var"),
                ],
            },
        )

        node = nodes[0]
        variable = graph.find_variable("var")
        await og.Controller.evaluate(graph)
        self.assertEqual(node.get_attribute("outputs:value").get_resolved_type(), original_type)

        og_type = og.Type(og.BaseDataType.INT, 2)
        omni.kit.commands.set_logging_enabled(False)
        await og.Controller.evaluate(graph)
        og.cmds.ChangeVariableType(variable=variable, variable_type=og_type)
        self.assertEqual(node.get_attribute("outputs:value").get_resolved_type(), og_type)
