Generic fetcher: Add lockfile

This commit adds custom lockfile for the generic feature, used in
order to specify list of files to download and checksums to verify
against.

Signed-off-by: Jan Koscielniak <[email protected]>

cachi2/core/package_managers/generic.py

@@ -1,13 +1,54 @@
-from cachi2.core.errors import PackageRejected
+import logging
+from typing import Literal, Optional
+
+import yaml
+from pydantic import BaseModel, ValidationError, field_validator
+
+from cachi2.core.errors import PackageManagerError, PackageRejected
 from cachi2.core.models.input import Request
 from cachi2.core.models.output import RequestOutput
 from cachi2.core.models.sbom import Component
 from cachi2.core.rooted_path import RootedPath
 
+LOG = logging.getLogger(__name__)
 DEFAULT_LOCKFILE_NAME = "cachi2_generic.yaml"
 DEFAULT_DEPS_DIR = "deps/generic"
 
 
+class LockfileMetadata(BaseModel):
+    """Defines format of the metadata section in the lockfile."""
+
+    version: Literal["1.0"]
+
+
+class LockfileArtifact(BaseModel):
+    """Defines format of a single artifact in the lockfile."""
+
+    download_url: str
+    target: Optional[str] = None
+    checksums: dict[str, str]
+
+    @field_validator("checksums")
+    @classmethod
+    def validate_checksums(cls, value: dict[str, str]) -> dict[str, str]:
+        """
+        Validate that at least one checksum is present for an artifact.
+
+        :param value: the checksums dict to validate
+        :return: the validated checksum dict
+        """
+        if len(value) == 0:
+            raise ValueError("At least one checksum must be provided.")
+        return value
+
+
+class GenericLockfileV1(BaseModel):
+    """Defines format of the cachi2 generic lockfile, version 1.0."""
+
+    metadata: LockfileMetadata
+    artifacts: list[LockfileArtifact]
+
+
 def fetch_generic_source(request: Request) -> RequestOutput:
     """
     Resolve and fetch generic dependencies for a given request.
@@ -22,12 +63,53 @@ def fetch_generic_source(request: Request) -> RequestOutput:
 
 
 def _resolve_generic_lockfile(source_dir: RootedPath, output_dir: RootedPath) -> list[Component]:
-    if not source_dir.join_within_root(DEFAULT_LOCKFILE_NAME).path.exists():
+    """
+    Resolve the generic lockfile and pre-fetch the dependencies.
+
+    :param source_dir: the source directory to resolve the lockfile from
+    :param output_dir: the output directory to store the dependencies
+    """
+    lockfile_path = source_dir.join_within_root(DEFAULT_LOCKFILE_NAME)
+    if not lockfile_path.path.exists():
         raise PackageRejected(
             f"Cachi2 generic lockfile '{DEFAULT_LOCKFILE_NAME}' missing, refusing to continue.",
             solution=(
                 f"Make sure your repository has cachi2 generic lockfile '{DEFAULT_LOCKFILE_NAME}' checked in "
                 "to the repository."
             ),
         )
+
+    LOG.info(f"Reading generic lockfile: {lockfile_path}")
+    lockfile = _load_lockfile(lockfile_path)
+    for artifact in lockfile.artifacts:
+        LOG.info(f"Resolving artifact: {artifact.download_url}")
     return []
+
+
+def _load_lockfile(lockfile_path: RootedPath) -> GenericLockfileV1:
+    """
+    Load the cachi2 generic lockfile from the given path.
+
+    :param lockfile_path: the path to the lockfile
+    """
+    with open(lockfile_path, "r") as f:
+        try:
+            lockfile_data = yaml.safe_load(f)
+        except yaml.YAMLError as e:
+            raise PackageRejected(
+                f"Cachi2 lockfile '{lockfile_path}' yaml format is not correct: {e}",
+                solution="Check correct 'yaml' syntax in the lockfile.",
+            )
+
+        try:
+            lockfile = GenericLockfileV1.model_validate(lockfile_data, context={})
+        except ValidationError as e:
+            loc = e.errors()[0]["loc"]
+            msg = e.errors()[0]["msg"]
+            raise PackageManagerError(
+                f"Cachi2 lockfile '{lockfile_path}' format is not valid: '{loc}: {msg}'",
+                solution=(
+                    "Check the correct format and whether any keys are missing in the lockfile."
+                ),
+            )
+    return lockfile

tests/unit/package_managers/test_generic.py

@@ -1,17 +1,53 @@
+from typing import Type
 from unittest import mock
 
 import pytest
 
-from cachi2.core.errors import PackageRejected
+from cachi2.core.errors import Cachi2Error, PackageManagerError, PackageRejected
 from cachi2.core.models.input import GenericPackageInput
 from cachi2.core.models.sbom import Component
 from cachi2.core.package_managers.generic import (
     DEFAULT_LOCKFILE_NAME,
+    GenericLockfileV1,
+    _load_lockfile,
     _resolve_generic_lockfile,
     fetch_generic_source,
 )
 from cachi2.core.rooted_path import RootedPath
 
+LOCKFILE_WRONG_VERSION = """
+metadata:
+    version: '0.42'
+artifacts:
+    - download_url: https://redhat.com/artifact
+      checksums:
+        md5: 3a18656e1cea70504b905836dee14db0
+"""
+
+LOCKFILE_NO_CHECKSUM_1 = """
+metadata:
+    version: '1.0'
+artifacts:
+    - download_url: https://redhat.com/artifact
+"""
+
+LOCKFILE_NO_CHECKSUM_2 = """
+metadata:
+    version: '1.0'
+artifacts:
+    - download_url: https://redhat.com/artifact
+      checksums: {}
+"""
+
+LOCKFILE_NORMAL = """
+metadata:
+    version: '1.0'
+artifacts:
+    - download_url: https://redhat.com/artifact
+      checksums:
+        md5: 3a18656e1cea70504b905836dee14db0
+"""
+
 
 @pytest.mark.parametrize(
     ["model_input", "components"],
@@ -39,10 +75,79 @@ def test_fetch_generic_source(
     mock_from_obj_list.assert_called_with(components=components)
 
 
-def test_resolve_generic_no_lockfile(rooted_tmp_path: RootedPath) -> None:
+@mock.patch("cachi2.core.package_managers.generic._load_lockfile")
+def test_resolve_generic_no_lockfile(mock_load: mock.Mock, rooted_tmp_path: RootedPath) -> None:
     with pytest.raises(PackageRejected) as exc_info:
         _resolve_generic_lockfile(rooted_tmp_path, rooted_tmp_path)
     assert (
         f"Cachi2 generic lockfile '{DEFAULT_LOCKFILE_NAME}' missing, refusing to continue"
         in str(exc_info.value)
     )
+    mock_load.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    ["lockfile", "expected_exception", "expected_err"],
+    [
+        pytest.param("{", PackageRejected, "yaml format is not correct", id="invalid_yaml"),
+        pytest.param(
+            LOCKFILE_WRONG_VERSION, PackageManagerError, "Input should be '1.0'", id="wrong_version"
+        ),
+        pytest.param(
+            LOCKFILE_NO_CHECKSUM_1, PackageManagerError, "Field required", id="no_checksum_1"
+        ),
+        pytest.param(
+            LOCKFILE_NO_CHECKSUM_2,
+            PackageManagerError,
+            "At least one checksum must be provided",
+            id="no_checksum_2",
+        ),
+    ],
+)
+def test_resolve_generic_lockfile_invalid(
+    lockfile: str,
+    expected_exception: Type[Cachi2Error],
+    expected_err: str,
+    rooted_tmp_path: RootedPath,
+) -> None:
+    # setup lockfile
+    with open(rooted_tmp_path.join_within_root(DEFAULT_LOCKFILE_NAME), "w") as f:
+        f.write(lockfile)
+
+    with pytest.raises(expected_exception) as exc_info:
+        _resolve_generic_lockfile(rooted_tmp_path, rooted_tmp_path)
+
+    assert expected_err in str(exc_info.value)
+
+
+@pytest.mark.parametrize(
+    ["lockfile", "expected_lockfile"],
+    [
+        pytest.param(
+            LOCKFILE_NORMAL,
+            GenericLockfileV1.model_validate(
+                {
+                    "metadata": {"version": "1.0"},
+                    "artifacts": [
+                        {
+                            "download_url": "https://redhat.com/artifact",
+                            "checksums": {"md5": "3a18656e1cea70504b905836dee14db0"},
+                        }
+                    ],
+                }
+            ),
+        ),
+    ],
+)
+def test_resolve_generic_lockfile_valid(
+    lockfile: str,
+    expected_lockfile: GenericLockfileV1,
+    rooted_tmp_path: RootedPath,
+) -> None:
+    # setup lockfile
+    with open(rooted_tmp_path.join_within_root(DEFAULT_LOCKFILE_NAME), "w") as f:
+        f.write(lockfile)
+
+    assert (
+        _load_lockfile(rooted_tmp_path.join_within_root(DEFAULT_LOCKFILE_NAME)) == expected_lockfile
+    )