import collections
import fractions
import sys

import pytest
from packaging.specifiers import SpecifierSet

import asdf
from asdf import AsdfFile, config_context
from asdf.exceptions import AsdfManifestURIMismatchWarning, AsdfSerializationError, ValidationError
from asdf.extension import (
    Compressor,
    Converter,
    ConverterProxy,
    Extension,
    ExtensionManager,
    ExtensionProxy,
    ManifestExtension,
    TagDefinition,
    Validator,
    get_cached_extension_manager,
)
from asdf.extension._manager import _resolve_type
from asdf.testing.helpers import roundtrip_object


class MinimumExtension:
    extension_uri = "asdf://somewhere.org/extensions/minimum-1.0"


class MinimumExtensionSubclassed(Extension):
    extension_uri = "asdf://somewhere.org/extensions/minimum-1.0"


class FullExtension:
    extension_uri = "asdf://somewhere.org/extensions/full-1.0"

    def __init__(
        self,
        converters=None,
        compressors=None,
        validators=None,
        asdf_standard_requirement=None,
        tags=None,
        legacy_class_names=None,
    ):
        self._converters = [] if converters is None else converters
        self._compressors = [] if compressors is None else compressors
        self._validators = [] if validators is None else validators
        self._asdf_standard_requirement = asdf_standard_requirement
        self._tags = [] if tags is None else tags
        self._legacy_class_names = [] if legacy_class_names is None else legacy_class_names

    @property
    def converters(self):
        return self._converters

    @property
    def compressors(self):
        return self._compressors

    @property
    def validators(self):
        return self._validators

    @property
    def asdf_standard_requirement(self):
        return self._asdf_standard_requirement

    @property
    def tags(self):
        return self._tags

    @property
    def legacy_class_names(self):
        return self._legacy_class_names


class MinimumConverter:
    def __init__(self, tags=None, types=None):
        if tags is None:
            self._tags = []
        else:
            self._tags = tags

        if types is None:
            self._types = []
        else:
            self._types = types

    @property
    def tags(self):
        return self._tags

    @property
    def types(self):
        return self._types

    def to_yaml_tree(self, obj, tag, ctx):
        return "to_yaml_tree result"

    def from_yaml_tree(self, obj, tag, ctx):
        return "from_yaml_tree result"


class FullConverter(MinimumConverter):
    def select_tag(self, obj, tags, ctx):
        return "select_tag result"


class MinimalCompressor(Compressor):
    @staticmethod
    def compress(data):
        return b""

    @property
    def label(self):
        return b"mini"


class MinimalValidator(Validator):
    schema_property = "fail"
    tags = ["**"]

    def validate(self, fail, node, schema):
        if fail:
            yield ValidationError("Node was doomed to fail")


# Some dummy types for testing converters:
class FooType:
    pass


class SubFooType(FooType):
    pass


class BarType:
    pass


class BazType:
    pass


def test_extension_proxy_maybe_wrap():
    extension = MinimumExtension()
    proxy = ExtensionProxy.maybe_wrap(extension)
    assert proxy.delegate is extension
    assert ExtensionProxy.maybe_wrap(proxy) is proxy

    with pytest.raises(TypeError, match=r"Extension must implement the Extension interface"):
        ExtensionProxy.maybe_wrap(object())


def test_extension_proxy():
    # Test with minimum properties:
    extension = MinimumExtension()
    proxy = ExtensionProxy(extension)

    assert isinstance(proxy, Extension)

    assert proxy.extension_uri == "asdf://somewhere.org/extensions/minimum-1.0"
    assert proxy.legacy_class_names == set()
    assert proxy.asdf_standard_requirement == SpecifierSet()
    assert proxy.converters == []
    assert proxy.compressors == []
    assert proxy.validators == []
    assert proxy.tags == []
    assert proxy.types == []
    assert proxy.tag_mapping == []
    assert proxy.url_mapping == []
    assert proxy.delegate is extension
    assert proxy.legacy is False
    assert proxy.package_name is None
    assert proxy.package_version is None
    assert proxy.class_name == "asdf._tests.test_extension.MinimumExtension"

    # The subclassed version should have the same defaults:
    extension = MinimumExtensionSubclassed()
    subclassed_proxy = ExtensionProxy(extension)
    assert subclassed_proxy.extension_uri == proxy.extension_uri
    assert subclassed_proxy.legacy_class_names == proxy.legacy_class_names
    assert subclassed_proxy.asdf_standard_requirement == proxy.asdf_standard_requirement
    assert subclassed_proxy.converters == proxy.converters
    assert subclassed_proxy.compressors == proxy.compressors
    assert subclassed_proxy.validators == proxy.validators
    assert subclassed_proxy.tags == proxy.tags
    assert subclassed_proxy.types == proxy.types
    assert subclassed_proxy.tag_mapping == proxy.tag_mapping
    assert subclassed_proxy.url_mapping == proxy.url_mapping
    assert subclassed_proxy.delegate is extension
    assert subclassed_proxy.legacy == proxy.legacy
    assert subclassed_proxy.package_name == proxy.package_name
    assert subclassed_proxy.package_version == proxy.package_name
    assert subclassed_proxy.class_name == "asdf._tests.test_extension.MinimumExtensionSubclassed"

    # Test with all properties present:
    converters = [MinimumConverter(tags=["asdf://somewhere.org/extensions/full/tags/foo-*"], types=[])]
    compressors = [MinimalCompressor()]
    validators = [MinimalValidator()]
    extension = FullExtension(
        converters=converters,
        compressors=compressors,
        validators=validators,
        asdf_standard_requirement=">=1.4.0",
        tags=["asdf://somewhere.org/extensions/full/tags/foo-1.0"],
        legacy_class_names=["foo.extensions.SomeOldExtensionClass"],
    )
    proxy = ExtensionProxy(extension, package_name="foo", package_version="1.2.3")

    assert proxy.extension_uri == "asdf://somewhere.org/extensions/full-1.0"
    assert proxy.legacy_class_names == {"foo.extensions.SomeOldExtensionClass"}
    assert proxy.asdf_standard_requirement == SpecifierSet(">=1.4.0")
    assert proxy.converters == [ConverterProxy(c, proxy) for c in converters]
    assert proxy.compressors == compressors
    assert proxy.validators == validators
    assert len(proxy.tags) == 1
    assert proxy.tags[0].tag_uri == "asdf://somewhere.org/extensions/full/tags/foo-1.0"
    assert proxy.types == []
    assert proxy.tag_mapping == []
    assert proxy.url_mapping == []
    assert proxy.delegate is extension
    assert proxy.legacy is False
    assert proxy.package_name == "foo"
    assert proxy.package_version == "1.2.3"
    assert proxy.class_name == "asdf._tests.test_extension.FullExtension"

    # Should fail when the input is not one of the two extension interfaces:
    with pytest.raises(TypeError, match=r"Extension must implement the Extension interface"):
        ExtensionProxy(object)

    # Should fail with a bad converter:
    with pytest.raises(TypeError, match=r"Converter must implement the .* interface"):
        ExtensionProxy(FullExtension(converters=[object()]))

    # Should fail with a bad compressor:
    with pytest.raises(TypeError, match=r"Extension property 'compressors' must contain instances of .*"):
        ExtensionProxy(FullExtension(compressors=[object()]))

    # Should fail with a bad validator
    with pytest.raises(TypeError, match=r"Extension property 'validators' must contain instances of .*"):
        ExtensionProxy(FullExtension(validators=[object()]))

    # Unparsable ASDF Standard requirement:
    with pytest.raises(ValueError, match=r"Invalid specifier:.*"):
        ExtensionProxy(FullExtension(asdf_standard_requirement="asdf-standard >= 1.4.0"))

    # Unrecognized ASDF Standard requirement type:
    with pytest.raises(TypeError, match=r"Extension property 'asdf_standard_requirement' must be str or None"):
        ExtensionProxy(FullExtension(asdf_standard_requirement=object()))

    # Bad tag:
    with pytest.raises(TypeError, match=r"Extension property 'tags' must contain str or .* values"):
        ExtensionProxy(FullExtension(tags=[object()]))

    # Bad legacy class names:
    with pytest.raises(TypeError, match=r"Extension property 'legacy_class_names' must contain str values"):
        ExtensionProxy(FullExtension(legacy_class_names=[object]))


def test_extension_proxy_tags():
    """
    The tags behavior is a tad complex, so they get their own test.
    """
    foo_tag_uri = "asdf://somewhere.org/extensions/full/tags/foo-1.0"
    foo_tag_def = TagDefinition(
        foo_tag_uri,
        schema_uris="asdf://somewhere.org/extensions/full/schemas/foo-1.0",
        title="Some tag title",
        description="Some tag description",
    )

    bar_tag_uri = "asdf://somewhere.org/extensions/full/tags/bar-1.0"
    bar_tag_def = TagDefinition(
        bar_tag_uri,
        schema_uris="asdf://somewhere.org/extensions/full/schemas/bar-1.0",
        title="Some other tag title",
        description="Some other tag description",
    )

    # The converter should return only the tags
    # supported by the extension.
    converter = FullConverter(tags=["**"])
    extension = FullExtension(tags=[foo_tag_def], converters=[converter])
    proxy = ExtensionProxy(extension)
    assert proxy.converters[0].tags == [foo_tag_uri]

    # The converter should not return tags that
    # its patterns do not match.
    converter = FullConverter(tags=["**/foo-1.0"])
    extension = FullExtension(tags=[foo_tag_def, bar_tag_def], converters=[converter])
    proxy = ExtensionProxy(extension)
    assert proxy.converters[0].tags == [foo_tag_uri]

    # The process should still work if the extension property
    # contains str instead of TagDescription.
    converter = FullConverter(tags=["**/foo-1.0"])
    extension = FullExtension(tags=[foo_tag_uri, bar_tag_uri], converters=[converter])
    proxy = ExtensionProxy(extension)
    assert proxy.converters[0].tags == [foo_tag_uri]


def test_extension_proxy_hash_and_eq():
    extension = MinimumExtension()
    proxy1 = ExtensionProxy(extension)
    proxy2 = ExtensionProxy(extension, package_name="foo", package_version="1.2.3")

    assert proxy1 == proxy2
    assert hash(proxy1) == hash(proxy2)
    assert proxy1 != extension
    assert proxy2 != extension


def test_extension_proxy_repr():
    proxy = ExtensionProxy(MinimumExtension(), package_name="foo", package_version="1.2.3")
    assert "class: asdf._tests.test_extension.MinimumExtension" in repr(proxy)
    assert "package: foo==1.2.3" in repr(proxy)
    assert "legacy: False" in repr(proxy)

    proxy = ExtensionProxy(MinimumExtension())
    assert "class: asdf._tests.test_extension.MinimumExtension" in repr(proxy)
    assert "package: (none)" in repr(proxy)
    assert "legacy: False" in repr(proxy)


def test_extension_manager():
    converter1 = FullConverter(
        tags=[
            "asdf://somewhere.org/extensions/full/tags/foo-*",
            "asdf://somewhere.org/extensions/full/tags/bar-*",
        ],
        types=[
            FooType,
            "asdf._tests.test_extension.BarType",
        ],
    )
    converter2 = FullConverter(
        tags=[
            "asdf://somewhere.org/extensions/full/tags/baz-*",
        ],
        types=[BazType],
    )
    converter3 = FullConverter(
        tags=[
            "asdf://somewhere.org/extensions/full/tags/foo-*",
        ],
        types=[
            FooType,
            BarType,
        ],
    )
    extension1 = FullExtension(
        converters=[converter1, converter2],
        tags=[
            "asdf://somewhere.org/extensions/full/tags/foo-1.0",
            "asdf://somewhere.org/extensions/full/tags/baz-1.0",
        ],
    )
    extension2 = FullExtension(
        converters=[converter3],
        tags=[
            "asdf://somewhere.org/extensions/full/tags/foo-1.0",
        ],
    )

    manager = ExtensionManager([extension1, extension2])

    assert manager.extensions == [ExtensionProxy(extension1), ExtensionProxy(extension2)]

    assert manager.handles_tag("asdf://somewhere.org/extensions/full/tags/foo-1.0") is True
    assert manager.handles_tag("asdf://somewhere.org/extensions/full/tags/bar-1.0") is False
    assert manager.handles_tag("asdf://somewhere.org/extensions/full/tags/baz-1.0") is True

    assert manager.handles_type(FooType) is True
    assert manager.handles_type(SubFooType) is False
    # This should return True even though BarType was listed
    # as string class name:
    assert manager.handles_type(BarType) is True
    assert manager.handles_type(BazType) is True

    assert (
        manager.get_tag_definition("asdf://somewhere.org/extensions/full/tags/foo-1.0").tag_uri
        == "asdf://somewhere.org/extensions/full/tags/foo-1.0"
    )
    assert (
        manager.get_tag_definition("asdf://somewhere.org/extensions/full/tags/baz-1.0").tag_uri
        == "asdf://somewhere.org/extensions/full/tags/baz-1.0"
    )
    with pytest.raises(KeyError, match=r"No support available for YAML tag.*"):
        manager.get_tag_definition("asdf://somewhere.org/extensions/full/tags/bar-1.0")

    assert manager.get_converter_for_tag("asdf://somewhere.org/extensions/full/tags/foo-1.0").delegate is converter1
    assert manager.get_converter_for_tag("asdf://somewhere.org/extensions/full/tags/baz-1.0").delegate is converter2
    with pytest.raises(KeyError, match=r"No support available for YAML tag.*"):
        manager.get_converter_for_tag("asdf://somewhere.org/extensions/full/tags/bar-1.0")

    assert manager.get_converter_for_type(FooType).delegate is converter1
    assert manager.get_converter_for_type(BarType).delegate is converter1
    assert manager.get_converter_for_type(BazType).delegate is converter2
    with pytest.raises(KeyError, match=r"\"No support available for Python type .*\""):
        manager.get_converter_for_type(object)
    with pytest.raises(KeyError, match=r"\"No support available for Python type .*\""):
        manager.get_converter_for_type(SubFooType)


def test_get_cached_extension_manager():
    extension = MinimumExtension()
    extension_manager = get_cached_extension_manager([extension])
    assert get_cached_extension_manager([extension]) is extension_manager
    assert get_cached_extension_manager([MinimumExtension()]) is not extension_manager


def test_tag_definition():
    tag_def = TagDefinition(
        "asdf://somewhere.org/extensions/foo/tags/foo-1.0",
        schema_uris="asdf://somewhere.org/extensions/foo/schemas/foo-1.0",
        title="Some title",
        description="Some description",
    )

    assert tag_def.tag_uri == "asdf://somewhere.org/extensions/foo/tags/foo-1.0"
    assert tag_def.schema_uris == ["asdf://somewhere.org/extensions/foo/schemas/foo-1.0"]
    assert tag_def.title == "Some title"
    assert tag_def.description == "Some description"

    assert "URI: asdf://somewhere.org/extensions/foo/tags/foo-1.0" in repr(tag_def)

    tag_def = TagDefinition(
        "asdf://somewhere.org/extensions/foo/tags/foo-1.0",
        schema_uris=[
            "asdf://somewhere.org/extensions/foo/schemas/foo-1.0",
            "asdf://somewhere.org/extensions/foo/schemas/base-1.0",
        ],
        title="Some title",
        description="Some description",
    )

    assert tag_def.schema_uris == [
        "asdf://somewhere.org/extensions/foo/schemas/foo-1.0",
        "asdf://somewhere.org/extensions/foo/schemas/base-1.0",
    ]

    with pytest.raises(ValueError, match=r"URI patterns are not permitted in TagDefinition"):
        TagDefinition("asdf://somewhere.org/extensions/foo/tags/foo-*")


def test_converter():
    class ConverterNoSubclass:
        tags = []
        types = []

        def to_yaml_tree(self, *args):
            pass

        def from_yaml_tree(self, *args):
            pass

    assert issubclass(ConverterNoSubclass, Converter)


def test_converter_proxy():
    # Test the minimum set of converter methods:
    extension = ExtensionProxy(MinimumExtension())
    converter = MinimumConverter()
    proxy = ConverterProxy(converter, extension)

    assert isinstance(proxy, Converter)

    assert proxy.tags == []
    assert proxy.types == []
    assert proxy.to_yaml_tree(None, None, None) == "to_yaml_tree result"
    assert proxy.from_yaml_tree(None, None, None) == "from_yaml_tree result"
    assert proxy.tags == []
    assert proxy.delegate is converter
    assert proxy.extension == extension
    assert proxy.package_name is None
    assert proxy.package_version is None
    assert proxy.class_name == "asdf._tests.test_extension.MinimumConverter"

    # Check the __eq__ and __hash__ behavior:
    assert proxy == ConverterProxy(converter, extension)
    assert proxy != ConverterProxy(MinimumConverter(), extension)
    assert proxy != ConverterProxy(converter, MinimumExtension())
    assert proxy in {ConverterProxy(converter, extension)}
    assert proxy not in {ConverterProxy(MinimumConverter(), extension), ConverterProxy(converter, MinimumExtension())}

    # Check the __repr__:
    assert "class: asdf._tests.test_extension.MinimumConverter" in repr(proxy)
    assert "package: (none)" in repr(proxy)

    # Test the full set of converter methods:
    converter = FullConverter(
        tags=[
            "asdf://somewhere.org/extensions/test/tags/foo-*",
            "asdf://somewhere.org/extensions/test/tags/bar-*",
        ],
        types=[FooType, BarType],
    )

    extension = FullExtension(
        tags=[
            TagDefinition(
                "asdf://somewhere.org/extensions/test/tags/foo-1.0",
                schema_uris="asdf://somewhere.org/extensions/test/schemas/foo-1.0",
                title="Foo tag title",
                description="Foo tag description",
            ),
            TagDefinition(
                "asdf://somewhere.org/extensions/test/tags/bar-1.0",
                schema_uris="asdf://somewhere.org/extensions/test/schemas/bar-1.0",
                title="Bar tag title",
                description="Bar tag description",
            ),
        ],
    )

    extension_proxy = ExtensionProxy(extension, package_name="foo", package_version="1.2.3")
    proxy = ConverterProxy(converter, extension_proxy)
    assert len(proxy.tags) == 2
    assert "asdf://somewhere.org/extensions/test/tags/foo-1.0" in proxy.tags
    assert "asdf://somewhere.org/extensions/test/tags/bar-1.0" in proxy.tags
    assert proxy.types == [FooType, BarType]
    assert proxy.to_yaml_tree(None, None, None) == "to_yaml_tree result"
    assert proxy.from_yaml_tree(None, None, None) == "from_yaml_tree result"
    assert proxy.select_tag(None, None) == "select_tag result"
    assert proxy.delegate is converter
    assert proxy.extension == extension_proxy
    assert proxy.package_name == "foo"
    assert proxy.package_version == "1.2.3"
    assert proxy.class_name == "asdf._tests.test_extension.FullConverter"

    # Check the __repr__ since it will contain package info now:
    assert "class: asdf._tests.test_extension.FullConverter" in repr(proxy)
    assert "package: foo==1.2.3" in repr(proxy)

    # Should error because object() does fulfill the Converter interface:
    with pytest.raises(TypeError, match=r"Converter must implement the .*"):
        ConverterProxy(object(), extension)

    # Should fail because tags must be str:
    with pytest.raises(TypeError, match=r"Converter property .* must contain str values"):
        ConverterProxy(MinimumConverter(tags=[object()]), extension)

    # Should fail because types must instances of type:
    with pytest.raises(TypeError, match=r"Converter property .* must contain str or type values"):
        # as the code will ignore types if no relevant tags are found
        # include a tag from this extension to make sure the proxy considers
        # the types
        ConverterProxy(MinimumConverter(tags=[extension.tags[0].tag_uri], types=[object()]), extension)


def test_converter_subclass_with_no_supported_tags():
    """
    Adding a Converter to an Extension that doesn't list support for the tags
    associated with the Converter should result in a failure to convert.
    """

    class Foo:
        pass

    class FooConverterWithSubclass(Converter):
        tags = ["asdf://somewhere.org/tags/foo-1.0.0"]
        types = [Foo]

        def to_yaml_tree(self, *args):
            pass

        def from_yaml_tree(self, *args):
            pass

    class FooExtension(Extension):
        tags = []
        converters = [FooConverterWithSubclass()]
        extension_uri = "asdf://somewhere.org/extensions/foo-1.0.0"

    tree = {"obj": Foo()}
    with config_context() as cfg:
        cfg.add_extension(FooExtension())
        with pytest.raises(AsdfSerializationError, match=r"is not serializable by asdf"):
            roundtrip_object(tree)


def test_manifest_extension():
    with config_context() as config:
        minimal_manifest = """%YAML 1.1
---
id: asdf://somewhere.org/manifests/foo
extension_uri: asdf://somewhere.org/extensions/foo
...
"""
        config.add_resource_mapping({"asdf://somewhere.org/extensions/foo": minimal_manifest})
        extension = ManifestExtension.from_uri("asdf://somewhere.org/extensions/foo")
        assert isinstance(extension, Extension)
        assert extension.extension_uri == "asdf://somewhere.org/extensions/foo"
        assert extension.legacy_class_names == []
        assert extension.asdf_standard_requirement is None
        assert extension.converters == []
        assert extension.compressors == []
        assert extension.validators == []
        assert extension.tags == []

        proxy = ExtensionProxy(extension)
        assert proxy.extension_uri == "asdf://somewhere.org/extensions/foo"
        assert proxy.legacy_class_names == set()
        assert proxy.asdf_standard_requirement == SpecifierSet()
        assert proxy.converters == []
        assert proxy.compressors == []
        assert proxy.validators == []
        assert proxy.tags == []

    with config_context() as config:
        full_manifest = """%YAML 1.1
---
id: asdf://somewhere.org/manifests/foo
extension_uri: asdf://somewhere.org/extensions/foo
asdf_standard_requirement:
  gte: 1.6.0
  lt: 2.0.0
tags:
  - asdf://somewhere.org/tags/bar
  - tag_uri: asdf://somewhere.org/tags/baz
    schema_uri: asdf://somewhere.org/schemas/baz
    title: Baz title
    description: Bar description
...
"""
        config.add_resource_mapping({"asdf://somewhere.org/extensions/foo": full_manifest})

        class FooConverter:
            tags = ["asdf://somewhere.org/tags/bar", "asdf://somewhere.org/tags/baz"]
            types = []

            def select_tag(self, *args):
                pass

            def to_yaml_tree(self, *args):
                pass

            def from_yaml_tree(self, *args):
                pass

        converter = FooConverter()
        validator = MinimalValidator()
        compressor = MinimalCompressor()

        extension = ManifestExtension.from_uri(
            "asdf://somewhere.org/extensions/foo",
            legacy_class_names=["foo.extension.LegacyExtension"],
            converters=[converter],
            compressors=[compressor],
            validators=[validator],
        )
        assert extension.extension_uri == "asdf://somewhere.org/extensions/foo"
        assert extension.legacy_class_names == ["foo.extension.LegacyExtension"]
        assert extension.asdf_standard_requirement == SpecifierSet(">=1.6.0,<2.0.0")
        assert extension.converters == [converter]
        assert extension.compressors == [compressor]
        assert extension.validators == [validator]
        assert len(extension.tags) == 2
        assert extension.tags[0] == "asdf://somewhere.org/tags/bar"
        assert extension.tags[1].tag_uri == "asdf://somewhere.org/tags/baz"
        assert extension.tags[1].schema_uris == ["asdf://somewhere.org/schemas/baz"]
        assert extension.tags[1].title == "Baz title"
        assert extension.tags[1].description == "Bar description"

        proxy = ExtensionProxy(extension)
        assert proxy.extension_uri == "asdf://somewhere.org/extensions/foo"
        assert proxy.legacy_class_names == {"foo.extension.LegacyExtension"}
        assert proxy.asdf_standard_requirement == SpecifierSet(">=1.6.0,<2.0.0")
        assert proxy.converters == [ConverterProxy(converter, proxy)]
        assert proxy.compressors == [compressor]
        assert proxy.validators == [validator]
        assert len(proxy.tags) == 2
        assert proxy.tags[0].tag_uri == "asdf://somewhere.org/tags/bar"
        assert proxy.tags[1].tag_uri == "asdf://somewhere.org/tags/baz"
        assert proxy.tags[1].schema_uris == ["asdf://somewhere.org/schemas/baz"]
        assert proxy.tags[1].title == "Baz title"
        assert proxy.tags[1].description == "Bar description"

    with config_context() as config:
        simple_asdf_standard_manifest = """%YAML 1.1
---
id: asdf://somewhere.org/manifests/foo
extension_uri: asdf://somewhere.org/extensions/foo
asdf_standard_requirement: 1.6.0
...
"""
        config.add_resource_mapping({"asdf://somewhere.org/extensions/foo": simple_asdf_standard_manifest})
        extension = ManifestExtension.from_uri("asdf://somewhere.org/extensions/foo")
        assert extension.asdf_standard_requirement == SpecifierSet("==1.6.0")

        proxy = ExtensionProxy(extension)
        assert proxy.asdf_standard_requirement == SpecifierSet("==1.6.0")


def test_validator():
    validator = MinimalValidator()
    extension = FullExtension(validators=[validator])

    failing_schema = """
        type: object
        properties:
          foo:
            fail: true
    """

    passing_schema = """
        type: object
        properties:
          foo:
            fail: false
    """

    with config_context() as config:
        config.add_extension(extension)
        config.add_resource_mapping(
            {
                "asdf://somewhere.org/schemas/failing": failing_schema,
                "asdf://somewhere.org/schemas/passing": passing_schema,
            },
        )

        with AsdfFile(custom_schema="asdf://somewhere.org/schemas/passing") as af:
            af["foo"] = "bar"
            af.validate()

        with AsdfFile(custom_schema="asdf://somewhere.org/schemas/failing") as af:
            af.validate()

            af["foo"] = "bar"
            with pytest.raises(ValidationError, match=r"Node was doomed to fail"):
                af.validate()


def test_converter_deferral():
    class Bar:
        def __init__(self, value):
            self.value = value

    class Foo(Bar):
        pass

    class Baz(Bar):
        pass

    class FooConverter:
        tags = []
        types = [Foo]

        def select_tag(self, *args):
            return None

        def to_yaml_tree(self, obj, tag, ctx):
            # convert Foo instance to Bar
            return Bar(obj.value)

        def from_yaml_tree(self, node, tag, ctx):
            raise NotImplementedError()

    class BarConverter:
        tags = ["asdf://somewhere.org/tags/bar"]
        types = [Bar]

        def to_yaml_tree(self, obj, tag, ctx):
            return {"value": obj.value}

        def from_yaml_tree(self, node, tag, ctx):
            return Bar(node["value"])

    class BazConverter:
        tags = []
        types = [Baz]

        def select_tag(self, *args):
            return None

        def to_yaml_tree(self, obj, tag, ctx):
            return Foo(obj.value)

        def from_yaml_tree(self, node, tag, ctx):
            raise NotImplementedError()

    extension = FullExtension(converters=[FooConverter(), BarConverter(), BazConverter()], tags=BarConverter.tags)
    with config_context() as config:
        config.add_extension(extension)

        foo = Foo(26)
        bar = Bar(42)
        baz = Baz(720)

        bar_rt = roundtrip_object(bar)
        assert isinstance(bar_rt, Bar)
        assert bar_rt.value == bar.value

        foo_rt = roundtrip_object(foo)
        assert isinstance(foo_rt, Bar)
        assert foo_rt.value == foo.value

        baz_rt = roundtrip_object(baz)
        assert isinstance(baz_rt, Bar)
        assert baz_rt.value == baz.value


def test_converter_loop():
    class Bar:
        def __init__(self, value):
            self.value = value

    class Foo(Bar):
        pass

    class Baz(Bar):
        pass

    class FooConverter:
        tags = []
        types = [Foo]

        def select_tag(self, *args):
            return None

        def to_yaml_tree(self, obj, tag, ctx):
            return Bar(obj.value)

        def from_yaml_tree(self, node, tag, ctx):
            raise NotImplementedError()

    class BarConverter:
        tags = []
        types = [Bar]

        def select_tag(self, *args):
            return None

        def to_yaml_tree(self, obj, tag, ctx):
            return Baz(obj.value)

        def from_yaml_tree(self, node, tag, ctx):
            raise NotImplementedError()

    class BazConverter:
        tags = []
        types = [Baz]

        def select_tag(self, *args):
            return None

        def to_yaml_tree(self, obj, tag, ctx):
            return Foo(obj.value)

        def from_yaml_tree(self, node, tag, ctx):
            raise NotImplementedError()

    extension = FullExtension(converters=[FooConverter(), BarConverter(), BazConverter()])
    with config_context() as config:
        config.add_extension(extension)

        for typ in (Foo, Bar, Baz):
            obj = typ(42)
            with pytest.raises(TypeError, match=r"Conversion cycle detected"):
                roundtrip_object(obj)


@pytest.mark.parametrize("is_subclass", [True, False])
@pytest.mark.parametrize("indirect", [True, False])
def test_warning_or_error_for_default_select_tag(is_subclass, indirect):
    class Foo:
        pass

    ParentClass = Converter if is_subclass else object

    if indirect:

        class IntermediateClass(ParentClass):
            pass

        ParentClass = IntermediateClass

    class FooConverter(ParentClass):
        tags = ["asdf://somewhere.org/tags/foo-*"]
        types = [Foo]

        def to_yaml_tree(self, obj, tag, ctx):
            return {}

        def from_yaml_tree(self, node, tag, ctx):
            return Foo()

    tags = [
        "asdf://somewhere.org/tags/foo-1.0.0",
        "asdf://somewhere.org/tags/foo-2.0.0",
    ]
    extension = FullExtension(converters=[FooConverter()], tags=tags)
    with config_context() as config:
        with pytest.raises(RuntimeError, match="Converter handles multiple tags"):
            config.add_extension(extension)


def test_reference_cycle(tmp_path, with_lazy_tree):
    class FractionWithInverse(fractions.Fraction):
        def __init__(self, *args, **kwargs):
            self._inverse = None

        @property
        def inverse(self):
            return self._inverse

        @inverse.setter
        def inverse(self, value):
            self._inverse = value

    class FractionWithInverseConverter:
        tags = ["asdf://example.com/fractions/tags/fraction-1.0.0"]
        types = [FractionWithInverse]

        def to_yaml_tree(self, obj, tag, ctx):
            return {
                "numerator": obj.numerator,
                "denominator": obj.denominator,
                "inverse": obj.inverse,
            }

        def from_yaml_tree(self, node, tag, ctx):
            obj = FractionWithInverse(node["numerator"], node["denominator"])
            yield obj
            obj.inverse = node["inverse"]

    class FractionWithInverseExtension:
        tags = FractionWithInverseConverter.tags
        converters = [FractionWithInverseConverter()]
        extension_uri = "asdf://example.com/fractions/extensions/fraction-1.0.0"

    with config_context() as cfg:
        cfg.add_extension(FractionWithInverseExtension())

        f1 = FractionWithInverse(3, 5)
        f2 = FractionWithInverse(5, 3)
        f1.inverse = f2
        f2.inverse = f1
        fn = tmp_path / "test.asdf"
        asdf.AsdfFile({"obj": f1}).write_to(fn)
        with asdf.open(fn) as af:
            read_f1 = af["obj"]
            assert read_f1.inverse.inverse is read_f1


def test_manifest_uri_id_mismatch_warning(tmp_path):
    with config_context() as config:
        # make an extension with a manifest (where id doesn't match the registered uri)
        full_manifest = """%YAML 1.1
---
id: asdf://somewhere.org/manifests/foo
extension_uri: asdf://somewhere.org/extensions/foo
tags:
  - asdf://somewhere.org/tags/bar
...
"""
        config.add_resource_mapping({"asdf://somewhere.org/extensions/foo": full_manifest})

        class Foo:
            pass

        class FooConverter:
            tags = ["asdf://somewhere.org/tags/bar"]
            types = [Foo]

            def to_yaml_tree(self, *args):
                return {}

            def from_yaml_tree(self, *args):
                return Foo()

        extension = ManifestExtension.from_uri(
            "asdf://somewhere.org/extensions/foo",
            converters=[FooConverter()],
        )

        # use the extension to write out a file
        config.add_extension(extension)

        af = AsdfFile()
        af["foo"] = Foo()
        fn = tmp_path / "foo.asdf"
        with pytest.warns(AsdfManifestURIMismatchWarning):
            af.write_to(fn)


def test_resolve_type_not_imported():
    path = "mailbox.Mailbox"

    if "mailbox" in sys.modules:
        del sys.modules["mailbox"]

    assert _resolve_type(path) is None

    import mailbox

    assert _resolve_type(path) is mailbox.Mailbox


@pytest.mark.parametrize(
    "path, obj", (("sys", sys), ("asdf.AsdfFile", AsdfFile), ("asdf.Missing", None), ("not_a_module", None))
)
def test_resolve_type(path, obj):
    assert _resolve_type(path) is obj


def test_extension_converter_by_class_path():
    class MailboxConverter:
        tags = ["asdf://example.com/tags/mailbox-1.0.0"]
        types = ["mailbox.Mailbox"]

        def to_yaml_tree(self, obj, tag, ctx):
            return {}

        def from_yaml_tree(self, node, tag, ctx):
            return None

    class MailboxExtension:
        tags = MailboxConverter.tags
        converters = [MailboxConverter()]
        extension_uri = "asdf://example.com/extensions/mailbox-1.0.0"

    # grab the type so we can use it for extension_manager.get_converter_for_type
    import mailbox

    typ = mailbox.Mailbox
    del sys.modules["mailbox"], mailbox

    with config_context() as cfg:
        cfg.add_extension(MailboxExtension())
        extension_manager = AsdfFile().extension_manager

        # make sure that registering the extension did not load the module
        assert "mailbox" not in sys.modules

        # as the module hasn't been loaded, the converter shouldn't be found
        with pytest.raises(KeyError, match="No support available for Python type 'mailbox.Mailbox'"):
            extension_manager.get_converter_for_type(typ)

        # make sure inspecting the type didn't import the module
        assert "mailbox" not in sys.modules

        # finally, import the module and check that the converter can now be found
        import mailbox

        converter = extension_manager.get_converter_for_type(mailbox.Mailbox)
        assert isinstance(converter.delegate, MailboxConverter)


def test_named_tuple_extension():
    Point = collections.namedtuple("Point", ["x", "y"])

    class PointConverter:
        tags = ["asdf://example.com/tags/point-1.0.0"]
        types = [Point]

        def to_yaml_tree(self, obj, tag, ctx):
            return list(obj)

        def from_yaml_tree(self, node, tag, ctx):
            return Point(*node)

    class PointExtension:
        tags = PointConverter.tags
        converters = [PointConverter()]
        extension_uri = "asdf://example.com/extensions/point-1.0.0"

    pt = Point(1, 2)

    # without the extension we can't serialize this
    with pytest.raises(AsdfSerializationError, match="is not serializable by asdf"):
        roundtrip_object(pt)

    with config_context() as cfg:
        cfg.add_extension(PointExtension())
        rpt = roundtrip_object(pt)
        assert isinstance(rpt, Point)