test_unit.py

#! /opt/local/bin/python

# Copyright (c) 2019 Shotgun Software Inc.
#
# CONFIDENTIAL AND PROPRIETARY
#
# This work is provided "AS IS" and subject to the Shotgun Pipeline Toolkit
# Source Code License included in this distribution package. See LICENSE.
# By accessing, using, copying or modifying this work you indicate your
# agreement to the Shotgun Pipeline Toolkit Source Code License. All rights
# not expressly granted therein are reserved by Shotgun Software Inc.

import os
import unittest
from unittest import mock
from .mock import patch
import shotgun_api3 as api
from shotgun_api3.shotgun import _is_mimetypes_broken
from shotgun_api3.lib.six.moves import range, urllib
from shotgun_api3.lib.httplib2 import Http, ssl_error_classes


class TestShotgunInit(unittest.TestCase):
    """Test case for Shotgun.__init__"""

    def setUp(self):
        self.server_path = "http://server_path"
        self.script_name = "script_name"
        self.api_key = "api_key"

    # Proxy Server Tests
    def test_http_proxy_server(self):
        proxy_server = "someserver.com"
        http_proxy = proxy_server
        sg = api.Shotgun(
            self.server_path,
            self.script_name,
            self.api_key,
            http_proxy=http_proxy,
            connect=False,
        )
        self.assertEqual(sg.config.proxy_server, proxy_server)
        self.assertEqual(sg.config.proxy_port, 8080)
        proxy_server = "123.456.789.012"
        http_proxy = proxy_server
        sg = api.Shotgun(
            self.server_path,
            self.script_name,
            self.api_key,
            http_proxy=http_proxy,
            connect=False,
        )
        self.assertEqual(sg.config.proxy_server, proxy_server)
        self.assertEqual(sg.config.proxy_port, 8080)

    def test_http_proxy_server_and_port(self):
        proxy_server = "someserver.com"
        proxy_port = 1234
        http_proxy = "%s:%d" % (proxy_server, proxy_port)
        sg = api.Shotgun(
            self.server_path,
            self.script_name,
            self.api_key,
            http_proxy=http_proxy,
            connect=False,
        )
        self.assertEqual(sg.config.proxy_server, proxy_server)
        self.assertEqual(sg.config.proxy_port, proxy_port)
        proxy_server = "123.456.789.012"
        proxy_port = 1234
        http_proxy = "%s:%d" % (proxy_server, proxy_port)
        sg = api.Shotgun(
            self.server_path,
            self.script_name,
            self.api_key,
            http_proxy=http_proxy,
            connect=False,
        )
        self.assertEqual(sg.config.proxy_server, proxy_server)
        self.assertEqual(sg.config.proxy_port, proxy_port)

    def test_http_proxy_server_and_port_with_authentication(self):
        proxy_server = "someserver.com"
        proxy_port = 1234
        proxy_user = "user"
        proxy_pass = "password"
        http_proxy = "%s:%s@%s:%d" % (proxy_user, proxy_pass, proxy_server, proxy_port)
        sg = api.Shotgun(
            self.server_path,
            self.script_name,
            self.api_key,
            http_proxy=http_proxy,
            connect=False,
        )
        self.assertEqual(sg.config.proxy_server, proxy_server)
        self.assertEqual(sg.config.proxy_port, proxy_port)
        self.assertEqual(sg.config.proxy_user, proxy_user)
        self.assertEqual(sg.config.proxy_pass, proxy_pass)
        proxy_server = "123.456.789.012"
        proxy_port = 1234
        proxy_user = "user"
        proxy_pass = "password"
        http_proxy = "%s:%s@%s:%d" % (proxy_user, proxy_pass, proxy_server, proxy_port)
        sg = api.Shotgun(
            self.server_path,
            self.script_name,
            self.api_key,
            http_proxy=http_proxy,
            connect=False,
        )
        self.assertEqual(sg.config.proxy_server, proxy_server)
        self.assertEqual(sg.config.proxy_port, proxy_port)
        self.assertEqual(sg.config.proxy_user, proxy_user)
        self.assertEqual(sg.config.proxy_pass, proxy_pass)

    def test_http_proxy_with_at_in_password(self):
        proxy_server = "someserver.com"
        proxy_port = 1234
        proxy_user = "user"
        proxy_pass = "p@ssword"
        http_proxy = "%s:%s@%s:%d" % (proxy_user, proxy_pass, proxy_server, proxy_port)
        sg = api.Shotgun(
            self.server_path,
            self.script_name,
            self.api_key,
            http_proxy=http_proxy,
            connect=False,
        )
        self.assertEqual(sg.config.proxy_server, proxy_server)
        self.assertEqual(sg.config.proxy_port, proxy_port)
        self.assertEqual(sg.config.proxy_user, proxy_user)
        self.assertEqual(sg.config.proxy_pass, proxy_pass)

    def test_malformatted_proxy_info(self):
        conn_info = {
            "base_url": self.server_path,
            "script_name": self.script_name,
            "api_key": self.api_key,
            "connect": False,
        }
        conn_info["http_proxy"] = "http://someserver.com"
        self.assertRaises(ValueError, api.Shotgun, **conn_info)
        conn_info["http_proxy"] = "user@someserver.com"
        self.assertRaises(ValueError, api.Shotgun, **conn_info)
        conn_info["http_proxy"] = "someserver.com:1234:5678"
        self.assertRaises(ValueError, api.Shotgun, **conn_info)


class TestShotgunSummarize(unittest.TestCase):
    """Test case for _create_summary_request function and parameter
    validation as it exists in Shotgun.summarize.

    Does not require database connection or test data."""

    def setUp(self):
        self.sg = api.Shotgun(
            "http://server_path", "script_name", "api_key", connect=False
        )

    def test_filter_operator_none(self):
        expected_logical_operator = "and"
        filter_operator = None
        self._assert_filter_operator(expected_logical_operator, filter_operator)

    def _assert_filter_operator(self, expected_logical_operator, filter_operator):
        result = self.get_call_rpc_params(None, {"filter_operator": filter_operator})
        actual_logical_operator = result["filters"]["logical_operator"]
        self.assertEqual(expected_logical_operator, actual_logical_operator)

    def test_filter_operator_all(self):
        expected_logical_operator = "and"
        filter_operator = "all"
        self._assert_filter_operator(expected_logical_operator, filter_operator)

    def test_filter_operator_or(self):
        expected_logical_operator = "or"
        filter_operator = "or"
        self._assert_filter_operator(expected_logical_operator, filter_operator)

    def test_filters(self):
        path = "path"
        relation = "relation"
        value = "value"
        expected_condition = {"path": path, "relation": relation, "values": [value]}
        args = ["", [[path, relation, value]], None]
        result = self.get_call_rpc_params(args, {})
        actual_condition = result["filters"]["conditions"][0]
        self.assertEqual(expected_condition, actual_condition)

    @patch("shotgun_api3.Shotgun._call_rpc")
    def get_call_rpc_params(self, args, kws, call_rpc):
        """Return params sent to _call_rpc from summarize."""
        if not args:
            args = [None, [], None]
        self.sg.summarize(*args, **kws)
        return call_rpc.call_args[0][1]

    def test_grouping(self):
        result = self.get_call_rpc_params(None, {})
        self.assertFalse("grouping" in result)
        grouping = ["something"]
        kws = {"grouping": grouping}
        result = self.get_call_rpc_params(None, kws)
        self.assertEqual(grouping, result["grouping"])

    def test_grouping_type(self):
        """test_grouping_type tests that grouping parameter is a list or None"""
        self.assertRaises(
            ValueError, self.sg.summarize, "", [], [], grouping="Not a list"
        )


class TestShotgunBatch(unittest.TestCase):
    def setUp(self):
        self.sg = api.Shotgun(
            "http://server_path", "script_name", "api_key", connect=False
        )

    def test_missing_required_key(self):
        req = {}
        # requires keys request_type and entity_type
        self.assertRaises(api.ShotgunError, self.sg.batch, [req])
        req["entity_type"] = "Entity"
        self.assertRaises(api.ShotgunError, self.sg.batch, [req])
        req["request_type"] = "not_real_type"
        self.assertRaises(api.ShotgunError, self.sg.batch, [req])
        # create requires data key
        req["request_type"] = "create"
        self.assertRaises(api.ShotgunError, self.sg.batch, [req])
        # update requires entity_id and data
        req["request_type"] = "update"
        req["data"] = {}
        self.assertRaises(api.ShotgunError, self.sg.batch, [req])
        del req["data"]
        req["entity_id"] = 2334
        self.assertRaises(api.ShotgunError, self.sg.batch, [req])
        # delete requires entity_id
        req["request_type"] = "delete"
        del req["entity_id"]
        self.assertRaises(api.ShotgunError, self.sg.batch, [req])


class TestServerCapabilities(unittest.TestCase):
    def test_no_server_version(self):
        self.assertRaises(api.ShotgunError, api.shotgun.ServerCapabilities, "host", {})

    def test_bad_version(self):
        """test_bad_meta tests passing bad meta data type"""
        self.assertRaises(
            api.ShotgunError,
            api.shotgun.ServerCapabilities,
            "host",
            {"version": (0, 0, 0)},
        )

    def test_dev_version(self):
        serverCapabilities = api.shotgun.ServerCapabilities(
            "host", {"version": (3, 4, 0, "Dev")}
        )
        self.assertEqual(serverCapabilities.version, (3, 4, 0))
        self.assertTrue(serverCapabilities.is_dev)

        serverCapabilities = api.shotgun.ServerCapabilities(
            "host", {"version": (2, 4, 0)}
        )
        self.assertEqual(serverCapabilities.version, (2, 4, 0))
        self.assertFalse(serverCapabilities.is_dev)


class TestClientCapabilities(unittest.TestCase):

    def test_darwin(self):
        self.assert_platform("Darwin", "mac")

    def test_windows(self):
        self.assert_platform("win32", "windows")

    def test_linux(self):
        self.assert_platform("Linux", "linux")

    def assert_platform(self, sys_ret_val, expected):
        platform = api.shotgun.sys.platform
        try:
            api.shotgun.sys.platform = sys_ret_val
            expected_local_path_field = "local_path_%s" % expected

            client_caps = api.shotgun.ClientCapabilities()
            self.assertEqual(client_caps.platform, expected)
            self.assertEqual(client_caps.local_path_field, expected_local_path_field)
        finally:
            api.shotgun.sys.platform = platform

    def test_no_platform(self):
        platform = api.shotgun.sys.platform
        try:
            api.shotgun.sys.platform = "unsupported"
            client_caps = api.shotgun.ClientCapabilities()
            self.assertEqual(client_caps.platform, None)
            self.assertEqual(client_caps.local_path_field, None)
        finally:
            api.shotgun.sys.platform = platform

    @patch("shotgun_api3.shotgun.sys")
    def test_py_version(self, mock_sys):
        major = 2
        minor = 7
        micro = 3
        mock_sys.version_info = (major, minor, micro, "final", 0)
        expected_py_version = "%s.%s" % (major, minor)
        client_caps = api.shotgun.ClientCapabilities()
        self.assertEqual(client_caps.py_version, expected_py_version)


class TestFilters(unittest.TestCase):
    maxDiff = None

    def test_empty(self):
        expected = {"logical_operator": "and", "conditions": []}

        result = api.shotgun._translate_filters([], None)
        self.assertEqual(result, expected)

    def test_simple(self):
        filters = [["code", "is", "test"], ["sg_status_list", "is", "ip"]]

        expected = {
            "logical_operator": "or",
            "conditions": [
                {"path": "code", "relation": "is", "values": ["test"]},
                {"path": "sg_status_list", "relation": "is", "values": ["ip"]},
            ],
        }

        result = api.shotgun._translate_filters(filters, "any")
        self.assertEqual(result, expected)

    # Test both styles of passing arrays
    def test_arrays(self):
        expected = {
            "logical_operator": "and",
            "conditions": [
                {
                    "path": "code",
                    "relation": "in",
                    "values": ["test1", "test2", "test3"],
                }
            ],
        }

        filters = [["code", "in", "test1", "test2", "test3"]]

        result = api.shotgun._translate_filters(filters, "all")
        self.assertEqual(result, expected)

        filters = [["code", "in", ["test1", "test2", "test3"]]]

        result = api.shotgun._translate_filters(filters, "all")
        self.assertEqual(result, expected)

    def test_nested(self):
        filters = [
            ["code", "in", "test"],
            {
                "filter_operator": "any",
                "filters": [
                    ["sg_status_list", "is", "ip"],
                    ["sg_status_list", "is", "fin"],
                    {
                        "filter_operator": "all",
                        "filters": [
                            ["sg_status_list", "is", "hld"],
                            ["assets", "is", {"type": "Asset", "id": 9}],
                        ],
                    },
                ],
            },
        ]

        expected = {
            "logical_operator": "and",
            "conditions": [
                {"path": "code", "relation": "in", "values": ["test"]},
                {
                    "logical_operator": "or",
                    "conditions": [
                        {"path": "sg_status_list", "relation": "is", "values": ["ip"]},
                        {"path": "sg_status_list", "relation": "is", "values": ["fin"]},
                        {
                            "logical_operator": "and",
                            "conditions": [
                                {
                                    "path": "sg_status_list",
                                    "relation": "is",
                                    "values": ["hld"],
                                },
                                {
                                    "path": "assets",
                                    "relation": "is",
                                    "values": [{"type": "Asset", "id": 9}],
                                },
                            ],
                        },
                    ],
                },
            ],
        }

        result = api.shotgun._translate_filters(filters, "all")
        self.assertEqual(result, expected)

    def test_invalid(self):
        self.assertRaises(api.ShotgunError, api.shotgun._translate_filters, [], "bogus")
        self.assertRaises(
            api.ShotgunError, api.shotgun._translate_filters, ["bogus"], "all"
        )

        filters = [{"filter_operator": "bogus", "filters": []}]

        self.assertRaises(
            api.ShotgunError, api.shotgun._translate_filters, filters, "all"
        )

        filters = [{"filters": []}]

        self.assertRaises(
            api.ShotgunError, api.shotgun._translate_filters, filters, "all"
        )

        filters = [{"filter_operator": "all", "filters": {"bogus": "bogus"}}]

        self.assertRaises(
            api.ShotgunError, api.shotgun._translate_filters, filters, "all"
        )

    @mock.patch.dict(os.environ, {"SHOTGUN_API_DISABLE_ENTITY_OPTIMIZATION": "1"})
    def test_related_object(self):
        filters = [
            [
                "project",
                "is",
                {
                    "foo": "foo",
                    "bar": "bar",
                    "id": 999,
                    "baz": "baz",
                    "type": "Anything",
                },
            ],
        ]
        expected = {
            "logical_operator": "and",
            "conditions": [
                {
                    "path": "project",
                    "relation": "is",
                    "values": [
                        {
                            "foo": "foo",
                            "bar": "bar",
                            "baz": "baz",
                            "id": 999,
                            "type": "Anything",
                        }
                    ],
                }
            ],
        }
        api.Shotgun("http://server_path", "script_name", "api_key", connect=False)
        result = api.shotgun._translate_filters(filters, "all")
        self.assertEqual(result, expected)

    @mock.patch("shotgun_api3.shotgun.SHOTGUN_API_DISABLE_ENTITY_OPTIMIZATION", False)
    def test_related_object_entity_optimization_is(self):
        filters = [
            [
                "project",
                "is",
                {
                    "foo": "foo",
                    "bar": "bar",
                    "id": 999,
                    "baz": "baz",
                    "type": "Anything",
                },
            ],
        ]
        expected = {
            "logical_operator": "and",
            "conditions": [
                {
                    "path": "project",
                    "relation": "is",
                    "values": [
                        {
                            "id": 999,
                            "type": "Anything",
                        }
                    ],
                }
            ],
        }
        api.Shotgun("http://server_path", "script_name", "api_key", connect=False)
        result = api.shotgun._translate_filters(filters, "all")
        self.assertEqual(result, expected)

        # Now test a non-related object. The expected result should not be optimized.
        filters = [
            [
                "something",
                "is",
                {"foo": "foo", "bar": "bar"},
            ],
        ]
        expected = {
            "logical_operator": "and",
            "conditions": [
                {
                    "path": "something",
                    "relation": "is",
                    "values": [{"bar": "bar", "foo": "foo"}],
                }
            ],
        }
        api.Shotgun("http://server_path", "script_name", "api_key", connect=False)
        result = api.shotgun._translate_filters(filters, "all")
        self.assertEqual(result, expected)

    @mock.patch("shotgun_api3.shotgun.SHOTGUN_API_DISABLE_ENTITY_OPTIMIZATION", False)
    def test_related_object_entity_optimization_in(self):
        filters = [
            [
                "project",
                "in",
                [
                    {
                        "foo1": "foo1",
                        "bar1": "bar1",
                        "id": 999,
                        "baz1": "baz1",
                        "type": "Anything",
                    },
                    {
                        "foo2": "foo2",
                        "bar2": "bar2",
                        "id": 998,
                        "baz2": "baz2",
                        "type": "Anything",
                    },
                    {"foo3": "foo3", "bar3": "bar3"},
                ],
            ],
        ]
        expected = {
            "logical_operator": "and",
            "conditions": [
                {
                    "path": "project",
                    "relation": "in",
                    "values": [
                        {
                            "id": 999,
                            "type": "Anything",
                        },
                        {
                            "id": 998,
                            "type": "Anything",
                        },
                        {
                            "foo3": "foo3",
                            "bar3": "bar3",
                        },
                    ],
                }
            ],
        }
        api.Shotgun("http://server_path", "script_name", "api_key", connect=False)
        result = api.shotgun._translate_filters(filters, "all")
        self.assertEqual(result, expected)

    def test_related_object_update_entity(self):
        entity_type = "Anything"
        entity_id = 999
        multi_entity_update_modes = {"link": "set", "name": "set"}
        data = {
            "name": "test",
            "link": {
                "name": "test",
                "url": "http://test.com",
            },
        }
        expected = {
            "id": 999,
            "type": "Anything",
            "fields": [
                {
                    "field_name": "name",
                    "value": "test",
                    "multi_entity_update_mode": "set",
                },
                {
                    "field_name": "link",
                    "value": {
                        "name": "test",
                        "url": "http://test.com",
                    },
                    "multi_entity_update_mode": "set",
                },
            ],
        }
        sg = api.Shotgun("http://server_path", "script_name", "api_key", connect=False)
        result = sg._translate_update_params(
            entity_type, entity_id, data, multi_entity_update_modes
        )
        self.assertEqual(result, expected)

    @mock.patch("shotgun_api3.shotgun.SHOTGUN_API_DISABLE_ENTITY_OPTIMIZATION", False)
    def test_related_object_update_optimization_entity(self):
        entity_type = "Anything"
        entity_id = 999
        multi_entity_update_modes = {"project": "set", "link": "set", "name": "set"}
        data = {
            "name": "test",
            "link": {
                "name": "test",
                "url": "http://test.com",
            },
            "project": {
                "foo1": "foo1",
                "bar1": "bar1",
                "id": 888,
                "baz1": "baz1",
                "type": "Project",
            },
        }
        expected = {
            "id": 999,
            "type": "Anything",
            "fields": [
                {
                    "field_name": "name",
                    "value": "test",
                    "multi_entity_update_mode": "set",
                },
                {
                    "field_name": "link",
                    "value": {
                        "name": "test",
                        "url": "http://test.com",
                    },
                    "multi_entity_update_mode": "set",
                },
                {
                    "field_name": "project",
                    "multi_entity_update_mode": "set",
                    "value": {
                        # Entity is optimized with type/id fields.
                        "id": 888,
                        "type": "Project",
                    },
                },
            ],
        }
        sg = api.Shotgun("http://server_path", "script_name", "api_key", connect=False)
        result = sg._translate_update_params(
            entity_type, entity_id, data, multi_entity_update_modes
        )
        self.assertEqual(result, expected)

    @mock.patch("shotgun_api3.shotgun.SHOTGUN_API_DISABLE_ENTITY_OPTIMIZATION", False)
    def test_related_object_update_optimization_entity_multi(self):
        entity_type = "Asset"
        entity_id = 6626
        data = {
            "sg_status_list": "ip",
            "project": {"id": 70, "type": "Project", "name": "disposable name 70"},
            "sg_vvv": [
                {"id": 6441, "type": "Asset", "name": "disposable name 6441"},
                {"id": 6440, "type": "Asset"},
            ],
            "sg_class": {
                "id": 1,
                "type": "CustomEntity53",
                "name": "disposable name 1",
            },
        }
        expected = {
            "type": "Asset",
            "id": 6626,
            "fields": [
                {"field_name": "sg_status_list", "value": "ip"},
                {"field_name": "project", "value": {"type": "Project", "id": 70}},
                {
                    "field_name": "sg_vvv",
                    "value": [
                        {"id": 6441, "type": "Asset"},
                        {"id": 6440, "type": "Asset"},
                    ],
                },
                {
                    "field_name": "sg_class",
                    "value": {"type": "CustomEntity53", "id": 1},
                },
            ],
        }
        sg = api.Shotgun("http://server_path", "script_name", "api_key", connect=False)
        result = sg._translate_update_params(entity_type, entity_id, data, None)
        self.assertEqual(result, expected)


class TestCerts(unittest.TestCase):
    # A dummy bad url provided by Amazon
    bad_url = "https://untrusted-root.badssl.com/"
    # A list of Amazon cert URLS, taken from here:
    # https://aws.amazon.com/blogs/security/how-to-prepare-for-aws-move-to-its-own-certificate-authority/
    test_urls = [
        "https://good.sca1a.amazontrust.com",
        "https://good.sca2a.amazontrust.com",
        "https://good.sca3a.amazontrust.com",
        "https://good.sca4a.amazontrust.com",
        "https://good.sca0a.amazontrust.com",
    ]

    def setUp(self):
        self.sg = api.Shotgun(
            "http://server_path", "script_name", "api_key", connect=False
        )

        # Get the location of the certs file
        self.certs = self.sg._get_certs_file(None)

    def _check_url_with_sg_api_httplib2(self, url, certs):
        """
        Given a url and the certs file, it will do a simple
        request and return the result.
        """
        http = Http(ca_certs=certs)
        return http.request(url)

    def _check_url_with_urllib(self, url):
        """
        Given a url it will perform a simple request and return a result.
        """
        # create a request using the opener generated by the PTR API.
        # The `_build_opener` method internally should use the correct certs.
        opener = self.sg._build_opener(urllib.request.HTTPHandler)
        request = urllib.request.Request(url)
        return opener.open(request)

    def test_found_correct_cert(self):
        """
        Checks that the cert file the API is finding,
        (when a cert path isn't passed and the SHOTGUN_API_CACERTS
        isn't set), is the one bundled with this API
        """
        # Get the path to the cert file we expect the Shotgun API to find
        cert_path = os.path.normpath(
            # Get the path relative to where we picked up the API and not relative
            # to file on disk. On CI we pip install the API to run the tests
            # so we have to pick the location from the installed copy.
            # Call dirname to remove from __init__.py
            os.path.join(os.path.dirname(api.__file__), "lib", "certifi", "cacert.pem")
        )
        # Now ensure that the path the PTR API has found is correct.
        self.assertEqual(cert_path, self.certs)
        self.assertTrue(os.path.isfile(self.certs))

    def test_httplib(self):
        """
        Checks that we can access the amazon urls using our bundled
        certificate with httplib.
        """
        # First check that we get an error when trying to connect to a known dummy bad URL
        self.assertRaises(
            ssl_error_classes,
            self._check_url_with_sg_api_httplib2,
            self.bad_url,
            self.certs,
        )

        # Now check that the good urls connect properly using the certs
        for url in self.test_urls:
            response, message = self._check_url_with_sg_api_httplib2(url, self.certs)
            self.assertEqual(response["status"], "200")

    def test_urlib(self):
        """
        Checks that we can access the amazon urls using our bundled
        certificate with urllib.
        """
        # First check that we get an error when trying to connect to a known dummy bad URL
        self.assertRaises(
            urllib.error.URLError, self._check_url_with_urllib, self.bad_url
        )

        # Now check that the good urls connect properly using the certs
        for url in self.test_urls:
            response = self._check_url_with_urllib(url)
            assert response is not None


class TestFilterDeduplication(unittest.TestCase):
    """Test filter deduplication utility functions"""

    def setUp(self):
        """Set up test fixtures"""
        from shotgun_api3.shotgun import (
            normalize_filter,
            complex_filter,
            remove_duplicate_filters
        )
        self.normalize_filter = normalize_filter
        self.complex_filter = complex_filter
        self.remove_duplicate_filters = remove_duplicate_filters

    def test_normalize_filter_simple_list(self):
        """Test normalizing simple list-based filters"""
        filter_obj = ["project", "is", {"type": "Project", "id": 123}]
        result = self.normalize_filter(filter_obj)
        expected = ("project", "is", (("id", 123), ("type", "Project")))
        self.assertEqual(result, expected)

    def test_normalize_filter_simple_dict(self):
        """Test normalizing simple dict-based filters"""
        filter_obj = {"path": "project", "relation": "is", "values": [{"type": "Project", "id": 123}]}
        result = self.normalize_filter(filter_obj)

        # Should be a tuple with sorted dict keys
        result_str = str(result)
        self.assertIsInstance(result, tuple)
        self.assertIn("path", result_str)
        self.assertIn("relation", result_str)
        self.assertIn("values", result_str)

    def test_normalize_filter_complex_nested(self):
        """Test normalizing complex nested filters"""
        complex_filter_obj = {
            "filter_operator": "and",
            "filters": [
                ["project", "is", {"type": "Project", "id": 123}],
                ["sg_status_list", "is", "rev"]
            ]
        }
        result = self.normalize_filter(complex_filter_obj)

        # Should contain sorted keys and nested normalized filters
        result_str = str(result)
        self.assertIsInstance(result, tuple)
        self.assertIn("filter_operator", result_str)
        self.assertIn("filters", result_str)

    def test_complex_filter_detection_true(self):
        """Test complex_filter() returns True for complex filters"""
        complex_filters = [
            {"filter_operator": "and", "filters": []},
            {"filters": [["project", "is", {"type": "Project", "id": 123}]]},
            {"filter_operator": "or", "filters": [["id", "is", 1]]}
        ]
        for f in complex_filters:
            with self.subTest(filter=f):
                self.assertTrue(self.complex_filter(f))

    def test_complex_filter_detection_false(self):
        """Test complex_filter() returns False for simple filters"""
        simple_filters = [
            ["project", "is", {"type": "Project", "id": 123}],
            {"path": "id", "values": [1]},
            {"path": "sg_status_list", "relation": "is", "values": ["rev"]},
            "simple_string",
            123,
            None
        ]
        for f in simple_filters:
            with self.subTest(filter=f):
                self.assertFalse(self.complex_filter(f))

    def test_remove_duplicate_filters_simple(self):
        """Test removing duplicates from simple filter list"""
        project_filter = ["project", "is", {"type": "Project", "id": 123}]
        status_filter = ["sg_status_list", "is", "rev"]
        filters = [
            project_filter,
            status_filter,
            project_filter,  # duplicate
            ["entity", "type_is", "Shot"],
            project_filter   # duplicate
        ]
        result = self.remove_duplicate_filters(filters)

        # Should have 3 unique sorted filters
        self.assertEqual(len(result), 3)
        self.assertEqual(result[0], project_filter)
        self.assertEqual(result[1], status_filter)
        self.assertEqual(result[2], ["entity", "type_is", "Shot"])

    def test_remove_duplicate_filters_complex(self):
        """Test removing duplicates with complex nested filters"""
        simple_filter = ["project", "is", {"type": "Project", "id": 123}]
        complex_filter_1 = {
            "filter_operator": "and",
            "filters": [
                ["sg_status_list", "is", "rev"],
                ["entity", "type_is", "Shot"]
            ]
        }
        complex_filter_2 = {
            "filter_operator": "or",
            "filters": [["id", "is", 1]]
        }
        filters = [
            simple_filter,
            complex_filter_1,
            simple_filter,      # duplicate
            complex_filter_2,
            complex_filter_1    # duplicate
        ]
        result = self.remove_duplicate_filters(filters)

        # Should have 3 unique sorted filters
        self.assertEqual(len(result), 3)
        self.assertEqual(result[0], simple_filter)
        self.assertEqual(result[1], complex_filter_1)
        self.assertEqual(result[2], complex_filter_2)

    def test_remove_duplicate_filters_preserves_order(self):
        """Test that order is preserved for remaining filters"""
        filter_a = ["field_a", "is", "value_a"]
        filter_b = ["field_b", "is", "value_b"]
        filter_c = ["field_c", "is", "value_c"]
        filters = [filter_a, filter_b, filter_a, filter_c, filter_b]
        result = self.remove_duplicate_filters(filters)
        expected = [filter_a, filter_b, filter_c] # Should preserve order

        self.assertEqual(result, expected)

    def test_remove_duplicate_filters_edge_cases(self):
        """Test edge cases: empty lists, single item, no duplicates"""
        # Empty list
        self.assertEqual(self.remove_duplicate_filters([]), [])

        # Single item
        single_filter = [["project", "is", {"type": "Project", "id": 123}]]
        self.assertEqual(self.remove_duplicate_filters(single_filter), single_filter)

        # No duplicates
        unique_filters = [
            ["project", "is", {"type": "Project", "id": 123}],
            ["sg_status_list", "is", "rev"],
            ["entity", "type_is", "Shot"]
        ]
        self.assertEqual(self.remove_duplicate_filters(unique_filters), unique_filters)

    def test_deduplicate_nested_conditions(self):
        """Test deduplicating nested conditions in complex filters - testing via remove_duplicate_filters"""
        nested_filter = {
            "filter_operator": "and",
            "filters": [
                ["project", "is", {"type": "Project", "id": 123}],
                ["sg_status_list", "is", "rev"],
                ["project", "is", {"type": "Project", "id": 123}]  # duplicate
            ]
        }
        result = self.remove_duplicate_filters([nested_filter])

        # Should have one complex filter with deduplicated nested conditions
        result_filter = result[0]
        self.assertEqual(len(result), 1)
        self.assertEqual(result_filter["filter_operator"], "and")
        self.assertEqual(len(result_filter["filters"]), 2)
        self.assertEqual(result_filter["filters"][0], ["project", "is", {"type": "Project", "id": 123}])
        self.assertEqual(result_filter["filters"][1], ["sg_status_list", "is", "rev"])

    def test_deduplicate_function_mixed_filters(self):
        """Test main function with mixed simple and complex filters"""
        project_filter = ["project", "is", {"type": "Project", "id": 123}]
        complex_filter = {
            "filter_operator": "and",
            "filters": [
                ["sg_status_list", "is", "rev"],
                project_filter,  # Nested duplication
                ["entity", "type_is", "Shot"]
            ]
        }
        filters = [
            project_filter,
            complex_filter,
            project_filter,  # Top-level duplication
        ]
        result = self.remove_duplicate_filters(filters)

        # Should have 2 items (project_filter + deduplicated complex_filter)
        self.assertEqual(len(result), 2)
        self.assertEqual(result[0], project_filter)

        # Complex filter should be processed and nested duplicates removed
        self.assertIsInstance(result[1], dict)
        self.assertEqual(result[1]["filter_operator"], "and")


class TestMimetypesFix(unittest.TestCase):
    """
    Makes sure that the mimetypes fix will be imported.
    """

    @patch("shotgun_api3.shotgun.sys")
    def _test_mimetypes_import(
        self, platform, major, minor, patch_number, result, mock
    ):
        """
        Mocks sys.platform and sys.version_info to test the mimetypes import code.
        """

        mock.version_info = [major, minor, patch_number]
        mock.platform = platform
        self.assertEqual(_is_mimetypes_broken(), result)


if __name__ == "__main__":
    unittest.main()