fixed bug and adjusted tests

Author: notactuallyfinn

src/hermes/commands/deposit/base.py

@@ -38,7 +38,7 @@ def __call__(self, command: HermesCommand) -> None:
         deposit = self.map_metadata()
         self.ctx.prepare_step("deposit")
         with self.ctx[command.settings.target] as cache:
-            cache["deposit"] = deposit.compact()
+            cache["deposit"] = deposit
         self.ctx.finalize_step("deposit")
 
         if self.is_initial_publication():
@@ -48,10 +48,8 @@ def __call__(self, command: HermesCommand) -> None:
 
         deposit = self.update_metadata()
         self.ctx.prepare_step("deposit")
-        with self.ctx[command.settings.target] as cache:
-            cache["codemeta"] = deposit.compact()
-            cache["expanded"] = deposit.ld_value
-            cache["context"] = {"@context": deposit.full_context}
+        with self.ctx["deposit"] as cache:
+            cache["result"] = deposit
         self.ctx.finalize_step("deposit")
         self.delete_artifacts()
         self.upload_artifacts()
@@ -67,7 +65,7 @@ def prepare(self) -> None:
         pass
 
     @abc.abstractmethod
-    def map_metadata(self) -> SoftwareMetadata:
+    def map_metadata(self) -> dict:
         """Map the given metadata to the target schema of the deposition platform and return it.
 
         When mapping metadata, make sure to add traces to the HERMES software, e.g. via
@@ -97,9 +95,10 @@ def create_new_version(self) -> None:
         """Create a new version of an existing publication on the target platform."""
         pass
 
-    def update_metadata(self) -> SoftwareMetadata:
+    @abc.abstractmethod
+    def update_metadata(self) -> dict:
         """Update the metadata of the newly created version and return it even if it hasn't changed."""
-        return self.metadata
+        pass
 
     def delete_artifacts(self) -> None:
         """Delete any superfluous artifacts taken from the previous version of the publication."""

src/hermes/commands/deposit/file.py

@@ -11,7 +11,7 @@
 from pydantic import BaseModel
 
 from hermes.commands.deposit.base import BaseDepositPlugin
-from hermes.model import SoftwareMetadata
+
 
 class FileDepositSettings(BaseModel):
     filename: str = 'codemeta.json'
@@ -20,8 +20,11 @@ class FileDepositSettings(BaseModel):
 class FileDepositPlugin(BaseDepositPlugin):
     settings_class = FileDepositSettings
 
-    def map_metadata(self) -> SoftwareMetadata:
-        return self.metadata
+    def map_metadata(self) -> dict:
+        return self.metadata.compact()
+
+    def update_metadata(self) -> dict:
+        return self.metadata.compact()
 
     def publish(self) -> None:
         file_config = self.command.settings.file

src/hermes/commands/deposit/invenio.py

@@ -19,7 +19,6 @@
 from hermes.commands.deposit.base import BaseDepositPlugin
 from hermes.commands.deposit.error import DepositionUnauthorizedError
 from hermes.error import MisconfigurationError
-from hermes.model import SoftwareMetadata
 from hermes.model.error import HermesValidationError
 from hermes.utils import hermes_doi, hermes_user_agent
 
@@ -320,7 +319,12 @@ def prepare(self) -> None:
             record_id=rec_id, doi=doi, codemeta_identifier=codemeta_identifier
         )
 
-        version = self.metadata["version"]
+        if len(self.metadata.get("version", [])) > 1:
+            raise HermesValidationError("Too many licenses for invenio deposit.")
+        if len(self.metadata.get("version", [])) == 1:
+            version = self.metadata["version"][0]
+        else:
+            version = None
         if rec_meta and (version == rec_meta.get("version")):
             raise ValueError(f"Version {version} already deposited.")
 
@@ -336,10 +340,10 @@ def prepare(self) -> None:
 
         self.invenio_ctx = deposition_data
 
-    def map_metadata(self) -> SoftwareMetadata:
+    def map_metadata(self) -> dict:
         """Map the harvested metadata onto the Invenio schema and return it."""
         self.invenio_ctx["depositionMetadata"] = self._codemeta_to_invenio_deposition()
-        return SoftwareMetadata(self.invenio_ctx["depositionMetadata"])
+        return self.invenio_ctx["depositionMetadata"]
 
     def is_initial_publication(self) -> bool:
         latest_record_id = self.invenio_ctx.get("latestRecord", {}).get("id")
@@ -398,7 +402,7 @@ def related_identifiers(self):
             },
         ]
 
-    def update_metadata(self) -> SoftwareMetadata:
+    def update_metadata(self) -> dict:
         """Update the metadata of a draft and return it."""
 
         draft_url = self.links["latest_draft"]
@@ -418,7 +422,7 @@ def update_metadata(self) -> SoftwareMetadata:
         self.links.update(deposit["links"])
 
         _log.debug("Created new version deposit: %s", self.links["html"])
-        return SoftwareMetadata(deposit.get("metadata", {}))
+        return deposit
 
     def delete_artifacts(self) -> None:
         """Delete existing file artifacts.
@@ -508,21 +512,25 @@ def _codemeta_to_invenio_deposition(self) -> dict:
         access_conditions = self.invenio_ctx["access_conditions"]
 
         creators = []
-        for author in metadata["author"]:
+        for author in metadata.get("author", []):
             creator = {}
-            if len(affils := [name for affil in author["affiliation"] for name in affil["legalname"]]) != 0:
+            if len(
+                affils := [
+                    name for affil in author.get("affiliation", []) for name in affil.get("legalname", [])
+                ]
+            ) != 0:
                 creator["affiliation"] = affils
-            if len(author["familyName"]) > 1:
-                raise HermesValidationError(f"Author has too many family names: {author.to_python()}")
-            if len(author["familyName"]) == 1:
-                given_names_str = " ".join(author["givenName"])
+
+            if len(author.get("familyName", [])) > 1:
+                raise HermesValidationError(f"Author has too many family names: {author}")
+            if len(author.get("familyName", [])) == 1:
+                given_names_str = " ".join(author.get("givenName", []))
                 name = f"{author["familyName"][0]}, {given_names_str}"
-            elif len(author["name"]) != 1:
-                raise HermesValidationError(f"Author has too many names: {author.to_python()}")
+            elif len(author.get("name", [])) != 1:
+                raise HermesValidationError(f"Author has too many or no names: {author}")
             else:
                 name = author["name"][0]
-            if len(name) != 0:
-                creator["name"] = name
+            creator["name"] = name
             if (id := author.get("@id", None)) is not None:
                 creator["orcid"] = id.replace("https://orcid.org/", "")
             if creator:
@@ -545,6 +553,7 @@ def _codemeta_to_invenio_deposition(self) -> dict:
             for author in metadata["author"]
         ]"""
 
+        # TODO: reimplement with new api
         # This is not used at the moment. See comment below in `deposition_metadata` dict.
         contributors = [  # noqa: F841
             # TODO: Distinguish between @type "Person" and others
@@ -566,27 +575,33 @@ def _codemeta_to_invenio_deposition(self) -> dict:
             for contributor in metadata.get("contributor", []) if contributor.get("name") != "GitHub"
         ]
 
-        if len(metadata["name"]) != 1:
+        if len(metadata.get("name", [])) != 1:
             _log.error("More than one or zero names for the Software are given.")
             raise HermesValidationError("More than one or zerno names for the Software.")
         name = metadata["name"][0]
 
-        if len(metadata["schema:description"]) > 1:
+        if len(metadata.get("schema:description", [])) > 1:
             _log.error("More than one descriptions of the Software are given.")
             raise HermesValidationError("More than one descriptions of the Software are given.")
-        if len(metadata["schema:description"]) == 1:
+        if len(metadata.get("schema:description", [])) == 1:
             description = metadata["schema:description"][0]
         else:
             description = None
 
-        if len(metadata["schema:version"]) > 1:
+        if len(metadata.get("schema:version", [])) > 1:
             _log.error("More than one version of the Software are given.")
             raise HermesValidationError("More than one version of the Software are given.")
-        if len(metadata["schema:version"]) == 1:
+        if len(metadata.get("schema:version", [])) == 1:
             version = metadata["schema:version"][0]
         else:
             version = None
 
+        keywords = metadata.get("schema:keywords", [])
+        if len(keywords) == 0:
+            keywords = None
+        else:
+            keywords = keywords.to_python()
+
         # TODO: Use the fields currently set to `None`.
         # Some more fields are available but they most likely don't relate to software
         # publications targeted by hermes.
@@ -602,9 +617,6 @@ def _codemeta_to_invenio_deposition(self) -> dict:
             "publication_date": date.today().isoformat(),
             "title": name,
             "creators": creators,
-            # TODO: Use a real description here. Possible sources could be
-            # `tool.poetry.description` from pyproject.toml or `abstract` from
-            # CITATION.cff. This should then be stored in codemeta description field.
             "description": description,
             "access_right": access_right,
             "license": license,
@@ -618,8 +630,8 @@ def _codemeta_to_invenio_deposition(self) -> dict:
             # them.
             # TODO: Use the DOI we get back from this.
             "prereserve_doi": True,
-            # TODO: A good source for this could be `tool.poetry.keywords` in pyproject.toml.
-            "keywords": None,
+            "keywords": keywords,
+            # TODO: Is there a good codemeta/ schema field?
             "notes": None,
             "related_identifiers": self.related_identifiers(),
             # TODO: Use `contributors`. In the case of the hermes workflow itself, the
@@ -641,6 +653,10 @@ def _get_license_identifier(self) -> Union[str, None]:
 
         If no license is configured, ``None``  will be returned.
         """
+        if "license" not in self.metadata:
+            raise HermesValidationError("No license is given.")
+        if len(self.metadata["license"]) > 1:
+            raise HermesValidationError("Too many licenses for invenio deposit.")
         license_url = self.metadata["license"][0]
         return self.resolver.resolve_license_id(license_url)
 

src/hermes/model/types/ld_dict.py

@@ -22,14 +22,7 @@ def __init__(self, data, *, parent=None, key=None, index=None, context=None):
 
     def __getitem__(self, key):
         full_iri = self.ld_proc.expand_iri(self.active_ctx, key)
-        if full_iri == "@id":
-            return self._to_python(full_iri, self.data_dict[full_iri])
-        try:
-            ld_value = self.data_dict[full_iri]
-        except KeyError:
-            self[key] = []
-            ld_value = self.data_dict[full_iri]
-        return self._to_python(full_iri, ld_value)
+        return self._to_python(full_iri, self.data_dict[full_iri])
 
     def __setitem__(self, key, value):
         ld_value = self._to_expanded_json({key: value})
@@ -41,12 +34,7 @@ def __delitem__(self, key):
 
     def __contains__(self, key):
         full_iri = self.ld_proc.expand_iri(self.active_ctx, key)
-        if full_iri == "@id":
-            return "@id" in self.data_dict
-        try:
-            return len(self[full_iri]) != 0
-        except KeyError:
-            return False
+        return full_iri in self.data_dict
 
     def __eq__(self, other):
         if not isinstance(other, (dict, ld_dict)):
@@ -89,6 +77,15 @@ def get(self, key, default=_NO_DEFAULT):
             return default
         return self[key]
 
+    def setdefault(self, key, default):
+        if key not in self:
+            self[key] = default
+        return self[key]
+
+    def emplace(self, key):
+        if key not in self:
+            self[key] = []
+
     def update(self, other):
         for key, value in other.items():
             self[key] = value
@@ -136,7 +133,7 @@ def from_dict(cls, value, *, parent=None, key=None, context=None, ld_type=None):
             full_context = parent.full_context + merged_contexts
 
         ld_value = cls.ld_proc.expand(ld_data, {"expandContext": full_context, "documentLoader": bundled_loader})
-        ld_value = cls(ld_value, parent=parent, key=key, context=merged_contexts)
+        ld_value = ld_dict(ld_value, parent=parent, key=key, context=merged_contexts)
 
         return ld_value
 

test/hermes_test/model/test_api.py

@@ -53,16 +53,18 @@ def test_init_nested_object():
 
 def test_append():
     data = SoftwareMetadata()
+    data.emplace("schema:name")
     data["schema:name"].append("a")
     assert type(data["schema:name"]) is ld_list
     assert data["schema:name"][0] == "a" and data["schema:name"].item_list == [{"@value": "a"}]
     data["schema:name"].append("b")
     assert type(data["schema:name"]) is ld_list and data["schema:name"].item_list == [{"@value": "a"}, {"@value": "b"}]
+    data.emplace("schema:name")
     data["schema:name"].append("c")
     assert data["schema:name"].item_list == [{"@value": "a"}, {"@value": "b"}, {"@value": "c"}]
 
     data = SoftwareMetadata()
-    data["schema:Person"].append({"schema:name": "foo"})
+    data.setdefault("schema:Person", []).append({"schema:name": "foo"})
     assert type(data["schema:Person"]) is ld_list and type(data["schema:Person"][0]) is ld_dict
     assert data["schema:Person"][0].data_dict == {"http://schema.org/name": [{"@value": "foo"}]}
     data["schema:Person"].append({"schema:name": "foo"})
@@ -94,7 +96,7 @@ def test_usage():
     data["author"][0]["email"].append("foo@baz.com")
     assert len(data["author"]) == 2
     assert len(data["author"][0]["email"]) == 2
-    assert len(data["author"][1]["email"]) == 0
+    assert len(data["author"][1].get("email", [])) == 0
     harvest = {
         "authors": [
             {"name": "Foo", "affiliation": ["Uni A", "Lab B"], "kw": ["a", "b", "c"]},
@@ -103,17 +105,19 @@ def test_usage():
         ]
     }
     for author in harvest["authors"]:
-        for exist_author in data["author"]:
-            if author["name"] == exist_author["name"][0]:
+        for exist_author in data.get("author", []):
+            if author["name"] in exist_author.get("name", []):
                 exist_author["affiliation"] = author["affiliation"]
                 if "email" in author:
+                    exist_author.emplace("email")
                     exist_author["email"].append(author["email"])
                 if "kw" in author:
+                    exist_author.emplace("schema:knowsAbout")
                     exist_author["schema:knowsAbout"].extend(author["kw"])
                 break
         else:
-            data["author"].append(author)
-    assert len(data["author"]) == 3
+            data.setdefault("author", []).append(author)
+    assert len(data.get("author", [])) == 3
     foo, bar, baz = data["author"]
     assert foo["name"][0] == "Foo"
     assert foo["affiliation"].to_python() == ["Uni A", "Lab B"]
@@ -124,8 +128,8 @@ def test_usage():
     assert bar["email"].to_python() == ["bar@c.edu"]
     assert baz["name"][0] == "Baz"
     assert baz["affiliation"].to_python() == ["Lab E"]
-    assert len(baz["schema:knowsAbout"]) == 0
-    assert len(baz["email"]) == 0
+    assert len(baz.get("schema:knowsAbout", [])) == 0
+    assert len(baz.get("email", [])) == 0
     for author in data["author"]:
         assert "name" in author
         if "Baz" not in author["name"]: