diff --git a/docs/changes.rst b/docs/changes.rst index 22876b1..16ca81f 100644 --- a/docs/changes.rst +++ b/docs/changes.rst @@ -2,6 +2,12 @@ Changelog ========= +Unreleased +---------- + +* Make ``Registry.get_or_retrieve`` recognize empty-fragment URI aliases for + resources already present in, or found by crawling, the registry. + v0.37.0 ------- diff --git a/referencing/_core.py b/referencing/_core.py index db926a2..5b9dbdd 100644 --- a/referencing/_core.py +++ b/referencing/_core.py @@ -410,12 +410,24 @@ def get_or_retrieve(self, uri: URI) -> Retrieved[D, Resource[D]]: so the returned object is a `Retrieved` object which contains both the resource value as well as the registry which ultimately contained it. """ + # Keep exact-key lookups first for compatibility with registries + # constructed directly from a resource mapping. Resources added via + # ``with_resource(s)`` normalize empty trailing fragments, so fall + # back to that spelling when the requested URI has one. resource = self._resources.get(uri) + if resource is None: + base, separator, fragment = uri.rpartition("#") + if separator and not fragment and "#" not in base: + resource = self._resources.get(base) if resource is not None: return Retrieved(registry=self, value=resource) registry = self.crawl() resource = registry._resources.get(uri) + if resource is None: + base, separator, fragment = uri.rpartition("#") + if separator and not fragment and "#" not in base: + resource = registry._resources.get(base) if resource is not None: return Retrieved(registry=registry, value=resource) diff --git a/referencing/tests/test_core.py b/referencing/tests/test_core.py index 3edddbc..af2b0cd 100644 --- a/referencing/tests/test_core.py +++ b/referencing/tests/test_core.py @@ -518,6 +518,116 @@ def test_retrieve_already_available_resource(self): assert registry["urn:example"] == foo assert registry.get_or_retrieve("urn:example").value == foo + def test_retrieve_already_available_resource_with_empty_fragment(self): + foo = Resource.opaque({"foo": "bar"}) + registry = Registry(retrieve=blow_up).with_resource("urn:foo#", foo) + retrieved = registry.get_or_retrieve("urn:foo#") + + assert retrieved.value is foo + assert retrieved.registry is registry + + def test_retrieve_crawlable_resource_with_empty_fragment(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) + retrieved = registry.get_or_retrieve("urn:child#") + + assert retrieved.value == child + + def test_retrieve_empty_fragment_preserves_original_callback_uri(self): + foo = Resource.opaque({"foo": "bar"}) + calls = [] + + def retrieve(uri): + calls.append(uri) + return foo + + registry = Registry(retrieve=retrieve) + retrieved = registry.get_or_retrieve("urn:new#") + + assert retrieved.value is foo + assert calls == ["urn:new#"] + assert retrieved.registry.contents("urn:new") == {"foo": "bar"} + assert retrieved.registry.get_or_retrieve("urn:new#").value is foo + assert calls == ["urn:new#"] + + def test_retrieve_constructor_raw_key_has_priority(self): + canonical = Resource.opaque({"value": "canonical"}) + raw = Resource.opaque({"value": "raw"}) + registry = Registry( + {"urn:foo": canonical, "urn:foo#": raw}, + retrieve=blow_up, + ) + + assert registry.get_or_retrieve("urn:foo#").value is raw + assert registry.get_or_retrieve("urn:foo").value is canonical + + def test_retrieve_nonempty_and_escaped_fragments_are_unchanged(self): + nonempty = Resource.opaque({"fragment": "nonempty"}) + escaped = Resource.opaque({"fragment": "escaped"}) + registry = Registry(retrieve=blow_up).with_resources( + [("urn:foo#bar", nonempty), ("urn:foo%23", escaped)], + ) + + assert registry.get_or_retrieve("urn:foo#bar").value is nonempty + assert registry.get_or_retrieve("urn:foo%23").value is escaped + + def test_retrieve_multiple_or_nonempty_trailing_fragments_are_not_aliases( + self, + ): + calls = [] + known = Resource.opaque({"uri": "known"}) + known_fragment = Resource.opaque({"uri": "known#bar"}) + + def retrieve(uri): + calls.append(uri) + return Resource.opaque({"uri": uri}) + + registry = Registry(retrieve=retrieve).with_resources( + [("urn:foo", known), ("urn:foo#bar", known_fragment)], + ) + assert registry.get_or_retrieve("urn:foo#bar#").value.contents == { + "uri": "urn:foo#bar#", + } + assert registry.get_or_retrieve("urn:foo##").value.contents == { + "uri": "urn:foo##", + } + assert calls == ["urn:foo#bar#", "urn:foo##"] + assert registry["urn:foo"] is known + assert registry["urn:foo#bar"] is known_fragment + + def test_retrieve_crawl_does_not_alias_multiple_or_nonempty_fragments( + self, + ): + calls = [] + known = ID_AND_CHILDREN.create_resource({"ID": "urn:foo"}) + known_fragment = ID_AND_CHILDREN.create_resource({"ID": "urn:foo#bar"}) + + def retrieve(uri): + calls.append(uri) + return Resource.opaque({"uri": uri}) + + root = ID_AND_CHILDREN.create_resource( + {"children": [known.contents, known_fragment.contents]}, + ) + registry = Registry(retrieve=retrieve).with_resource("urn:root", root) + assert registry.get_or_retrieve("urn:foo#bar#").value.contents == { + "uri": "urn:foo#bar#", + } + assert registry.get_or_retrieve("urn:foo##").value.contents == { + "uri": "urn:foo##", + } + assert calls == ["urn:foo#bar#", "urn:foo##"] + + def test_retrieve_empty_fragment_no_such_resource_preserves_ref(self): + def retrieve(uri): + raise exceptions.NoSuchResource(ref=uri) + + with pytest.raises(exceptions.NoSuchResource) as error: + Registry(retrieve=retrieve).get_or_retrieve("urn:missing#") + + assert error.value == exceptions.NoSuchResource(ref="urn:missing#") + 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]})