added first e2e harvest test for SoftwareMetadata

Author: Michael Fritzsche

src/hermes/commands/__init__.py

@@ -8,12 +8,12 @@
 # "unused import" errors.
 # flake8: noqa
 
-from hermes.commands.base import HermesHelpCommand
-from hermes.commands.base import HermesVersionCommand
-from hermes.commands.clean.base import HermesCleanCommand
-from hermes.commands.init.base import HermesInitCommand
-from hermes.commands.curate.base import HermesCurateCommand
+# from hermes.commands.base import HermesHelpCommand
+# from hermes.commands.base import HermesVersionCommand
+# from hermes.commands.clean.base import HermesCleanCommand
+# from hermes.commands.init.base import HermesInitCommand
+# from hermes.commands.curate.base import HermesCurateCommand
 from hermes.commands.harvest.base import HermesHarvestCommand
-from hermes.commands.process.base import HermesProcessCommand
-from hermes.commands.deposit.base import HermesDepositCommand
-from hermes.commands.postprocess.base import HermesPostprocessCommand
+# from hermes.commands.process.base import HermesProcessCommand
+# from hermes.commands.deposit.base import HermesDepositCommand
+# from hermes.commands.postprocess.base import HermesPostprocessCommand

src/hermes/commands/base.py

@@ -9,19 +9,20 @@
 import logging
 import pathlib
 from importlib import metadata
-from typing import Dict, Optional, Type
+from typing import Type, Union
 
 import toml
 from pydantic import BaseModel
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
-class _HermesSettings(BaseSettings):
+
+class HermesSettings(BaseSettings):
     """Root class for HERMES configuration model."""
 
     model_config = SettingsConfigDict(env_file_encoding='utf-8')
 
-    logging: Dict = {}
+    logging: dict = {}
 
 
 class HermesCommand(abc.ABC):
@@ -31,7 +32,7 @@ class HermesCommand(abc.ABC):
     """
 
     command_name: str = ""
-    settings_class: Type = _HermesSettings
+    settings_class: Type = HermesSettings
 
     def __init__(self, parser: argparse.ArgumentParser):
         """Initialize a new instance of any HERMES command.
@@ -45,28 +46,27 @@ def __init__(self, parser: argparse.ArgumentParser):
         self.log = logging.getLogger(f"hermes.{self.command_name}")
         self.errors = []
 
-    @classmethod
-    def init_plugins(cls):
+    def init_plugins(self):
         """Collect and initialize the plugins available for the HERMES command."""
 
         # Collect all entry points for this group (i.e., all valid plug-ins for the step)
-        entry_point_group = f"hermes.{cls.command_name}"
-        group_plugins = {
-            entry_point.name: entry_point.load()
-            for entry_point in metadata.entry_points(group=entry_point_group)
-        }
-
-        # Collect the plug-in specific configurations
-        cls.derive_settings_class({
-            plugin_name: plugin_class.settings_class
-            for plugin_name, plugin_class in group_plugins.items()
-            if hasattr(plugin_class, "settings_class") and plugin_class.settings_class is not None
-        })
+        entry_point_group = f"hermes.{self.command_name}"
+        group_plugins = {}
+        group_settings = {}
+
+        for entry_point in metadata.entry_points(group=entry_point_group):
+            plugin_cls = entry_point.load()
+
+            group_plugins[entry_point.name] = plugin_cls
+            if hasattr(plugin_cls, 'settings_class') and plugin_cls.settings_class is not None:
+                group_settings[entry_point.name] = plugin_cls.settings_class
+
+        self.derive_settings_class(group_settings)
 
         return group_plugins
 
     @classmethod
-    def derive_settings_class(cls, setting_types: Dict[str, Type]) -> None:
+    def derive_settings_class(cls, setting_types: dict[str, Type]) -> None:
         """Build a new Pydantic data model class for configuration.
 
         This will create a new class that includes all settings from the plugins available.
@@ -131,13 +131,10 @@ def init_command_parser(self, command_parser: argparse.ArgumentParser) -> None:
 
     def load_settings(self, args: argparse.Namespace):
         """Load settings from the configuration file (passed in from command line)."""
-        try:
-            toml_data = toml.load(args.path / args.config)
-            self.root_settings = HermesCommand.settings_class.model_validate(toml_data)
-            self.settings = getattr(self.root_settings, self.command_name)
-        except FileNotFoundError as e:
-            self.log.error("hermes.toml was not found. Try to run 'hermes init' first or create one manually.")
-            raise e  # This will lead to our default error message & sys.exit
+
+        toml_data = toml.load(args.path / args.config)
+        self.root_settings = HermesCommand.settings_class.model_validate(toml_data)
+        self.settings = getattr(self.root_settings, self.command_name)
 
     def patch_settings(self, args: argparse.Namespace):
         """Process command line options for the settings."""
@@ -164,7 +161,9 @@ def __call__(self, args: argparse.Namespace):
 class HermesPlugin(abc.ABC):
     """Base class for all HERMES plugins."""
 
-    settings_class: Optional[Type] = None
+    pluing_node = None
+
+    settings_class: Union[Type, None] = None
 
     @abc.abstractmethod
     def __call__(self, command: HermesCommand) -> None:
@@ -202,27 +201,3 @@ def __call__(self, args: argparse.Namespace) -> None:
             # Otherwise, simply show the general help and exit (cleanly).
             self.parser.print_help()
             self.parser.exit()
-
-    def load_settings(self, args: argparse.Namespace):
-        """No settings are needed for the help command."""
-        pass
-
-
-class HermesVersionSettings(BaseModel):
-    """Intentionally empty settings class for the version command."""
-    pass
-
-
-class HermesVersionCommand(HermesCommand):
-    """Show HERMES version and exit."""
-
-    command_name = "version"
-    settings_class = HermesVersionSettings
-
-    def load_settings(self, args: argparse.Namespace):
-        """Pass loading settings as not necessary for this command."""
-        pass
-
-    def __call__(self, args: argparse.Namespace) -> None:
-        self.log.info(metadata.version("hermes"))
-        self.parser.exit()

src/hermes/commands/harvest/base.py

@@ -5,14 +5,13 @@
 # SPDX-FileContributor: Michael Meinel
 
 import argparse
-import typing as t
-from datetime import datetime
 
 from pydantic import BaseModel
 
 from hermes.commands.base import HermesCommand, HermesPlugin
-from hermes.model.context import HermesContext, HermesHarvestContext
-from hermes.model.error import HermesValidationError, HermesMergeError
+from hermes.model.context_manager import HermesContext
+from hermes.model.error import HermesValidationError
+from hermes.model import SoftwareMetadata
 
 
 class HermesHarvestPlugin(HermesPlugin):
@@ -21,11 +20,11 @@ class HermesHarvestPlugin(HermesPlugin):
     TODO: describe the harvesting process and how this is mapped to this plugin.
     """
 
-    def __call__(self, command: HermesCommand) -> t.Tuple[t.Dict, t.Dict]:
+    def __call__(self, command: HermesCommand) -> tuple[SoftwareMetadata, dict]:
         pass
 
 
-class _HarvestSettings(BaseModel):
+class HarvestSettings(BaseModel):
     """Generic harvesting settings."""
 
     sources: list[str] = []
@@ -35,32 +34,31 @@ class HermesHarvestCommand(HermesCommand):
     """ Harvest metadata from configured sources. """
 
     command_name = "harvest"
-    settings_class = _HarvestSettings
+    settings_class = HarvestSettings
 
     def __call__(self, args: argparse.Namespace) -> None:
         self.args = args
-        ctx = HermesContext()
 
         # Initialize the harvest cache directory here to indicate the step ran
-        ctx.init_cache("harvest")
+        ctx = HermesContext()
+        ctx.prepare_step('harvest')
 
         for plugin_name in self.settings.sources:
+            plugin_cls = self.plugins[plugin_name]
+
             try:
-                plugin_func = self.plugins[plugin_name]()
-                harvested_data, tags = plugin_func(self)
-
-                with HermesHarvestContext(ctx, plugin_name) as harvest_ctx:
-                    harvest_ctx.update_from(harvested_data,
-                                            plugin=plugin_name,
-                                            timestamp=datetime.now().isoformat(), **tags)
-                    for _key, ((_value, _tag), *_trace) in harvest_ctx._data.items():
-                        if any(v != _value and t == _tag for v, t in _trace):
-                            raise HermesMergeError(_key, None, _value)
-
-            except KeyError as e:
-                self.log.error("Plugin '%s' not found.", plugin_name)
-                self.errors.append(e)
+                # Load plugin and run the harvester
+                plugin_func = plugin_cls()
+                harvested_data = plugin_func(self)
+
+                with ctx[plugin_name] as plugin_ctx:
+                    plugin_ctx["codemeta"] = harvested_data.compact()
+                    plugin_ctx["context"] = {"@context": harvested_data.full_context}
+
+                    plugin_ctx["expanded"] = harvested_data.ld_value
 
             except HermesValidationError as e:
                 self.log.error("Error while executing %s: %s", plugin_name, e)
                 self.errors.append(e)
+
+        ctx.finalize_step('harvest')

src/hermes/commands/harvest/cff.py

@@ -9,16 +9,16 @@
 import logging
 import pathlib
 import urllib.request
-import typing as t
 
 from pydantic import BaseModel
 from ruamel.yaml import YAML
 import jsonschema
 from cffconvert import Citation
+from typing import Any, Union
 
-from hermes.model.context import ContextPath
-from hermes.model.errors import HermesValidationError
+from hermes.model.error import HermesValidationError
 from hermes.commands.harvest.base import HermesHarvestPlugin, HermesHarvestCommand
+from hermes.model import SoftwareMetadata
 
 
 # TODO: should this be configurable via a CLI option?
@@ -35,7 +35,7 @@ class CffHarvestSettings(BaseModel):
 class CffHarvestPlugin(HermesHarvestPlugin):
     settings_class = CffHarvestSettings
 
-    def __call__(self, command: HermesHarvestCommand) -> t.Tuple[t.Dict, t.Dict]:
+    def __call__(self, command: HermesHarvestCommand) -> tuple[SoftwareMetadata, dict]:
         # Get source files
         cff_file = self._get_single_cff(command.args.path)
         if not cff_file:
@@ -44,23 +44,24 @@ def __call__(self, command: HermesHarvestCommand) -> t.Tuple[t.Dict, t.Dict]:
 
         # Read the content
         cff_data = cff_file.read_text()
-
-        # Validate the content to be correct CFF
         cff_dict = self._load_cff_from_file(cff_data)
 
-        if command.settings.cff.enable_validation and not self._validate(cff_file, cff_dict):
-            raise HermesValidationError(cff_file)
+        if command.settings.cff.enable_validation:
+            # Validate the content to be correct CFF
+            if not self._validate(cff_file, cff_dict):
+                raise HermesValidationError(cff_file)
 
         # Convert to CodeMeta using cffconvert
         codemeta_dict = self._convert_cff_to_codemeta(cff_data)
-        # TODO Replace the following temp patch for #112 once there is a new cffconvert version with cffconvert#309
-        codemeta_dict = self._patch_author_emails(cff_dict, codemeta_dict)
         if "version" in codemeta_dict:
             codemeta_dict["version"] = str(codemeta_dict["version"])   # Convert Version to string
 
-        return codemeta_dict, {'local_path': str(cff_file)}
+        # TODO Replace the following temp patch for #112 once there is a new cffconvert version with cffconvert#309
+        codemeta_dict = self._patch_author_emails(cff_dict, codemeta_dict)
+        ld_codemeta = SoftwareMetadata(codemeta_dict, extra_vocabs={'legalName': {'@id': "http://schema.org/name"}})
+        return ld_codemeta, {}
 
-    def _load_cff_from_file(self, cff_data: str) -> t.Any:
+    def _load_cff_from_file(self, cff_data: str) -> Any:
         yaml = YAML(typ='safe')
         yaml.constructor.yaml_constructors[u'tag:yaml.org,2002:timestamp'] = yaml.constructor.yaml_constructors[
             u'tag:yaml.org,2002:str']
@@ -73,11 +74,11 @@ def _patch_author_emails(self, cff: dict, codemeta: dict) -> dict:
                 codemeta["author"][i]["email"] = author["email"]
         return codemeta
 
-    def _convert_cff_to_codemeta(self, cff_data: str) -> t.Any:
+    def _convert_cff_to_codemeta(self, cff_data: str) -> Any:
         codemeta_str = Citation(cff_data).as_codemeta()
         return json.loads(codemeta_str)
 
-    def _validate(self, cff_file: pathlib.Path, cff_dict: t.Dict) -> bool:
+    def _validate(self, cff_file: pathlib.Path, cff_dict: dict) -> bool:
         audit_log = logging.getLogger('audit.cff')
 
         cff_schema_url = f'https://citation-file-format.github.io/{_CFF_VERSION}/schema.json'
@@ -93,7 +94,7 @@ def _validate(self, cff_file: pathlib.Path, cff_dict: t.Dict) -> bool:
             audit_log.warning('!!! warning "%s is not valid according to <%s>"', cff_file, cff_schema_url)
 
             for error in errors:
-                path = ContextPath.make(error.absolute_path or ['root'])
+                path = error.absolute_path or ['root']
                 audit_log.info('    Invalid input for `%s`.', str(path))
                 audit_log.info('    !!! message "%s"', error.message)
                 audit_log.debug('    !!! value "%s"', error.instance)
@@ -108,7 +109,7 @@ def _validate(self, cff_file: pathlib.Path, cff_dict: t.Dict) -> bool:
             audit_log.info('- Found valid Citation File Format file at: %s', cff_file)
             return True
 
-    def _get_single_cff(self, path: pathlib.Path) -> t.Optional[pathlib.Path]:
+    def _get_single_cff(self, path: pathlib.Path) -> Union[pathlib.Path, None]:
         # Find CFF files in directories and subdirectories
         cff_file = path / 'CITATION.cff'
         if cff_file.exists():

src/hermes/model/__init__.py

@@ -2,4 +2,8 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
+# This is an interface file that only provides a public interface, hence linter is disabled to avoid
+# "unused import" errors.
+# flake8: noqa
+
 from hermes.model.api import SoftwareMetadata

test/hermes_test/model/test_api_e2e.py

@@ -0,0 +1,56 @@
+import pytest
+from hermes.commands.harvest.cff import CffHarvestPlugin, CffHarvestSettings
+from hermes.model import SoftwareMetadata
+
+
+@pytest.mark.parametrize(
+    "cff, res",
+    [
+        (
+            """cff-version: 1.2.0
+title: Temp\nmessage: >-
+  If you use this software, please cite it using the
+  metadata from this file.
+type: software
+authors:
+  - given-names: Max
+    family-names: Mustermann
+    email: max@muster.mann""",
+            SoftwareMetadata({
+                "@type": "SoftwareSourceCode",
+                "schema:author": {
+                    "@list": [{
+                        "@type": "Person",
+                        "email": ["max@muster.mann"],
+                        "familyName": ["Mustermann"],
+                        "givenName": ["Max"]
+                    }]
+                },
+                "schema:name": ["Temp"]
+            })
+        )
+    ]
+)
+def test_cff_harvest(tmp_path, cff, res):
+    class Args:
+        def __init__(self, path):
+            self.path = path
+
+    class Settings:
+        def __init__(self, cff_settings):
+            self.cff = cff_settings
+
+    class Command:
+        def __init__(self, args, settings):
+            self.args = args
+            self.settings = settings
+
+    command = Command(Args(tmp_path), Settings(CffHarvestSettings()))
+
+    cff_file = tmp_path / "CITATION.cff"
+    cff_file.write_text(cff)
+
+    result = CffHarvestPlugin().__call__(command)
+    # FIXME: update to compare the SoftwareMetadata objects instead of the data_dicts
+    # after merge with refactor/data-model and/or refactor/423-implement-public-api
+    assert result[0].data_dict == res.data_dict