"""Tests related to the helper scripts"""
import json
import re
from contextlib import suppress
from pathlib import Path
from typing import Dict, Optional, Union

import omni.graph.core as og
import omni.graph.core.tests as ogts
import omni.graph.tools as ogt
import omni.graph.tools.ogn as ogn
import omni.kit
import omni.usd
from omni.graph.core.typing import NodeType_t
from pxr import Sdf, Usd

# ==============================================================================================================
# Iteration helpers
GRAPH_BY_GRAPH = "og.Graph"
GRAPH_BY_PATH = "path"
GRAPH_BY_SDF_PATH = "Sdf path"
GRAPH_NONE = "None"
GRAPH_LOOKUPS = [GRAPH_BY_GRAPH, GRAPH_BY_PATH, GRAPH_BY_SDF_PATH, GRAPH_NONE]

NODE_BY_NODE = "og.Node"
NODE_BY_PATH = "path"
NODE_BY_USD_PRIM = "Usd.Prim"
NODE_NONE = "None"
NODE_BY_TUPLE = "tuple"
NODE_LOOKUPS = [NODE_BY_NODE, NODE_BY_PATH, NODE_BY_USD_PRIM, NODE_NONE, NODE_BY_TUPLE]

NODE_TYPE_BY_NODE = "og.Node"
NODE_TYPE_BY_USD_PRIM = "Usd.Prim"
NODE_TYPE_BY_NAME = "name"
NODE_TYPE_BY_NODE_TYPE = "og.NodeType"
NODE_TYPE_LOOKUPS = [NODE_TYPE_BY_NODE, NODE_TYPE_BY_NAME, NODE_TYPE_BY_USD_PRIM, NODE_TYPE_BY_NODE_TYPE]

ATTRIBUTE_BY_ATTRIBUTE = "og.Attribute"
ATTRIBUTE_BY_PATH = "path"
ATTRIBUTE_BY_SDF_PATH = "Sdf.Path"
ATTRIBUTE_BY_USD_ATTRIBUTE = "Usd.Attribute"
ATTRIBUTE_BY_TUPLE = "tuple"
ATTRIBUTE_LOOKUPS = [
    ATTRIBUTE_BY_ATTRIBUTE,
    ATTRIBUTE_BY_PATH,
    ATTRIBUTE_BY_SDF_PATH,
    ATTRIBUTE_BY_USD_ATTRIBUTE,
    ATTRIBUTE_BY_TUPLE,
]


# ==============================================================================================================
class TestHelpers(ogts.OmniGraphTestCase):
    """Testing to ensure OGN helpers work"""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.outer_graph1_path = None
        self.outer_graph2_path = None
        self.inner_graph1_1_path = None
        self.inner_graph1_2_path = None
        self.inner_graph2_1_path = None
        self.inner_graph2_2_path = None
        self.graph_paths = None
        self.node1_path = None
        self.node2_path = None
        self.node3_path = None
        self.node4_path = None
        self.node5_path = None
        self.node1_1_path = None
        self.node1_2_path = None
        self.node2_1_path = None
        self.node2_2_path = None
        self.node_paths = None
        self.attribute1_path = None
        self.attribute2_path = None
        self.attribute1_1_path = None
        self.attribute1_2_path = None
        self.attribute2_1_path = None
        self.attribute2_2_path = None
        self.attribute1_target_input_path = None
        self.attribute1_target_output_path = None
        self.attribute1_bundle_input_path = None
        self.attribute_paths = None
        self.variable_bool_name = None
        self.variable_float_name = None
        self.variable_double_name = None
        self.variable_names = None

    # --------------------------------------------------------------------------------------------------------------
    async def __populate_data(self):
        """
        Read in the test file and populate the local variables to use for testing.
        Note that this test file has graph prims whose parents are other graph
        prims, i.e., the old subgraphs. While these are no longer supported (will not
        evaluate), we still keep this test file as-is in to ensure that the helper
        scripts continue to operate correctly even in situations that, while technically
        authorable, are no longer officially supported in OG.
        """
        (result, error) = await ogts.load_test_file("TestObjectIdentification.usda", use_caller_subdirectory=True)
        self.assertTrue(result, error)

        self.outer_graph1_path = "/World/OuterGraph1"
        self.outer_graph2_path = "/World/OuterGraph2"
        self.inner_graph1_1_path = f"{self.outer_graph1_path}/InnerGraph1"
        self.inner_graph1_2_path = f"{self.outer_graph1_path}/InnerGraph2"
        self.inner_graph2_1_path = f"{self.outer_graph2_path}/InnerGraph1"
        self.inner_graph2_2_path = f"{self.outer_graph2_path}/InnerGraph2"
        self.graph_paths = [
            self.outer_graph1_path,
            self.outer_graph2_path,
            self.inner_graph1_1_path,
            self.inner_graph1_2_path,
            self.inner_graph2_1_path,
            self.inner_graph2_2_path,
        ]

        self.node1_path = f"{self.outer_graph1_path}/constant_bool"
        self.node2_path = f"{self.outer_graph2_path}/constant_double"
        self.node3_path = f"{self.outer_graph1_path}/constant_prims"
        self.node4_path = f"{self.outer_graph1_path}/get_prims_at_path"
        self.node5_path = f"{self.outer_graph1_path}/extract_bundle"
        self.node1_1_path = f"{self.inner_graph1_1_path}/constant_float"
        self.node1_2_path = f"{self.inner_graph1_2_path}/constant_int"
        self.node2_1_path = f"{self.inner_graph2_1_path}/constant_uint"
        self.node2_2_path = f"{self.inner_graph2_2_path}/constant_uint64"
        self.node_paths = [
            self.node1_path,
            self.node2_path,
            self.node3_path,
            self.node4_path,
            self.node5_path,
            self.node1_1_path,
            self.node1_2_path,
            self.node2_1_path,
            self.node2_2_path,
        ]

        self.attribute1_path = f"{self.node1_path}.inputs:value"
        self.attribute2_path = f"{self.node2_path}.inputs:value"
        self.attribute1_1_path = f"{self.node1_1_path}.inputs:value"
        self.attribute1_2_path = f"{self.node1_2_path}.inputs:value"
        self.attribute2_1_path = f"{self.node2_1_path}.inputs:value"
        self.attribute2_2_path = f"{self.node2_2_path}.inputs:value"
        # additional attributes of relationship types
        self.attribute1_target_input_path = f"{self.node3_path}.inputs:value"
        self.attribute1_target_output_path = f"{self.node4_path}.outputs:prims"
        self.attribute1_bundle_input_path = f"{self.node5_path}.inputs:bundle"

        self.attribute_paths = [
            self.attribute1_path,
            self.attribute2_path,
            self.attribute1_1_path,
            self.attribute1_2_path,
            self.attribute2_1_path,
            self.attribute2_2_path,
            self.attribute1_target_input_path,
            self.attribute1_target_output_path,
            self.attribute1_bundle_input_path,
        ]

        self.variable_bool_name = "variable_bool"
        self.variable_float_name = "variable_float"
        self.variable_double_name = "variable_double"
        self.variable_names = [
            self.variable_bool_name,
            self.variable_float_name,
            self.variable_double_name,
        ]

    # --------------------------------------------------------------------------------------------------------------
    @staticmethod
    def __graph_id_from_path(graph_path: str, lookup_type: str) -> Union[str, Sdf.Path, og.Graph, None]:
        """Return a typed graph lookup value based on the lookup requested"""
        if lookup_type == GRAPH_BY_GRAPH:
            return og.get_graph_by_path(graph_path)
        if lookup_type == GRAPH_BY_PATH:
            return graph_path
        if lookup_type == GRAPH_BY_SDF_PATH:
            return Sdf.Path(graph_path)
        return None

    # --------------------------------------------------------------------------------------------------------------
    @staticmethod
    def __node_id_from_path(node_path: str, lookup_type: str) -> Union[str, Usd.Prim, og.Node, None]:
        """Return a typed node lookup value based on the lookup requested"""
        if lookup_type == NODE_BY_NODE:
            return og.get_node_by_path(node_path)
        if lookup_type == NODE_BY_PATH:
            return node_path
        if lookup_type == NODE_BY_USD_PRIM:
            return omni.usd.get_context().get_stage().GetPrimAtPath(node_path)
        return None

    # --------------------------------------------------------------------------------------------------------------
    @staticmethod
    def __node_type_id_from_info(type_name: str, node_type: og.NodeType, node: og.Node, lookup_type: str) -> NodeType_t:
        """Return a typed node type lookup value based on the lookup requested"""
        if lookup_type == NODE_TYPE_BY_NODE:
            return node
        if lookup_type == NODE_TYPE_BY_NODE_TYPE:
            return node_type
        if lookup_type == NODE_TYPE_BY_NAME:
            return og.get_node_type(type_name).get_node_type()
        if lookup_type == NODE_TYPE_BY_USD_PRIM:
            return omni.usd.get_context().get_stage().GetPrimAtPath(node.get_prim_path())
        return None

    # --------------------------------------------------------------------------------------------------------------
    @staticmethod
    def __attribute_id_from_path(
        attribute_path: str, attribute_type: str, node: Optional[og.Node] = None
    ) -> Union[str, Usd.Attribute, og.Attribute, None]:
        """Return a typed attribute lookup value based on the lookup requested.

        Args:
            attribute_path: Absolute (if node is None) or relative (if node is not None) path to attribute
            attribute_type: ID type to return that represents the attribute
            node: Optional node to which the attribute belongs

        Returns:
            ID type requested that corresponds to the attribute that was found

        Raises:
            og.OmniGraphError if the attribute configuration information does not provide a unique lookup location
        """
        (node_path, attribute_relative_path) = attribute_path.split(".")
        if not attribute_relative_path:
            # No separator means the path is a relative attribute path and the node must be specified, and contain
            # the attribute passed in.
            if node is None:
                raise og.OmniGraphError(f"Relative attribute path {attribute_path} requires a node to resolve")
            attribute_relative_path = node_path
            node_path = node.get_prim_path()
        else:
            node_lookup = og.get_node_by_path(node_path)
            try:
                node_lookup_handle = node_lookup.get_handle() if node_lookup is not None else None
            except AttributeError as error:
                raise og.OmniGraphError(
                    f"Node extracted from attribute path '{attribute_path}' was not a valid og.Node"
                ) from error

            # If a node was specified, and was also a part of the attribute path, make sure they match
            if node is not None:
                try:
                    node_handle = node.get_handle()
                except AttributeError as error:
                    raise og.OmniGraphError(f"Node description passed in ({node}) was not a valid og.Node") from error

                if node_lookup is None or node_handle != node_lookup_handle:
                    raise og.OmniGraphError(
                        f"Node extracted from attribute path '{attribute_path}' "
                        "did not match the node passed in {node.get_prim_path()}"
                    )
            else:
                node = node_lookup

        if attribute_type == ATTRIBUTE_BY_ATTRIBUTE:
            if not node.get_attribute_exists(attribute_relative_path):
                raise og.OmniGraphError(f"No attribute {attribute_relative_path} exists on node {node.get_prim_path()}")
            return node.get_attribute(attribute_relative_path)

        if attribute_type == ATTRIBUTE_BY_PATH:
            return f"{node_path}.{attribute_relative_path}"

        if attribute_type == ATTRIBUTE_BY_SDF_PATH:
            return Sdf.Path(f"{node_path}.{attribute_relative_path}")

        if attribute_type == ATTRIBUTE_BY_USD_ATTRIBUTE:
            return (
                omni.usd.get_context().get_stage().GetPropertyAtPath(Sdf.Path(f"{node_path}.{attribute_relative_path}"))
            )

        return None

    # --------------------------------------------------------------------------------------------------------------
    def __test_graph_lookup(self, graphs: Dict[str, og.Graph]):
        """Run the test on this method that looks up a graph from identifying information.

            def graph(cls, graph_id: GraphSpecs_t) -> Union[og.Graph, List[og.Graph]]:

        Args:
            graphs: Dictionary of path:graph for expected graphs being looked up
        """
        # Test graph lookup
        for path, graph in graphs.items():
            for lookup in GRAPH_LOOKUPS:
                graph_lookup = og.ObjectLookup.graph(self.__graph_id_from_path(path, lookup))
                lookup_msg = f"Looking up graph {path} by {lookup}"
                if lookup == GRAPH_NONE:
                    self.assertIsNone(graph_lookup, lookup_msg)
                else:
                    self.assertIsNotNone(graph_lookup, lookup_msg)
                    self.assertEqual(graph.get_handle(), graph_lookup.get_handle(), lookup_msg)

        # Test list of graph lookup
        graph_paths = list(graphs.keys())
        all_graphs = og.ObjectLookup.graph(graph_paths)
        for index, path in enumerate(graph_paths):
            graph_lookup = all_graphs[index]
            self.assertIsNotNone(graph_lookup, f"List-based lookup of graph {path}")
            self.assertEqual(
                graph_lookup.get_handle(), graphs[path].get_handle(), f"Lookup graph by list, item {index} = {path}"
            )

        # Test failing graph lookups
        self.assertIsNone(og.ObjectLookup.graph(None), "Looking up None graph")
        self.assertIsNone(og.ObjectLookup.graph("/NoGraph"), "Looking up unknown graph path")
        self.assertIsNone(og.ObjectLookup.graph(self.node1_path), "Looking up non-graph path")
        self.assertIsNone(og.ObjectLookup.graph(Sdf.Path("/NoGraph")), "Looking up unknown graph path")
        self.assertIsNone(og.ObjectLookup.graph(Sdf.Path(self.node1_path)), "Looking up non-graph Sdf path")

    # --------------------------------------------------------------------------------------------------------------
    def __test_node_lookup(self, nodes: Dict[str, og.Node]):
        """Run the set of tests to look up nodes from identifying information using these API functions.

            def node(cls, node_id: NodeSpecs_t, graph: GraphSpec_t = None) -> Union[og.Node, List[og.Node]]:
            def prim(cls, node_id: NodeSpecs_t, graph: GraphSpec_t = None) -> Union[Usd.Prim, List[Usd.Prim]]:
            def node_path(cls, node_id: NodeSpecs_t, graph: GraphSpec_t = None) -> Union[str, List[str]]:

        Args:
            nodes: Dictionary of path:node for expected nodes being looked up
        """
        # Test various node lookups
        for graph_lookup_type in GRAPH_LOOKUPS:
            for path, node in nodes.items():
                graph_path = node.get_graph().get_path_to_graph()
                graph_id = self.__graph_id_from_path(graph_path, graph_lookup_type)
                for lookup in NODE_LOOKUPS:
                    # Looking up a node without a node identifier is not possible
                    if lookup == NODE_NONE:
                        continue
                    # If a node spec requires a graph and there isn't one then skip it (it will be tested below)
                    if lookup in [NODE_BY_PATH, NODE_BY_TUPLE] and graph_id is None:
                        continue
                    if lookup == NODE_BY_TUPLE:
                        (graph, node_path) = og.ObjectLookup.split_graph_from_node_path(path)
                        node_lookup = og.ObjectLookup.node((node_path, graph))
                    else:
                        node_id = self.__node_id_from_path(path, lookup)
                        node_lookup = og.ObjectLookup.node(node_id, graph_id)
                    lookup_msg = (
                        f"Looking up node {path} by {lookup} in graph {graph_path} found by {graph_lookup_type}"
                    )
                    if lookup == NODE_NONE:
                        self.assertIsNone(node_lookup, lookup_msg)
                    else:
                        self.assertIsNotNone(node_lookup, lookup_msg)
                        self.assertEqual(node.get_handle(), node_lookup.get_handle(), lookup_msg)

                    if graph_lookup_type == GRAPH_NONE:
                        prim_lookup = og.ObjectLookup.prim(node_id)
                        if lookup == NODE_NONE:
                            self.assertIsNone(prim_lookup, lookup_msg)
                        else:
                            self.assertIsNotNone(prim_lookup, lookup_msg)
                            self.assertEqual(node.get_prim_path(), prim_lookup.GetPrimPath(), lookup_msg)

                        node_path_lookup = og.ObjectLookup.node_path(node_id)
                        if lookup == NODE_NONE:
                            self.assertIsNone(node_path_lookup, lookup_msg)
                        else:
                            self.assertIsNotNone(node_path_lookup, lookup_msg)
                            self.assertEqual(node.get_prim_path(), node_path_lookup, lookup_msg)

        # Test list of node lookups
        node_paths = list(nodes.keys())
        all_nodes = og.ObjectLookup.node(node_paths)
        for index, path in enumerate(node_paths):
            node_lookup = all_nodes[index]
            self.assertIsNotNone(node_lookup, f"List-based lookup of node {path}")
            self.assertEqual(
                node_lookup.get_handle(), nodes[path].get_handle(), f"Lookup by node, item {index} = {path}"
            )

        # Test failing node lookups
        with self.assertRaises(og.OmniGraphValueError):
            og.ObjectLookup.node(None)
        for parameters in [
            ["/NoNode"],  # Non-existent path
            [self.inner_graph2_1_path],  # Non-node path
            [self.node1_path, og.ObjectLookup.graph(self.outer_graph2_path)],  # Mismatched graph
        ]:
            with self.assertRaises(og.OmniGraphError):
                og.ObjectLookup.node(*parameters)
        with self.assertRaises(og.OmniGraphError):
            og.ObjectLookup.prim("/NotAPrim")

    # --------------------------------------------------------------------------------------------------------------
    def __test_node_type_lookup(self, nodes: Dict[str, og.Node]):
        """Run the set of tests to look up node types from identifying information using this API.

            def node_type(cls, type_id: NodeTypes_t) -> Union[og.NodeType, List[og.NodeType]]:

        Args:
            nodes: Dictionary of path:node for expected nodes being looked up
        """
        node_types = [(node.get_node_type().get_node_type(), node.get_node_type(), node) for node in nodes.values()]
        for name, expected_node_type, node in node_types:
            self.assertTrue(expected_node_type.is_valid(), f"Checking lookup of node type {name}")
            for lookup in NODE_TYPE_LOOKUPS:
                lookup_msg = f"Looking up node type {name} by {lookup}"
                node_type_lookup = og.ObjectLookup.node_type(
                    self.__node_type_id_from_info(name, expected_node_type, node, lookup)
                )
                self.assertIsNotNone(node_type_lookup, lookup_msg)
                self.assertTrue(node_type_lookup, lookup_msg)

        # Test list of node type lookups
        all_lookups = [name for name, _, _ in node_types]
        all_lookups += [node_type for _, node_type, _ in node_types]
        all_lookups += [node for _, _, node in node_types]
        all_lookups += [og.ObjectLookup.prim(node) for _, _, node in node_types]
        all_node_types = og.ObjectLookup.node_type(all_lookups)
        node_type_count = len(node_types)
        for index, (_, node_type, _) in enumerate(node_types):
            msg = f"Lookup of {all_lookups[index]} by"
            self.assertEqual(all_node_types[index], node_type, f"{msg} name")
            self.assertEqual(all_node_types[index + node_type_count], node_type, f"{msg} node type")
            self.assertEqual(all_node_types[index + node_type_count * 2], node_type, f"{msg} node")
            self.assertEqual(all_node_types[index + node_type_count * 3], node_type, f"{msg} prim")

        # Test failing node type lookups
        with self.assertRaises(og.OmniGraphError):
            og.ObjectLookup.node_type(None)
        with self.assertRaises(og.OmniGraphError):
            og.ObjectLookup.node_type("omni.this.type.is.invalid")
        with self.assertRaises(og.OmniGraphError):
            og.ObjectLookup.node_type(og.get_node_by_path("/This/Is/Not/A/Node"))
        with self.assertRaises(og.OmniGraphError):
            og.ObjectLookup.node_type(og.get_node_type("omni.this.type.is.invalid"))

    # --------------------------------------------------------------------------------------------------------------
    def __test_variable_lookup(self):
        """Run the set of tests for running variable lookup from identifying information

        def variable(cls, variable_id: Variables_t) -> Union[og.IVariable, List[og.IVariable]]:
        """
        graph_with_variables = og.ObjectLookup.graph(self.outer_graph1_path)
        graph_without_variables = og.ObjectLookup.graph(self.outer_graph2_path)

        variables = [graph_with_variables.find_variable(x) for x in self.variable_names]
        variable_tuples = [(graph_with_variables, x) for x in self.variable_names]

        # Test lookup of variable types
        for v in variables:
            self.assertTrue(bool(v))
            self.assertEquals(v, og.ObjectLookup.variable(v))
            self.assertEquals(v, og.ObjectLookup.variable((graph_with_variables, v.name)))
            self.assertEquals(v, og.ObjectLookup.variable(v.source_path))
            self.assertEquals(v, og.ObjectLookup.variable(Sdf.Path(v.source_path)))

            # Test lookup with wrong graph
            with self.assertRaises(og.OmniGraphError):
                og.ObjectLookup.variable((graph_without_variables, v.name))

            # Test lookup with wrong name
            with self.assertRaises(og.OmniGraphError):
                og.ObjectLookup.variable((graph_with_variables, "Variable4"))

        # Test list lookup
        self.assertEquals(variables, og.ObjectLookup.variable(variables))
        self.assertEquals(variables, og.ObjectLookup.variable(variable_tuples))

        with self.assertRaises(og.OmniGraphError):
            og.ObjectLookup.variable(None)
        with self.assertRaises(og.OmniGraphError):
            og.ObjectLookup.variable("Invalid string")
        with self.assertRaises(og.OmniGraphError):
            og.ObjectLookup.variable([variables[0], "Invalid Entry"])

    # --------------------------------------------------------------------------------------------------------------
    def __test_attribute_lookup(self):
        """Run the set of tests to look up attributes from identifying information using this API.
        def attribute(cls, attribute_id: AttributeSpecs_t, node: NodeSpec_t = None, graph: GraphSpec_t = None)
            -> Union[og.Attribute, List[og.Attribute]]:
        def usd_attribute(cls, attribute_id: AttributeSpecs_t) -> Union[Usd.Attribute, List[Usd.Attribute]]:
        """
        # Dictionary mapping the attribute path as a string to the actual attribute
        attributes = {
            attribute_path: self.__attribute_id_from_path(attribute_path, ATTRIBUTE_BY_ATTRIBUTE)
            for attribute_path in self.attribute_paths
        }
        # Dictionary mapping the attribute path as a string to the corresponding Usd.Attribute
        stage = omni.usd.get_context().get_stage()
        usd_attributes = {
            attribute_path: stage.GetPropertyAtPath(attribute_path) for attribute_path in self.attribute_paths
        }

        # Test various attribute lookups
        for graph_lookup_type in GRAPH_LOOKUPS:  # noqa: PLR1702
            for node_lookup_type in NODE_LOOKUPS:
                for path, attribute in attributes.items():
                    graph_path = attribute.get_node().get_graph().get_path_to_graph()
                    graph_lookup_msg = f"graph {graph_path} found by {graph_lookup_type}"
                    graph_lookup = self.__graph_id_from_path(graph_path, graph_lookup_type)

                    node_path = attribute.get_node().get_prim_path()
                    node_lookup_msg = f"in node {node_path} found by {node_lookup_type}"
                    node_lookup = self.__node_id_from_path(node_path, node_lookup_type)

                    for lookup in ATTRIBUTE_LOOKUPS:
                        try:
                            node = og.ObjectLookup.node(node_lookup)
                        except og.OmniGraphValueError:
                            node = None
                        if lookup == ATTRIBUTE_BY_TUPLE:
                            # The tuple lookup only works with valid nodes
                            if node is None:
                                continue
                            attribute_lookup = og.ObjectLookup.attribute((attribute.get_name(), node))
                            usd_property_lookup = og.ObjectLookup.usd_property((attribute.get_name(), node))
                        else:
                            attribute_id = self.__attribute_id_from_path(path, lookup, node)
                            attribute_lookup = og.ObjectLookup.attribute(attribute_id, node_lookup, graph_lookup)
                            if node_lookup is None and graph_lookup is None:
                                usd_property_lookup = og.ObjectLookup.usd_property(attribute_id)
                            else:
                                # We have to catch the case of the lookup being None so use a different value to
                                # flag the fact that this combination doesn't require testing of Usd.Attribute lookup
                                usd_property_lookup = ""
                        lookup_msg = f"Looking up attribute {path} by {lookup} {node_lookup_msg} {graph_lookup_msg}"
                        self.assertIsNotNone(usd_property_lookup, lookup_msg)
                        self.assertEqual(attribute.get_handle(), attribute_lookup.get_handle(), lookup_msg)

                        lookup_msg = lookup_msg.replace("attribute", "Usd.Attribute")
                        usd_attribute = usd_attributes[path]
                        self.assertIsNotNone(attribute_lookup, lookup_msg)
                        if usd_property_lookup != "":
                            self.assertEqual(usd_attribute.GetPath(), usd_property_lookup.GetPath(), lookup_msg)

        # Test list of attribute lookups
        attribute_paths = list(attributes.keys())
        all_attributes = og.ObjectLookup.attribute(attribute_paths)
        for index, path in enumerate(attribute_paths):
            attribute_lookup = all_attributes[index]
            self.assertIsNotNone(attribute_lookup, f"List-based lookup of attribute {path}")
            self.assertEqual(
                attribute_lookup.get_handle(),
                attributes[path].get_handle(),
                f"Lookup by attribute, item {index} = {path}",
            )

        # Test failing attribute lookups
        with self.assertRaises(og.OmniGraphError):
            og.ObjectLookup.attribute(None)  # None is illegal
        with self.assertRaises(og.OmniGraphError):
            og.ObjectLookup.attribute(f"{self.node1_path}.FooBar")  # Non-existent attribute on legal node
        with self.assertRaises(og.OmniGraphError):
            og.ObjectLookup.attribute(self.inner_graph2_1_path)  # Non-attribute path
        with self.assertRaises(og.OmniGraphError):
            og.ObjectLookup.attribute(self.attribute1_path, self.node1_1_path)  # Mismatched node

        with self.assertRaises(og.OmniGraphError):
            og.ObjectLookup.usd_attribute(None)  # None is illegal
        with self.assertRaises(og.OmniGraphError):
            og.ObjectLookup.usd_attribute(f"{self.node1_path}.FooBar")  # Non-existent attribute on legal node
        with self.assertRaises(og.OmniGraphError):
            og.ObjectLookup.usd_attribute(self.inner_graph2_1_path)  # Non-attribute path
        with self.assertRaises(og.OmniGraphError):
            og.ObjectLookup.usd_attribute((self.attribute1_path, self.node1_1_path))  # Mismatched node

    # --------------------------------------------------------------------------------------------------------------
    async def test_object_identification(self):
        """Test operation of the utility class that identifies nodes, attributes, and graphs from a variety of inputs.

        This is the structure of the relevant parts of the test file:

            def Xform "World"
                def OmniGraph "OuterGraph1"
                    custom bool graph:variable:var_bool
                    def OmniGraph "InnerGraph1"
                        def OmniGraphNode "constant_float"
                            custom float inputs:value
                    def OmniGraph "InnerGraph2"
                        def OmniGraphNode "constant_int"
                            custom int inputs:value
                    def OmniGraphNode "constant_bool"
                            custom bool inputs:value
                def OmniGraph "OuterGraph2"
                    def OmniGraphNode "constant_double"
                        custom double inputs:value
                    def ComputeGraph "InnerGraph1"
                        def OmniGraphNode "constant_uint"
                            custom uint inputs:value
                    def ComputeGraph "InnerGraph2"
                        def OmniGraphNode "constant_uint64"
                            custom uint64 inputs:value
        """
        await self.__populate_data()

        # Dictionary mapping the graph path as a string to the actual graph
        graphs = {graph_path: self.__graph_id_from_path(graph_path, GRAPH_BY_GRAPH) for graph_path in self.graph_paths}
        # Make sure all of the expected graphs from the file were found
        for path, graph in graphs.items():
            self.assertIsNotNone(graph, f"Graph at {path}")

        # Dictionary mapping the node path as a string to the actual node
        nodes = {node_path: self.__node_id_from_path(node_path, NODE_BY_NODE) for node_path in self.node_paths}

        # Make sure all of the excpected node paths from the file were found
        for path, node in nodes.items():
            self.assertIsNotNone(node, f"Node at {path}")
        # Make sure the nodes belong to their expected graph
        nodes_graph = {node: node.get_graph() for node in nodes.values()}
        for node, graph in nodes_graph.items():
            self.assertIsNotNone(graph, f"Graph of node {node.get_prim_path()}")
            node_lookup = graph.get_node(node.get_prim_path())
            self.assertIsNotNone(node_lookup, f"Lookup in graph of node {node.get_prim_path()}")
            self.assertEqual(node.get_handle(), node_lookup.get_handle(), f"Node {node.get_prim_path()} matches lookup")

            self.__test_graph_lookup(graphs)
            self.__test_node_lookup(nodes)
            self.__test_attribute_lookup()
            self.__test_node_type_lookup(nodes)
            self.__test_variable_lookup()

    # --------------------------------------------------------------------------------------------------------------
    async def test_attribute_type_identification(self):
        """The attribute type ID does not rely on file contents so it can be tested in isolation."""

        # AttributeType_t = Union[str, og.Type]
        # def attribute_type(cls, type_id: AttributeType_t) -> og.Type:
        attribute_types = {
            type_name: og.AttributeType.type_from_ogn_type_name(type_name)
            for type_name in ogn.supported_attribute_type_names()
        }

        # The list of all legal types
        for type_name, attr_type in attribute_types.items():
            type_from_string = og.ObjectLookup.attribute_type(type_name)
            self.assertEqual(attr_type, type_from_string, f"Type from string {type_name}")

            type_from_type = og.ObjectLookup.attribute_type(attr_type)
            self.assertEqual(attr_type, type_from_type, f"Type from string {type_name}")

        # Illegal type names and types
        for illegal_type in [
            "dubble",
            "double[5]",
            og.Type(og.BaseDataType.ASSET),
            og.Type(og.BaseDataType.TAG),
            og.Type(og.BaseDataType.PRIM),
            og.Type(og.BaseDataType.DOUBLE, 5),
            og.Type(og.BaseDataType.PRIM, 1, 1),
            og.Type(og.BaseDataType.INT, 3, 0, og.AttributeRole.COLOR),
        ]:
            if isinstance(illegal_type, og.Type):
                self.assertFalse(og.AttributeType.is_legal_ogn_type(illegal_type))

            with self.assertRaises(og.OmniGraphError):
                og.ObjectLookup.attribute_type(illegal_type)

    # --------------------------------------------------------------------------------------------------------------
    async def test_deprecation(self):
        """Uses the deprecation helpers to ensure the proper messaging is sent out for each type."""
        (was_silenced, was_showing_stack, current_messages) = (
            ogt.DeprecateMessage.SILENCE_LOG,
            ogt.DeprecateMessage.SHOW_STACK,
            ogt.DeprecateMessage._MESSAGES_LOGGED.copy(),  # noqa: PLW0212
        )
        ogt.DeprecateMessage.SILENCE_LOG = True
        ogt.DeprecateMessage.SHOW_STACK = True
        ogt.DeprecateMessage.clear_messages()
        try:

            def __has_deprecation(pattern: str, expected_count: int) -> bool:
                """Returns True if the pattern appears in the deprecation messages the given number of times"""
                return (
                    sum(
                        match is not None
                        for match in [
                            re.search(pattern, message)
                            for message in ogt.DeprecateMessage._MESSAGES_LOGGED  # noqa: PLW0212
                        ]
                    )
                    == expected_count
                )

            class NewClass:
                CLASS_MEMBER = 1

                def __init__(self, value: int):
                    self._ro_value = value
                    self._rw_value = value * 2

                @classmethod
                def class_method(cls):
                    pass

                @staticmethod
                def static_method():
                    pass

                def regular_method(self):
                    pass

                @property
                def ro_property(self):
                    return self._ro_value

                @property
                def rw_property(self):
                    return self._rw_value

                @rw_property.setter
                def rw_property(self, value):
                    self._rw_value = value

            @ogt.DeprecatedClass("Use NewClass")
            class DeadClass:
                pass

            renamed_class = ogt.RenamedClass(NewClass, "OldName", "Renamed to NewClass")

            @ogt.deprecated_function("Use new_function")
            def old_function():
                return 7

            self.assertFalse(
                ogt.DeprecateMessage._MESSAGES_LOGGED, "Defining deprecations should not log messages"  # noqa: PLW0212
            )

            dead_class = DeadClass()
            self.assertIsNotNone(dead_class)
            self.assertTrue(__has_deprecation(".*Use NewClass", 1), "Instantiated a deprecated class")
            ogt.DeprecateMessage.clear_messages()

            for dead_class in [renamed_class, renamed_class(5)]:
                self.assertIsNotNone(dead_class)
                self.assertTrue(isinstance(dead_class, NewClass))
                self.assertTrue(callable(dead_class.class_method))
                self.assertTrue(callable(dead_class.static_method))
                self.assertTrue(callable(dead_class.regular_method))
                if dead_class.__class__.__name__ == "NewClass":
                    self.assertEqual(dead_class.ro_property, 5)
                    self.assertEqual(dead_class.rw_property, 10)
                    dead_class.rw_property = 11
                    self.assertEqual(dead_class.rw_property, 11)
                self.assertEqual(dead_class.CLASS_MEMBER, 1)
            self.assertTrue(
                __has_deprecation(".*Renamed to NewClass", 2),
                f"Accessing a renamed class {ogt.DeprecateMessage.messages_logged}",
            )
            ogt.DeprecateMessage.clear_messages()

            dead_value = old_function()
            self.assertEqual(dead_value, 7, "Return value from a deprecated function")
            self.assertTrue(__has_deprecation("old_function.*Use new_function", 1), "Called a deprecated function")
            ogt.DeprecateMessage.clear_messages()

            from .deprecated_module import test_function

            test_function()
            self.assertTrue(
                __has_deprecation("deprecated_module.*import omni.graph.core", 1), "Importing a deprecated module"
            )
        finally:
            import sys

            ogt.DeprecateMessage.SILENCE_LOG = was_silenced
            ogt.DeprecateMessage.SHOW_STACK = was_showing_stack
            ogt.DeprecateMessage._MESSAGES_LOGGED = current_messages  # noqa: PLW0212
            with suppress(AttributeError):
                del omni.graph.test.tests.deprecated_module
                sys.modules.pop("omni.graph.test.tests.deprecated_module")

    # --------------------------------------------------------------------------------------------------------------
    async def test_inspector_format(self):
        """Dumps out graph and Fabric contents for a sample file, just to make sure the format stays legal. This
        does not test any of the semantics of the data, only that it will produce legal .json data.
        """
        await self.__populate_data()

        # Check the dump of the Fabric data
        contents = None
        try:
            context = og.get_compute_graph_contexts()[0]
            contents = og.OmniGraphInspector().as_json(context)
            json.loads(contents)
        except json.JSONDecodeError as error:
            self.assertFalse(True, f"Formatting of Fabric data inspection is not legal JSON ({error}) - {contents}")

        # Check the dump of the graph structure
        contents = None
        try:
            for graph in og.get_all_graphs():
                contents = og.OmniGraphInspector().as_json(graph)
                json.loads(contents)
        except json.JSONDecodeError as error:
            self.assertFalse(True, f"Formatting of Fabric data inspection is not legal JSON ({error}) - {contents}")

    # --------------------------------------------------------------------------------------------------------------
    async def test_extension_versions_match(self):
        """Checks to make sure the core versions cited in CHANGELOG.md and extensions.toml match."""
        extension_manager = omni.kit.app.get_app().get_extension_manager()

        for extension_name in [
            "omni.graph",
            "omni.graph.tools",
            "omni.graph.examples.cpp",
            "omni.graph.examples.python",
            "omni.graph.nodes",
            "omni.graph.tutorials",
            "omni.graph.action",
            "omni.graph.scriptnode",
            "omni.inspect",
        ]:

            # Find the canonical extension of the core, which will come from extension.toml
            enabled_version = None
            for version in extension_manager.fetch_extension_versions(extension_name):
                if version["enabled"]:
                    sem_ver = version["version"]
                    enabled_version = f"{sem_ver[0]}.{sem_ver[1]}.{sem_ver[2]}"
                    extension_path = Path(version["path"])
            self.assertIsNotNone(enabled_version, f"Failed to find enabled version of {extension_name}")

            # Find the latest version in the changelog and make sure it matches the enabled version
            re_version_changelog = re.compile(
                r"##\s*\[([0-9]+\.[0-9]+\.[0-9]+)\]\s*-\s*[0-9][0-9][0-9][0-9]-[0-9][0-9]-[0-9][0-9]\s*"
            )
            changelog_path = extension_path / "docs" / "CHANGELOG.md"
            self.assertTrue(changelog_path.is_file(), f"Changelog file {changelog_path} not found")
            changelog_version = None
            with open(changelog_path, "r", encoding="utf-8") as log_fd:
                for line in log_fd:
                    version_match = re_version_changelog.match(line)
                    if version_match:
                        changelog_version = version_match.group(1)
                        break
            self.assertEqual(enabled_version, changelog_version, f"Mismatch in versions of {changelog_path}")
