diff options
Diffstat (limited to '.venv/lib/python3.12/site-packages/referencing/tests')
6 files changed, 1645 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/referencing/tests/__init__.py b/.venv/lib/python3.12/site-packages/referencing/tests/__init__.py new file mode 100644 index 00000000..e69de29b --- /dev/null +++ b/.venv/lib/python3.12/site-packages/referencing/tests/__init__.py diff --git a/.venv/lib/python3.12/site-packages/referencing/tests/test_core.py b/.venv/lib/python3.12/site-packages/referencing/tests/test_core.py new file mode 100644 index 00000000..3edddbc3 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/referencing/tests/test_core.py @@ -0,0 +1,1057 @@ +from rpds import HashTrieMap +import pytest + +from referencing import Anchor, Registry, Resource, Specification, exceptions +from referencing.jsonschema import DRAFT202012 + +ID_AND_CHILDREN = Specification( + name="id-and-children", + id_of=lambda contents: contents.get("ID"), + subresources_of=lambda contents: contents.get("children", []), + anchors_in=lambda specification, contents: [ + Anchor( + name=name, + resource=specification.create_resource(contents=each), + ) + for name, each in contents.get("anchors", {}).items() + ], + maybe_in_subresource=lambda segments, resolver, subresource: ( + resolver.in_subresource(subresource) + if not len(segments) % 2 + and all(each == "children" for each in segments[::2]) + else resolver + ), +) + + +def blow_up(uri): # pragma: no cover + """ + A retriever suitable for use in tests which expect it never to be used. + """ + raise RuntimeError("This retrieve function expects to never be called!") + + +class TestRegistry: + def test_with_resource(self): + """ + Adding a resource to the registry then allows re-retrieving it. + """ + + resource = Resource.opaque(contents={"foo": "bar"}) + uri = "urn:example" + registry = Registry().with_resource(uri=uri, resource=resource) + assert registry[uri] is resource + + def test_with_resources(self): + """ + Adding multiple resources to the registry is like adding each one. + """ + + one = Resource.opaque(contents={}) + two = Resource(contents={"foo": "bar"}, specification=ID_AND_CHILDREN) + registry = Registry().with_resources( + [ + ("http://example.com/1", one), + ("http://example.com/foo/bar", two), + ], + ) + assert registry == Registry().with_resource( + uri="http://example.com/1", + resource=one, + ).with_resource( + uri="http://example.com/foo/bar", + resource=two, + ) + + def test_matmul_resource(self): + uri = "urn:example:resource" + resource = ID_AND_CHILDREN.create_resource({"ID": uri, "foo": 12}) + registry = resource @ Registry() + assert registry == Registry().with_resource(uri, resource) + + def test_matmul_many_resources(self): + one_uri = "urn:example:one" + one = ID_AND_CHILDREN.create_resource({"ID": one_uri, "foo": 12}) + + two_uri = "urn:example:two" + two = ID_AND_CHILDREN.create_resource({"ID": two_uri, "foo": 12}) + + registry = [one, two] @ Registry() + assert registry == Registry().with_resources( + [(one_uri, one), (two_uri, two)], + ) + + def test_matmul_resource_without_id(self): + resource = Resource.opaque(contents={"foo": "bar"}) + with pytest.raises(exceptions.NoInternalID) as e: + resource @ Registry() + assert e.value == exceptions.NoInternalID(resource=resource) + + def test_with_contents_from_json_schema(self): + uri = "urn:example" + schema = {"$schema": "https://json-schema.org/draft/2020-12/schema"} + registry = Registry().with_contents([(uri, schema)]) + + expected = Resource(contents=schema, specification=DRAFT202012) + assert registry[uri] == expected + + def test_with_contents_and_default_specification(self): + uri = "urn:example" + registry = Registry().with_contents( + [(uri, {"foo": "bar"})], + default_specification=Specification.OPAQUE, + ) + assert registry[uri] == Resource.opaque({"foo": "bar"}) + + def test_len(self): + total = 5 + registry = Registry().with_contents( + [(str(i), {"foo": "bar"}) for i in range(total)], + default_specification=Specification.OPAQUE, + ) + assert len(registry) == total + + def test_bool_empty(self): + assert not Registry() + + def test_bool_not_empty(self): + registry = Registry().with_contents( + [(str(i), {"foo": "bar"}) for i in range(3)], + default_specification=Specification.OPAQUE, + ) + assert registry + + def test_iter(self): + registry = Registry().with_contents( + [(str(i), {"foo": "bar"}) for i in range(8)], + default_specification=Specification.OPAQUE, + ) + assert set(registry) == {str(i) for i in range(8)} + + def test_crawl_still_has_top_level_resource(self): + resource = Resource.opaque({"foo": "bar"}) + uri = "urn:example" + registry = Registry({uri: resource}).crawl() + assert registry[uri] is resource + + def test_crawl_finds_a_subresource(self): + child_id = "urn:child" + root = ID_AND_CHILDREN.create_resource( + {"ID": "urn:root", "children": [{"ID": child_id, "foo": 12}]}, + ) + registry = root @ Registry() + with pytest.raises(LookupError): + registry[child_id] + + expected = ID_AND_CHILDREN.create_resource({"ID": child_id, "foo": 12}) + assert registry.crawl()[child_id] == expected + + def test_crawl_finds_anchors_with_id(self): + resource = ID_AND_CHILDREN.create_resource( + {"ID": "urn:bar", "anchors": {"foo": 12}}, + ) + registry = resource @ Registry() + + assert registry.crawl().anchor(resource.id(), "foo").value == Anchor( + name="foo", + resource=ID_AND_CHILDREN.create_resource(12), + ) + + def test_crawl_finds_anchors_no_id(self): + resource = ID_AND_CHILDREN.create_resource({"anchors": {"foo": 12}}) + registry = Registry().with_resource("urn:root", resource) + + assert registry.crawl().anchor("urn:root", "foo").value == Anchor( + name="foo", + resource=ID_AND_CHILDREN.create_resource(12), + ) + + def test_contents(self): + resource = Resource.opaque({"foo": "bar"}) + uri = "urn:example" + registry = Registry().with_resource(uri, resource) + assert registry.contents(uri) == {"foo": "bar"} + + def test_getitem_strips_empty_fragments(self): + uri = "http://example.com/" + resource = ID_AND_CHILDREN.create_resource({"ID": uri + "#"}) + registry = resource @ Registry() + assert registry[uri] == registry[uri + "#"] == resource + + def test_contents_strips_empty_fragments(self): + uri = "http://example.com/" + resource = ID_AND_CHILDREN.create_resource({"ID": uri + "#"}) + registry = resource @ Registry() + assert ( + registry.contents(uri) + == registry.contents(uri + "#") + == {"ID": uri + "#"} + ) + + def test_contents_nonexistent_resource(self): + registry = Registry() + with pytest.raises(exceptions.NoSuchResource) as e: + registry.contents("urn:example") + assert e.value == exceptions.NoSuchResource(ref="urn:example") + + def test_crawled_anchor(self): + resource = ID_AND_CHILDREN.create_resource({"anchors": {"foo": "bar"}}) + registry = Registry().with_resource("urn:example", resource) + retrieved = registry.anchor("urn:example", "foo") + assert retrieved.value == Anchor( + name="foo", + resource=ID_AND_CHILDREN.create_resource("bar"), + ) + assert retrieved.registry == registry.crawl() + + def test_anchor_in_nonexistent_resource(self): + registry = Registry() + with pytest.raises(exceptions.NoSuchResource) as e: + registry.anchor("urn:example", "foo") + assert e.value == exceptions.NoSuchResource(ref="urn:example") + + def test_init(self): + one = Resource.opaque(contents={}) + two = ID_AND_CHILDREN.create_resource({"foo": "bar"}) + registry = Registry( + { + "http://example.com/1": one, + "http://example.com/foo/bar": two, + }, + ) + assert ( + registry + == Registry() + .with_resources( + [ + ("http://example.com/1", one), + ("http://example.com/foo/bar", two), + ], + ) + .crawl() + ) + + def test_dict_conversion(self): + """ + Passing a `dict` to `Registry` gets converted to a `HashTrieMap`. + + So continuing to use the registry works. + """ + + one = Resource.opaque(contents={}) + two = ID_AND_CHILDREN.create_resource({"foo": "bar"}) + registry = Registry( + {"http://example.com/1": one}, + ).with_resource("http://example.com/foo/bar", two) + assert ( + registry.crawl() + == Registry() + .with_resources( + [ + ("http://example.com/1", one), + ("http://example.com/foo/bar", two), + ], + ) + .crawl() + ) + + def test_no_such_resource(self): + registry = Registry() + with pytest.raises(exceptions.NoSuchResource) as e: + registry["urn:bigboom"] + assert e.value == exceptions.NoSuchResource(ref="urn:bigboom") + + def test_combine(self): + one = Resource.opaque(contents={}) + two = ID_AND_CHILDREN.create_resource({"foo": "bar"}) + three = ID_AND_CHILDREN.create_resource({"baz": "quux"}) + four = ID_AND_CHILDREN.create_resource({"anchors": {"foo": 12}}) + + first = Registry({"http://example.com/1": one}) + second = Registry().with_resource("http://example.com/foo/bar", two) + third = Registry( + { + "http://example.com/1": one, + "http://example.com/baz": three, + }, + ) + fourth = ( + Registry() + .with_resource( + "http://example.com/foo/quux", + four, + ) + .crawl() + ) + assert first.combine(second, third, fourth) == Registry( + [ + ("http://example.com/1", one), + ("http://example.com/baz", three), + ("http://example.com/foo/quux", four), + ], + anchors=HashTrieMap( + { + ("http://example.com/foo/quux", "foo"): Anchor( + name="foo", + resource=ID_AND_CHILDREN.create_resource(12), + ), + }, + ), + ).with_resource("http://example.com/foo/bar", two) + + def test_combine_self(self): + """ + Combining a registry with itself short-circuits. + + This is a performance optimization -- otherwise we do lots more work + (in jsonschema this seems to correspond to making the test suite take + *3x* longer). + """ + + registry = Registry({"urn:foo": "bar"}) + assert registry.combine(registry) is registry + + def test_combine_with_uncrawled_resources(self): + one = Resource.opaque(contents={}) + two = ID_AND_CHILDREN.create_resource({"foo": "bar"}) + three = ID_AND_CHILDREN.create_resource({"baz": "quux"}) + + first = Registry().with_resource("http://example.com/1", one) + second = Registry().with_resource("http://example.com/foo/bar", two) + third = Registry( + { + "http://example.com/1": one, + "http://example.com/baz": three, + }, + ) + expected = Registry( + [ + ("http://example.com/1", one), + ("http://example.com/foo/bar", two), + ("http://example.com/baz", three), + ], + ) + combined = first.combine(second, third) + assert combined != expected + assert combined.crawl() == expected + + def test_combine_with_single_retrieve(self): + one = Resource.opaque(contents={}) + two = ID_AND_CHILDREN.create_resource({"foo": "bar"}) + three = ID_AND_CHILDREN.create_resource({"baz": "quux"}) + + def retrieve(uri): # pragma: no cover + pass + + first = Registry().with_resource("http://example.com/1", one) + second = Registry( + retrieve=retrieve, + ).with_resource("http://example.com/2", two) + third = Registry().with_resource("http://example.com/3", three) + + assert first.combine(second, third) == Registry( + retrieve=retrieve, + ).with_resources( + [ + ("http://example.com/1", one), + ("http://example.com/2", two), + ("http://example.com/3", three), + ], + ) + assert second.combine(first, third) == Registry( + retrieve=retrieve, + ).with_resources( + [ + ("http://example.com/1", one), + ("http://example.com/2", two), + ("http://example.com/3", three), + ], + ) + + def test_combine_with_common_retrieve(self): + one = Resource.opaque(contents={}) + two = ID_AND_CHILDREN.create_resource({"foo": "bar"}) + three = ID_AND_CHILDREN.create_resource({"baz": "quux"}) + + def retrieve(uri): # pragma: no cover + pass + + first = Registry(retrieve=retrieve).with_resource( + "http://example.com/1", + one, + ) + second = Registry( + retrieve=retrieve, + ).with_resource("http://example.com/2", two) + third = Registry(retrieve=retrieve).with_resource( + "http://example.com/3", + three, + ) + + assert first.combine(second, third) == Registry( + retrieve=retrieve, + ).with_resources( + [ + ("http://example.com/1", one), + ("http://example.com/2", two), + ("http://example.com/3", three), + ], + ) + assert second.combine(first, third) == Registry( + retrieve=retrieve, + ).with_resources( + [ + ("http://example.com/1", one), + ("http://example.com/2", two), + ("http://example.com/3", three), + ], + ) + + def test_combine_conflicting_retrieve(self): + one = Resource.opaque(contents={}) + two = ID_AND_CHILDREN.create_resource({"foo": "bar"}) + three = ID_AND_CHILDREN.create_resource({"baz": "quux"}) + + def foo_retrieve(uri): # pragma: no cover + pass + + def bar_retrieve(uri): # pragma: no cover + pass + + first = Registry(retrieve=foo_retrieve).with_resource( + "http://example.com/1", + one, + ) + second = Registry().with_resource("http://example.com/2", two) + third = Registry(retrieve=bar_retrieve).with_resource( + "http://example.com/3", + three, + ) + + with pytest.raises(Exception, match="conflict.*retriev"): + first.combine(second, third) + + def test_remove(self): + one = Resource.opaque(contents={}) + two = ID_AND_CHILDREN.create_resource({"foo": "bar"}) + registry = Registry({"urn:foo": one, "urn:bar": two}) + assert registry.remove("urn:foo") == Registry({"urn:bar": two}) + + def test_remove_uncrawled(self): + one = Resource.opaque(contents={}) + two = ID_AND_CHILDREN.create_resource({"foo": "bar"}) + registry = Registry().with_resources( + [("urn:foo", one), ("urn:bar", two)], + ) + assert registry.remove("urn:foo") == Registry().with_resource( + "urn:bar", + two, + ) + + def test_remove_with_anchors(self): + one = Resource.opaque(contents={}) + two = ID_AND_CHILDREN.create_resource({"anchors": {"foo": "bar"}}) + registry = ( + Registry() + .with_resources( + [("urn:foo", one), ("urn:bar", two)], + ) + .crawl() + ) + assert ( + registry.remove("urn:bar") + == Registry() + .with_resource( + "urn:foo", + one, + ) + .crawl() + ) + + def test_remove_nonexistent_uri(self): + with pytest.raises(exceptions.NoSuchResource) as e: + Registry().remove("urn:doesNotExist") + assert e.value == exceptions.NoSuchResource(ref="urn:doesNotExist") + + def test_retrieve(self): + foo = Resource.opaque({"foo": "bar"}) + registry = Registry(retrieve=lambda uri: foo) + assert registry.get_or_retrieve("urn:example").value == foo + + def test_retrieve_arbitrary_exception(self): + foo = Resource.opaque({"foo": "bar"}) + + def retrieve(uri): + if uri == "urn:succeed": + return foo + raise Exception("Oh no!") + + registry = Registry(retrieve=retrieve) + assert registry.get_or_retrieve("urn:succeed").value == foo + with pytest.raises(exceptions.Unretrievable): + registry.get_or_retrieve("urn:uhoh") + + def test_retrieve_no_such_resource(self): + foo = Resource.opaque({"foo": "bar"}) + + def retrieve(uri): + if uri == "urn:succeed": + return foo + raise exceptions.NoSuchResource(ref=uri) + + registry = Registry(retrieve=retrieve) + assert registry.get_or_retrieve("urn:succeed").value == foo + with pytest.raises(exceptions.NoSuchResource): + registry.get_or_retrieve("urn:uhoh") + + def test_retrieve_cannot_determine_specification(self): + def retrieve(uri): + return Resource.from_contents({}) + + registry = Registry(retrieve=retrieve) + with pytest.raises(exceptions.CannotDetermineSpecification): + registry.get_or_retrieve("urn:uhoh") + + def test_retrieve_already_available_resource(self): + foo = Resource.opaque({"foo": "bar"}) + registry = Registry({"urn:example": foo}, retrieve=blow_up) + assert registry["urn:example"] == foo + assert registry.get_or_retrieve("urn:example").value == foo + + def test_retrieve_first_checks_crawlable_resource(self): + child = ID_AND_CHILDREN.create_resource({"ID": "urn:child", "foo": 12}) + root = ID_AND_CHILDREN.create_resource({"children": [child.contents]}) + registry = Registry(retrieve=blow_up).with_resource("urn:root", root) + assert registry.crawl()["urn:child"] == child + + def test_resolver(self): + one = Resource.opaque(contents={}) + registry = Registry({"http://example.com": one}) + resolver = registry.resolver(base_uri="http://example.com") + assert resolver.lookup("#").contents == {} + + def test_resolver_with_root_identified(self): + root = ID_AND_CHILDREN.create_resource({"ID": "http://example.com"}) + resolver = Registry().resolver_with_root(root) + assert resolver.lookup("http://example.com").contents == root.contents + assert resolver.lookup("#").contents == root.contents + + def test_resolver_with_root_unidentified(self): + root = Resource.opaque(contents={}) + resolver = Registry().resolver_with_root(root) + assert resolver.lookup("#").contents == root.contents + + def test_repr(self): + one = Resource.opaque(contents={}) + two = ID_AND_CHILDREN.create_resource({"foo": "bar"}) + registry = Registry().with_resources( + [ + ("http://example.com/1", one), + ("http://example.com/foo/bar", two), + ], + ) + assert repr(registry) == "<Registry (2 uncrawled resources)>" + assert repr(registry.crawl()) == "<Registry (2 resources)>" + + def test_repr_mixed_crawled(self): + one = Resource.opaque(contents={}) + two = ID_AND_CHILDREN.create_resource({"foo": "bar"}) + registry = ( + Registry( + {"http://example.com/1": one}, + ) + .crawl() + .with_resource(uri="http://example.com/foo/bar", resource=two) + ) + assert repr(registry) == "<Registry (2 resources, 1 uncrawled)>" + + def test_repr_one_resource(self): + registry = Registry().with_resource( + uri="http://example.com/1", + resource=Resource.opaque(contents={}), + ) + assert repr(registry) == "<Registry (1 uncrawled resource)>" + + def test_repr_empty(self): + assert repr(Registry()) == "<Registry (0 resources)>" + + +class TestResource: + def test_from_contents_from_json_schema(self): + schema = {"$schema": "https://json-schema.org/draft/2020-12/schema"} + resource = Resource.from_contents(schema) + assert resource == Resource(contents=schema, specification=DRAFT202012) + + def test_from_contents_with_no_discernible_information(self): + """ + Creating a resource with no discernible way to see what + specification it belongs to (e.g. no ``$schema`` keyword for JSON + Schema) raises an error. + """ + + with pytest.raises(exceptions.CannotDetermineSpecification): + Resource.from_contents({"foo": "bar"}) + + def test_from_contents_with_no_discernible_information_and_default(self): + resource = Resource.from_contents( + {"foo": "bar"}, + default_specification=Specification.OPAQUE, + ) + assert resource == Resource.opaque(contents={"foo": "bar"}) + + def test_from_contents_unneeded_default(self): + schema = {"$schema": "https://json-schema.org/draft/2020-12/schema"} + resource = Resource.from_contents( + schema, + default_specification=Specification.OPAQUE, + ) + assert resource == Resource( + contents=schema, + specification=DRAFT202012, + ) + + def test_non_mapping_from_contents(self): + resource = Resource.from_contents( + True, + default_specification=ID_AND_CHILDREN, + ) + assert resource == Resource( + contents=True, + specification=ID_AND_CHILDREN, + ) + + def test_from_contents_with_fallback(self): + resource = Resource.from_contents( + {"foo": "bar"}, + default_specification=Specification.OPAQUE, + ) + assert resource == Resource.opaque(contents={"foo": "bar"}) + + def test_id_delegates_to_specification(self): + specification = Specification( + name="", + id_of=lambda contents: "urn:fixedID", + subresources_of=lambda contents: [], + anchors_in=lambda specification, contents: [], + maybe_in_subresource=( + lambda segments, resolver, subresource: resolver + ), + ) + resource = Resource( + contents={"foo": "baz"}, + specification=specification, + ) + assert resource.id() == "urn:fixedID" + + def test_id_strips_empty_fragment(self): + uri = "http://example.com/" + root = ID_AND_CHILDREN.create_resource({"ID": uri + "#"}) + assert root.id() == uri + + def test_subresources_delegates_to_specification(self): + resource = ID_AND_CHILDREN.create_resource({"children": [{}, 12]}) + assert list(resource.subresources()) == [ + ID_AND_CHILDREN.create_resource(each) for each in [{}, 12] + ] + + def test_subresource_with_different_specification(self): + schema = {"$schema": "https://json-schema.org/draft/2020-12/schema"} + resource = ID_AND_CHILDREN.create_resource({"children": [schema]}) + assert list(resource.subresources()) == [ + DRAFT202012.create_resource(schema), + ] + + def test_anchors_delegates_to_specification(self): + resource = ID_AND_CHILDREN.create_resource( + {"anchors": {"foo": {}, "bar": 1, "baz": ""}}, + ) + assert list(resource.anchors()) == [ + Anchor(name="foo", resource=ID_AND_CHILDREN.create_resource({})), + Anchor(name="bar", resource=ID_AND_CHILDREN.create_resource(1)), + Anchor(name="baz", resource=ID_AND_CHILDREN.create_resource("")), + ] + + def test_pointer_to_mapping(self): + resource = Resource.opaque(contents={"foo": "baz"}) + resolver = Registry().resolver() + assert resource.pointer("/foo", resolver=resolver).contents == "baz" + + def test_pointer_to_array(self): + resource = Resource.opaque(contents={"foo": {"bar": [3]}}) + resolver = Registry().resolver() + assert resource.pointer("/foo/bar/0", resolver=resolver).contents == 3 + + def test_root_pointer(self): + contents = {"foo": "baz"} + resource = Resource.opaque(contents=contents) + resolver = Registry().resolver() + assert resource.pointer("", resolver=resolver).contents == contents + + def test_opaque(self): + contents = {"foo": "bar"} + assert Resource.opaque(contents) == Resource( + contents=contents, + specification=Specification.OPAQUE, + ) + + +class TestResolver: + def test_lookup_exact_uri(self): + resource = Resource.opaque(contents={"foo": "baz"}) + resolver = Registry({"http://example.com/1": resource}).resolver() + resolved = resolver.lookup("http://example.com/1") + assert resolved.contents == resource.contents + + def test_lookup_subresource(self): + root = ID_AND_CHILDREN.create_resource( + { + "ID": "http://example.com/", + "children": [ + {"ID": "http://example.com/a", "foo": 12}, + ], + }, + ) + registry = root @ Registry() + resolved = registry.resolver().lookup("http://example.com/a") + assert resolved.contents == {"ID": "http://example.com/a", "foo": 12} + + def test_lookup_anchor_with_id(self): + root = ID_AND_CHILDREN.create_resource( + { + "ID": "http://example.com/", + "anchors": {"foo": 12}, + }, + ) + registry = root @ Registry() + resolved = registry.resolver().lookup("http://example.com/#foo") + assert resolved.contents == 12 + + def test_lookup_anchor_without_id(self): + root = ID_AND_CHILDREN.create_resource({"anchors": {"foo": 12}}) + resolver = Registry().with_resource("urn:example", root).resolver() + resolved = resolver.lookup("urn:example#foo") + assert resolved.contents == 12 + + def test_lookup_unknown_reference(self): + resolver = Registry().resolver() + ref = "http://example.com/does/not/exist" + with pytest.raises(exceptions.Unresolvable) as e: + resolver.lookup(ref) + assert e.value == exceptions.Unresolvable(ref=ref) + + def test_lookup_non_existent_pointer(self): + resource = Resource.opaque({"foo": {}}) + resolver = Registry({"http://example.com/1": resource}).resolver() + ref = "http://example.com/1#/foo/bar" + with pytest.raises(exceptions.Unresolvable) as e: + resolver.lookup(ref) + assert e.value == exceptions.PointerToNowhere( + ref="/foo/bar", + resource=resource, + ) + assert str(e.value) == "'/foo/bar' does not exist within {'foo': {}}" + + def test_lookup_non_existent_pointer_to_array_index(self): + resource = Resource.opaque([1, 2, 4, 8]) + resolver = Registry({"http://example.com/1": resource}).resolver() + ref = "http://example.com/1#/10" + with pytest.raises(exceptions.Unresolvable) as e: + resolver.lookup(ref) + assert e.value == exceptions.PointerToNowhere( + ref="/10", + resource=resource, + ) + + def test_lookup_pointer_to_empty_string(self): + resolver = Registry().resolver_with_root(Resource.opaque({"": {}})) + assert resolver.lookup("#/").contents == {} + + def test_lookup_non_existent_pointer_to_empty_string(self): + resource = Resource.opaque({"foo": {}}) + resolver = Registry().resolver_with_root(resource) + with pytest.raises( + exceptions.Unresolvable, + match="^'/' does not exist within {'foo': {}}.*'#'", + ) as e: + resolver.lookup("#/") + assert e.value == exceptions.PointerToNowhere( + ref="/", + resource=resource, + ) + + def test_lookup_non_existent_anchor(self): + root = ID_AND_CHILDREN.create_resource({"anchors": {}}) + resolver = Registry().with_resource("urn:example", root).resolver() + resolved = resolver.lookup("urn:example") + assert resolved.contents == root.contents + + ref = "urn:example#noSuchAnchor" + with pytest.raises(exceptions.Unresolvable) as e: + resolver.lookup(ref) + assert "'noSuchAnchor' does not exist" in str(e.value) + assert e.value == exceptions.NoSuchAnchor( + ref="urn:example", + resource=root, + anchor="noSuchAnchor", + ) + + def test_lookup_invalid_JSON_pointerish_anchor(self): + resolver = Registry().resolver_with_root( + ID_AND_CHILDREN.create_resource( + { + "ID": "http://example.com/", + "foo": {"bar": 12}, + }, + ), + ) + + valid = resolver.lookup("#/foo/bar") + assert valid.contents == 12 + + with pytest.raises(exceptions.InvalidAnchor) as e: + resolver.lookup("#foo/bar") + assert " '#/foo/bar'" in str(e.value) + + def test_lookup_retrieved_resource(self): + resource = Resource.opaque(contents={"foo": "baz"}) + resolver = Registry(retrieve=lambda uri: resource).resolver() + resolved = resolver.lookup("http://example.com/") + assert resolved.contents == resource.contents + + def test_lookup_failed_retrieved_resource(self): + """ + Unretrievable exceptions are also wrapped in Unresolvable. + """ + + uri = "http://example.com/" + + registry = Registry(retrieve=blow_up) + with pytest.raises(exceptions.Unretrievable): + registry.get_or_retrieve(uri) + + resolver = registry.resolver() + with pytest.raises(exceptions.Unresolvable): + resolver.lookup(uri) + + def test_repeated_lookup_from_retrieved_resource(self): + """ + A (custom-)retrieved resource is added to the registry returned by + looking it up. + """ + resource = Resource.opaque(contents={"foo": "baz"}) + once = [resource] + + def retrieve(uri): + return once.pop() + + resolver = Registry(retrieve=retrieve).resolver() + resolved = resolver.lookup("http://example.com/") + assert resolved.contents == resource.contents + + resolved = resolved.resolver.lookup("http://example.com/") + assert resolved.contents == resource.contents + + def test_repeated_anchor_lookup_from_retrieved_resource(self): + resource = Resource.opaque(contents={"foo": "baz"}) + once = [resource] + + def retrieve(uri): + return once.pop() + + resolver = Registry(retrieve=retrieve).resolver() + resolved = resolver.lookup("http://example.com/") + assert resolved.contents == resource.contents + + resolved = resolved.resolver.lookup("#") + assert resolved.contents == resource.contents + + # FIXME: The tests below aren't really representable in the current + # suite, though we should probably think of ways to do so. + + def test_in_subresource(self): + root = ID_AND_CHILDREN.create_resource( + { + "ID": "http://example.com/", + "children": [ + { + "ID": "child/", + "children": [{"ID": "grandchild"}], + }, + ], + }, + ) + registry = root @ Registry() + + resolver = registry.resolver() + first = resolver.lookup("http://example.com/") + assert first.contents == root.contents + + with pytest.raises(exceptions.Unresolvable): + first.resolver.lookup("grandchild") + + sub = first.resolver.in_subresource( + ID_AND_CHILDREN.create_resource(first.contents["children"][0]), + ) + second = sub.lookup("grandchild") + assert second.contents == {"ID": "grandchild"} + + def test_in_pointer_subresource(self): + root = ID_AND_CHILDREN.create_resource( + { + "ID": "http://example.com/", + "children": [ + { + "ID": "child/", + "children": [{"ID": "grandchild"}], + }, + ], + }, + ) + registry = root @ Registry() + + resolver = registry.resolver() + first = resolver.lookup("http://example.com/") + assert first.contents == root.contents + + with pytest.raises(exceptions.Unresolvable): + first.resolver.lookup("grandchild") + + second = first.resolver.lookup("#/children/0") + third = second.resolver.lookup("grandchild") + assert third.contents == {"ID": "grandchild"} + + def test_dynamic_scope(self): + one = ID_AND_CHILDREN.create_resource( + { + "ID": "http://example.com/", + "children": [ + { + "ID": "child/", + "children": [{"ID": "grandchild"}], + }, + ], + }, + ) + two = ID_AND_CHILDREN.create_resource( + { + "ID": "http://example.com/two", + "children": [{"ID": "two-child/"}], + }, + ) + registry = [one, two] @ Registry() + + resolver = registry.resolver() + first = resolver.lookup("http://example.com/") + second = first.resolver.lookup("#/children/0") + third = second.resolver.lookup("grandchild") + fourth = third.resolver.lookup("http://example.com/two") + assert list(fourth.resolver.dynamic_scope()) == [ + ("http://example.com/child/grandchild", fourth.resolver._registry), + ("http://example.com/child/", fourth.resolver._registry), + ("http://example.com/", fourth.resolver._registry), + ] + assert list(third.resolver.dynamic_scope()) == [ + ("http://example.com/child/", third.resolver._registry), + ("http://example.com/", third.resolver._registry), + ] + assert list(second.resolver.dynamic_scope()) == [ + ("http://example.com/", second.resolver._registry), + ] + assert list(first.resolver.dynamic_scope()) == [] + + +class TestSpecification: + def test_create_resource(self): + specification = Specification( + name="", + id_of=lambda contents: "urn:fixedID", + subresources_of=lambda contents: [], + anchors_in=lambda specification, contents: [], + maybe_in_subresource=( + lambda segments, resolver, subresource: resolver + ), + ) + resource = specification.create_resource(contents={"foo": "baz"}) + assert resource == Resource( + contents={"foo": "baz"}, + specification=specification, + ) + assert resource.id() == "urn:fixedID" + + def test_detect_from_json_schema(self): + schema = {"$schema": "https://json-schema.org/draft/2020-12/schema"} + specification = Specification.detect(schema) + assert specification == DRAFT202012 + + def test_detect_with_no_discernible_information(self): + with pytest.raises(exceptions.CannotDetermineSpecification): + Specification.detect({"foo": "bar"}) + + def test_detect_with_non_URI_schema(self): + with pytest.raises(exceptions.CannotDetermineSpecification): + Specification.detect({"$schema": 37}) + + def test_detect_with_no_discernible_information_and_default(self): + specification = Specification.OPAQUE.detect({"foo": "bar"}) + assert specification is Specification.OPAQUE + + def test_detect_unneeded_default(self): + schema = {"$schema": "https://json-schema.org/draft/2020-12/schema"} + specification = Specification.OPAQUE.detect(schema) + assert specification == DRAFT202012 + + def test_non_mapping_detect(self): + with pytest.raises(exceptions.CannotDetermineSpecification): + Specification.detect(True) + + def test_non_mapping_detect_with_default(self): + specification = ID_AND_CHILDREN.detect(True) + assert specification is ID_AND_CHILDREN + + def test_detect_with_fallback(self): + specification = Specification.OPAQUE.detect({"foo": "bar"}) + assert specification is Specification.OPAQUE + + def test_repr(self): + assert ( + repr(ID_AND_CHILDREN) == "<Specification name='id-and-children'>" + ) + + +class TestOpaqueSpecification: + THINGS = [{"foo": "bar"}, True, 37, "foo", object()] + + @pytest.mark.parametrize("thing", THINGS) + def test_no_id(self, thing): + """ + An arbitrary thing has no ID. + """ + + assert Specification.OPAQUE.id_of(thing) is None + + @pytest.mark.parametrize("thing", THINGS) + def test_no_subresources(self, thing): + """ + An arbitrary thing has no subresources. + """ + + assert list(Specification.OPAQUE.subresources_of(thing)) == [] + + @pytest.mark.parametrize("thing", THINGS) + def test_no_anchors(self, thing): + """ + An arbitrary thing has no anchors. + """ + + assert list(Specification.OPAQUE.anchors_in(thing)) == [] + + +@pytest.mark.parametrize( + "cls", + [Anchor, Registry, Resource, Specification, exceptions.PointerToNowhere], +) +def test_nonsubclassable(cls): + with pytest.raises(Exception, match="(?i)subclassing"): + + class Boom(cls): # pragma: no cover + pass diff --git a/.venv/lib/python3.12/site-packages/referencing/tests/test_exceptions.py b/.venv/lib/python3.12/site-packages/referencing/tests/test_exceptions.py new file mode 100644 index 00000000..85cf99ec --- /dev/null +++ b/.venv/lib/python3.12/site-packages/referencing/tests/test_exceptions.py @@ -0,0 +1,34 @@ +import itertools + +import pytest + +from referencing import Resource, exceptions + + +def pairs(choices): + return itertools.combinations(choices, 2) + + +TRUE = Resource.opaque(True) + + +thunks = ( + lambda: exceptions.CannotDetermineSpecification(TRUE), + lambda: exceptions.NoSuchResource("urn:example:foo"), + lambda: exceptions.NoInternalID(TRUE), + lambda: exceptions.InvalidAnchor(resource=TRUE, anchor="foo", ref="a#b"), + lambda: exceptions.NoSuchAnchor(resource=TRUE, anchor="foo", ref="a#b"), + lambda: exceptions.PointerToNowhere(resource=TRUE, ref="urn:example:foo"), + lambda: exceptions.Unresolvable("urn:example:foo"), + lambda: exceptions.Unretrievable("urn:example:foo"), +) + + +@pytest.mark.parametrize("one, two", pairs(each() for each in thunks)) +def test_eq_incompatible_types(one, two): + assert one != two + + +@pytest.mark.parametrize("thunk", thunks) +def test_hash(thunk): + assert thunk() in {thunk()} diff --git a/.venv/lib/python3.12/site-packages/referencing/tests/test_jsonschema.py b/.venv/lib/python3.12/site-packages/referencing/tests/test_jsonschema.py new file mode 100644 index 00000000..c80714d0 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/referencing/tests/test_jsonschema.py @@ -0,0 +1,382 @@ +import pytest + +from referencing import Registry, Resource, Specification +import referencing.jsonschema + + +@pytest.mark.parametrize( + "uri, expected", + [ + ( + "https://json-schema.org/draft/2020-12/schema", + referencing.jsonschema.DRAFT202012, + ), + ( + "https://json-schema.org/draft/2019-09/schema", + referencing.jsonschema.DRAFT201909, + ), + ( + "http://json-schema.org/draft-07/schema#", + referencing.jsonschema.DRAFT7, + ), + ( + "http://json-schema.org/draft-06/schema#", + referencing.jsonschema.DRAFT6, + ), + ( + "http://json-schema.org/draft-04/schema#", + referencing.jsonschema.DRAFT4, + ), + ( + "http://json-schema.org/draft-03/schema#", + referencing.jsonschema.DRAFT3, + ), + ], +) +def test_schemas_with_explicit_schema_keywords_are_detected(uri, expected): + """ + The $schema keyword in JSON Schema is a dialect identifier. + """ + contents = {"$schema": uri} + resource = Resource.from_contents(contents) + assert resource == Resource(contents=contents, specification=expected) + + +def test_unknown_dialect(): + dialect_id = "http://example.com/unknown-json-schema-dialect-id" + with pytest.raises(referencing.jsonschema.UnknownDialect) as excinfo: + Resource.from_contents({"$schema": dialect_id}) + assert excinfo.value.uri == dialect_id + + +@pytest.mark.parametrize( + "id, specification", + [ + ("$id", referencing.jsonschema.DRAFT202012), + ("$id", referencing.jsonschema.DRAFT201909), + ("$id", referencing.jsonschema.DRAFT7), + ("$id", referencing.jsonschema.DRAFT6), + ("id", referencing.jsonschema.DRAFT4), + ("id", referencing.jsonschema.DRAFT3), + ], +) +def test_id_of_mapping(id, specification): + uri = "http://example.com/some-schema" + assert specification.id_of({id: uri}) == uri + + +@pytest.mark.parametrize( + "specification", + [ + referencing.jsonschema.DRAFT202012, + referencing.jsonschema.DRAFT201909, + referencing.jsonschema.DRAFT7, + referencing.jsonschema.DRAFT6, + ], +) +@pytest.mark.parametrize("value", [True, False]) +def test_id_of_bool(specification, value): + assert specification.id_of(value) is None + + +@pytest.mark.parametrize( + "specification", + [ + referencing.jsonschema.DRAFT202012, + referencing.jsonschema.DRAFT201909, + referencing.jsonschema.DRAFT7, + referencing.jsonschema.DRAFT6, + ], +) +@pytest.mark.parametrize("value", [True, False]) +def test_anchors_in_bool(specification, value): + assert list(specification.anchors_in(value)) == [] + + +@pytest.mark.parametrize( + "specification", + [ + referencing.jsonschema.DRAFT202012, + referencing.jsonschema.DRAFT201909, + referencing.jsonschema.DRAFT7, + referencing.jsonschema.DRAFT6, + ], +) +@pytest.mark.parametrize("value", [True, False]) +def test_subresources_of_bool(specification, value): + assert list(specification.subresources_of(value)) == [] + + +@pytest.mark.parametrize( + "uri, expected", + [ + ( + "https://json-schema.org/draft/2020-12/schema", + referencing.jsonschema.DRAFT202012, + ), + ( + "https://json-schema.org/draft/2019-09/schema", + referencing.jsonschema.DRAFT201909, + ), + ( + "http://json-schema.org/draft-07/schema#", + referencing.jsonschema.DRAFT7, + ), + ( + "http://json-schema.org/draft-06/schema#", + referencing.jsonschema.DRAFT6, + ), + ( + "http://json-schema.org/draft-04/schema#", + referencing.jsonschema.DRAFT4, + ), + ( + "http://json-schema.org/draft-03/schema#", + referencing.jsonschema.DRAFT3, + ), + ], +) +def test_specification_with(uri, expected): + assert referencing.jsonschema.specification_with(uri) == expected + + +@pytest.mark.parametrize( + "uri, expected", + [ + ( + "http://json-schema.org/draft-07/schema", + referencing.jsonschema.DRAFT7, + ), + ( + "http://json-schema.org/draft-06/schema", + referencing.jsonschema.DRAFT6, + ), + ( + "http://json-schema.org/draft-04/schema", + referencing.jsonschema.DRAFT4, + ), + ( + "http://json-schema.org/draft-03/schema", + referencing.jsonschema.DRAFT3, + ), + ], +) +def test_specification_with_no_empty_fragment(uri, expected): + assert referencing.jsonschema.specification_with(uri) == expected + + +def test_specification_with_unknown_dialect(): + dialect_id = "http://example.com/unknown-json-schema-dialect-id" + with pytest.raises(referencing.jsonschema.UnknownDialect) as excinfo: + referencing.jsonschema.specification_with(dialect_id) + assert excinfo.value.uri == dialect_id + + +def test_specification_with_default(): + dialect_id = "http://example.com/unknown-json-schema-dialect-id" + specification = referencing.jsonschema.specification_with( + dialect_id, + default=Specification.OPAQUE, + ) + assert specification is Specification.OPAQUE + + +# FIXME: The tests below should move to the referencing suite but I haven't yet +# figured out how to represent dynamic (& recursive) ref lookups in it. +def test_lookup_trivial_dynamic_ref(): + one = referencing.jsonschema.DRAFT202012.create_resource( + {"$dynamicAnchor": "foo"}, + ) + resolver = Registry().with_resource("http://example.com", one).resolver() + resolved = resolver.lookup("http://example.com#foo") + assert resolved.contents == one.contents + + +def test_multiple_lookup_trivial_dynamic_ref(): + TRUE = referencing.jsonschema.DRAFT202012.create_resource(True) + root = referencing.jsonschema.DRAFT202012.create_resource( + { + "$id": "http://example.com", + "$dynamicAnchor": "fooAnchor", + "$defs": { + "foo": { + "$id": "foo", + "$dynamicAnchor": "fooAnchor", + "$defs": { + "bar": True, + "baz": { + "$dynamicAnchor": "fooAnchor", + }, + }, + }, + }, + }, + ) + resolver = ( + Registry() + .with_resources( + [ + ("http://example.com", root), + ("http://example.com/foo/", TRUE), + ("http://example.com/foo/bar", root), + ], + ) + .resolver() + ) + + first = resolver.lookup("http://example.com") + second = first.resolver.lookup("foo/") + resolver = second.resolver.lookup("bar").resolver + fourth = resolver.lookup("#fooAnchor") + assert fourth.contents == root.contents + + +def test_multiple_lookup_dynamic_ref_to_nondynamic_ref(): + one = referencing.jsonschema.DRAFT202012.create_resource( + {"$anchor": "fooAnchor"}, + ) + two = referencing.jsonschema.DRAFT202012.create_resource( + { + "$id": "http://example.com", + "$dynamicAnchor": "fooAnchor", + "$defs": { + "foo": { + "$id": "foo", + "$dynamicAnchor": "fooAnchor", + "$defs": { + "bar": True, + "baz": { + "$dynamicAnchor": "fooAnchor", + }, + }, + }, + }, + }, + ) + resolver = ( + Registry() + .with_resources( + [ + ("http://example.com", two), + ("http://example.com/foo/", one), + ("http://example.com/foo/bar", two), + ], + ) + .resolver() + ) + + first = resolver.lookup("http://example.com") + second = first.resolver.lookup("foo/") + resolver = second.resolver.lookup("bar").resolver + fourth = resolver.lookup("#fooAnchor") + assert fourth.contents == two.contents + + +def test_lookup_trivial_recursive_ref(): + one = referencing.jsonschema.DRAFT201909.create_resource( + {"$recursiveAnchor": True}, + ) + resolver = Registry().with_resource("http://example.com", one).resolver() + first = resolver.lookup("http://example.com") + resolved = referencing.jsonschema.lookup_recursive_ref( + resolver=first.resolver, + ) + assert resolved.contents == one.contents + + +def test_lookup_recursive_ref_to_bool(): + TRUE = referencing.jsonschema.DRAFT201909.create_resource(True) + registry = Registry({"http://example.com": TRUE}) + resolved = referencing.jsonschema.lookup_recursive_ref( + resolver=registry.resolver(base_uri="http://example.com"), + ) + assert resolved.contents == TRUE.contents + + +def test_multiple_lookup_recursive_ref_to_bool(): + TRUE = referencing.jsonschema.DRAFT201909.create_resource(True) + root = referencing.jsonschema.DRAFT201909.create_resource( + { + "$id": "http://example.com", + "$recursiveAnchor": True, + "$defs": { + "foo": { + "$id": "foo", + "$recursiveAnchor": True, + "$defs": { + "bar": True, + "baz": { + "$recursiveAnchor": True, + "$anchor": "fooAnchor", + }, + }, + }, + }, + }, + ) + resolver = ( + Registry() + .with_resources( + [ + ("http://example.com", root), + ("http://example.com/foo/", TRUE), + ("http://example.com/foo/bar", root), + ], + ) + .resolver() + ) + + first = resolver.lookup("http://example.com") + second = first.resolver.lookup("foo/") + resolver = second.resolver.lookup("bar").resolver + fourth = referencing.jsonschema.lookup_recursive_ref(resolver=resolver) + assert fourth.contents == root.contents + + +def test_multiple_lookup_recursive_ref_with_nonrecursive_ref(): + one = referencing.jsonschema.DRAFT201909.create_resource( + {"$recursiveAnchor": True}, + ) + two = referencing.jsonschema.DRAFT201909.create_resource( + { + "$id": "http://example.com", + "$recursiveAnchor": True, + "$defs": { + "foo": { + "$id": "foo", + "$recursiveAnchor": True, + "$defs": { + "bar": True, + "baz": { + "$recursiveAnchor": True, + "$anchor": "fooAnchor", + }, + }, + }, + }, + }, + ) + three = referencing.jsonschema.DRAFT201909.create_resource( + {"$recursiveAnchor": False}, + ) + resolver = ( + Registry() + .with_resources( + [ + ("http://example.com", three), + ("http://example.com/foo/", two), + ("http://example.com/foo/bar", one), + ], + ) + .resolver() + ) + + first = resolver.lookup("http://example.com") + second = first.resolver.lookup("foo/") + resolver = second.resolver.lookup("bar").resolver + fourth = referencing.jsonschema.lookup_recursive_ref(resolver=resolver) + assert fourth.contents == two.contents + + +def test_empty_registry(): + assert referencing.jsonschema.EMPTY_REGISTRY == Registry() diff --git a/.venv/lib/python3.12/site-packages/referencing/tests/test_referencing_suite.py b/.venv/lib/python3.12/site-packages/referencing/tests/test_referencing_suite.py new file mode 100644 index 00000000..4b8ae917 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/referencing/tests/test_referencing_suite.py @@ -0,0 +1,66 @@ +from pathlib import Path +import json +import os + +import pytest + +from referencing import Registry +from referencing.exceptions import Unresolvable +import referencing.jsonschema + + +class SuiteNotFound(Exception): + def __str__(self): # pragma: no cover + return ( + "Cannot find the referencing suite. " + "Set the REFERENCING_SUITE environment variable to the path to " + "the suite, or run the test suite from alongside a full checkout " + "of the git repository." + ) + + +if "REFERENCING_SUITE" in os.environ: # pragma: no cover + SUITE = Path(os.environ["REFERENCING_SUITE"]) / "tests" +else: + SUITE = Path(__file__).parent.parent.parent / "suite/tests" +if not SUITE.is_dir(): # pragma: no cover + raise SuiteNotFound() +DIALECT_IDS = json.loads(SUITE.joinpath("specifications.json").read_text()) + + +@pytest.mark.parametrize( + "test_path", + [ + pytest.param(each, id=f"{each.parent.name}-{each.stem}") + for each in SUITE.glob("*/**/*.json") + ], +) +def test_referencing_suite(test_path, subtests): + dialect_id = DIALECT_IDS[test_path.relative_to(SUITE).parts[0]] + specification = referencing.jsonschema.specification_with(dialect_id) + loaded = json.loads(test_path.read_text()) + registry = loaded["registry"] + registry = Registry().with_resources( + (uri, specification.create_resource(contents)) + for uri, contents in loaded["registry"].items() + ) + for test in loaded["tests"]: + with subtests.test(test=test): + if "normalization" in test_path.stem: + pytest.xfail("APIs need to change for proper URL support.") + + resolver = registry.resolver(base_uri=test.get("base_uri", "")) + + if test.get("error"): + with pytest.raises(Unresolvable): + resolver.lookup(test["ref"]) + else: + resolved = resolver.lookup(test["ref"]) + assert resolved.contents == test["target"] + + then = test.get("then") + while then: # pragma: no cover + with subtests.test(test=test, then=then): + resolved = resolved.resolver.lookup(then["ref"]) + assert resolved.contents == then["target"] + then = then.get("then") diff --git a/.venv/lib/python3.12/site-packages/referencing/tests/test_retrieval.py b/.venv/lib/python3.12/site-packages/referencing/tests/test_retrieval.py new file mode 100644 index 00000000..d0a8f8ad --- /dev/null +++ b/.venv/lib/python3.12/site-packages/referencing/tests/test_retrieval.py @@ -0,0 +1,106 @@ +from functools import lru_cache +import json + +import pytest + +from referencing import Registry, Resource, exceptions +from referencing.jsonschema import DRAFT202012 +from referencing.retrieval import to_cached_resource + + +class TestToCachedResource: + def test_it_caches_retrieved_resources(self): + contents = {"$schema": "https://json-schema.org/draft/2020-12/schema"} + stack = [json.dumps(contents)] + + @to_cached_resource() + def retrieve(uri): + return stack.pop() + + registry = Registry(retrieve=retrieve) + + expected = Resource.from_contents(contents) + + got = registry.get_or_retrieve("urn:example:schema") + assert got.value == expected + + # And a second time we get the same value. + again = registry.get_or_retrieve("urn:example:schema") + assert again.value is got.value + + def test_custom_loader(self): + contents = {"$schema": "https://json-schema.org/draft/2020-12/schema"} + stack = [json.dumps(contents)[::-1]] + + @to_cached_resource(loads=lambda s: json.loads(s[::-1])) + def retrieve(uri): + return stack.pop() + + registry = Registry(retrieve=retrieve) + + expected = Resource.from_contents(contents) + + got = registry.get_or_retrieve("urn:example:schema") + assert got.value == expected + + # And a second time we get the same value. + again = registry.get_or_retrieve("urn:example:schema") + assert again.value is got.value + + def test_custom_from_contents(self): + contents = {} + stack = [json.dumps(contents)] + + @to_cached_resource(from_contents=DRAFT202012.create_resource) + def retrieve(uri): + return stack.pop() + + registry = Registry(retrieve=retrieve) + + expected = DRAFT202012.create_resource(contents) + + got = registry.get_or_retrieve("urn:example:schema") + assert got.value == expected + + # And a second time we get the same value. + again = registry.get_or_retrieve("urn:example:schema") + assert again.value is got.value + + def test_custom_cache(self): + schema = {"$schema": "https://json-schema.org/draft/2020-12/schema"} + mapping = { + "urn:example:1": dict(schema, foo=1), + "urn:example:2": dict(schema, foo=2), + "urn:example:3": dict(schema, foo=3), + } + + resources = { + uri: Resource.from_contents(contents) + for uri, contents in mapping.items() + } + + @to_cached_resource(cache=lru_cache(maxsize=2)) + def retrieve(uri): + return json.dumps(mapping.pop(uri)) + + registry = Registry(retrieve=retrieve) + + got = registry.get_or_retrieve("urn:example:1") + assert got.value == resources["urn:example:1"] + assert registry.get_or_retrieve("urn:example:1").value is got.value + assert registry.get_or_retrieve("urn:example:1").value is got.value + + got = registry.get_or_retrieve("urn:example:2") + assert got.value == resources["urn:example:2"] + assert registry.get_or_retrieve("urn:example:2").value is got.value + assert registry.get_or_retrieve("urn:example:2").value is got.value + + # This still succeeds, but evicts the first URI + got = registry.get_or_retrieve("urn:example:3") + assert got.value == resources["urn:example:3"] + assert registry.get_or_retrieve("urn:example:3").value is got.value + assert registry.get_or_retrieve("urn:example:3").value is got.value + + # And now this fails (as we popped the value out of `mapping`) + with pytest.raises(exceptions.Unretrievable): + registry.get_or_retrieve("urn:example:1") |