Adding Yarn v1.x workspaces metadata extractor

cachi2/core/package_managers/yarn_classic/main.py

@@ -6,6 +6,7 @@
 from cachi2.core.models.input import Request
 from cachi2.core.models.output import Component, EnvironmentVariable, RequestOutput
 from cachi2.core.package_managers.yarn.utils import run_yarn_cmd
+from cachi2.core.package_managers.yarn_classic.workspaces import extract_workspace_metadata
 from cachi2.core.rooted_path import RootedPath
 
 log = logging.getLogger(__name__)
@@ -27,6 +28,11 @@ def _ensure_mirror_dir_exists(output_dir: RootedPath) -> None:
         prefetch_env = _get_prefetch_environment_variables(request.output_dir)
         _verify_corepack_yarn_version(path, prefetch_env)
         _fetch_dependencies(path, prefetch_env)
+        # Workspaces metadata is not used at the moment, but will
+        # eventualy be converted into components. Using a noop assertion
+        # to prevent linters from complaining.
+        workspaces = extract_workspace_metadata(package, request.source_dir)
+        assert workspaces is not None  # nosec -- see comment above
 
     return RequestOutput.from_obj_list(
         components, _generate_build_environment_variables(), project_files=[]

cachi2/core/package_managers/yarn_classic/workspaces.py

@@ -0,0 +1,106 @@
+import json
+from itertools import chain
+from pathlib import Path
+from typing import Any, Generator, Iterable
+
+import pydantic
+
+from cachi2.core.errors import PackageRejected
+from cachi2.core.models.input import YarnClassicPackageInput
+from cachi2.core.rooted_path import PathOutsideRoot, RootedPath
+
+
+class Workspace(pydantic.BaseModel):
+    """Workspace model."""
+
+    path: Path  # path to a workspace.
+    package_contents: dict  # package data extracted from path/"package.json".
+
+    @pydantic.field_validator("package_contents")
+    def _ensure_package_is_named(cls, package_contents: dict) -> dict:
+        if "name" not in package_contents:
+            raise ValueError("Workspaces must contain 'name' field.")
+        return package_contents
+
+
+def ensure_no_path_leads_out(
+    paths: Iterable[Path],
+    source_dir: RootedPath,
+) -> None:
+    """Ensure no path leads out of source directory.
+
+    Raises an exception when any path is not relative to source directory.
+    Does nothing when path does not exist in the file system.
+    """
+    for path in paths:
+        try:
+            source_dir.join_within_root(path)
+        except PathOutsideRoot:
+            raise PackageRejected(
+                f"Found a workspace path which is not relative to package: {path}",
+                solution=(
+                    "Avoid using packages which try to access your filesystem "
+                    "outside of package directory."
+                ),
+            )
+
+
+def _ensure_workspaces_are_well_formed(
+    paths: Iterable[Path],
+) -> None:
+    """Ensure that every workspace contains package.json.
+
+    Reject the package otherwise.
+    """
+    for p in paths:
+        if not Path(p, "package.json").is_file():
+            raise PackageRejected(
+                reason=f"Workspace {p} does not contain 'package.json'",
+                solution=None,
+            )
+
+
+def _get_workspace_paths(
+    workspaces_globs: list[str],
+    source_dir: RootedPath,
+) -> Iterable[Path]:
+    """Resolve globs within source directory."""
+
+    def all_paths_matching(glob: str) -> Generator[Path, None, None]:
+        return (pth.resolve() for pth in source_dir.path.glob(glob))
+
+    return chain.from_iterable(map(all_paths_matching, workspaces_globs))
+
+
+def _extract_workspaces_globs(
+    package: dict[str, Any],
+) -> list[str]:
+    """Extract globs from workspaces entry in package dict."""
+    workspaces_globs = package.get("workspaces", [])
+    return workspaces_globs
+
+
+def _read_package_from(path: RootedPath) -> dict[str, Any]:
+    """Read package.json from a path."""
+    return json.loads(path.join_within_root("package.json").path.read_text())
+
+
+def extract_workspace_metadata(
+    package: YarnClassicPackageInput,
+    source_dir: RootedPath,
+) -> list[Workspace]:
+    """Extract workspace metadata from a package."""
+    processed_package = _read_package_from(source_dir.join_within_root(package.path))
+    workspaces_globs = _extract_workspaces_globs(processed_package)
+    workspaces_paths = _get_workspace_paths(workspaces_globs, source_dir)
+    ensure_no_path_leads_out(workspaces_paths, source_dir)
+    _ensure_workspaces_are_well_formed(workspaces_paths)
+    parsed_workspaces = []
+    for wp in workspaces_paths:
+        parsed_workspaces.append(
+            Workspace(
+                path=wp,
+                package_contents=_read_package_from(source_dir.join_within_root(wp)),
+            )
+        )
+    return parsed_workspaces

tests/unit/package_managers/yarn_classic/test_main.py

@@ -53,7 +53,9 @@ def test_generate_build_environment_variables(
 @mock.patch("cachi2.core.package_managers.yarn_classic.main._verify_corepack_yarn_version")
 @mock.patch("cachi2.core.package_managers.yarn_classic.main._get_prefetch_environment_variables")
 @mock.patch("cachi2.core.package_managers.yarn_classic.main._fetch_dependencies")
+@mock.patch("cachi2.core.package_managers.yarn_classic.main.extract_workspace_metadata")
 def test_fetch_yarn_source(
+    mock_extract_metadata: mock.Mock,
     mock_fetch_dependencies: mock.Mock,
     mock_prefetch_env_vars: mock.Mock,
     mock_verify_yarn_version: mock.Mock,

tests/unit/package_managers/yarn_classic/test_workspaces.py

@@ -0,0 +1,74 @@
+from pathlib import Path
+from unittest import mock
+
+import pytest
+
+from cachi2.core.errors import PackageRejected
+from cachi2.core.models.input import YarnClassicPackageInput
+from cachi2.core.package_managers.yarn_classic.workspaces import (
+    Workspace,
+    _extract_workspaces_globs,
+    _get_workspace_paths,
+    extract_workspace_metadata,
+)
+from cachi2.core.rooted_path import RootedPath
+
+
+@mock.patch("cachi2.core.package_managers.yarn_classic.workspaces._read_package_from")
+@mock.patch("cachi2.core.package_managers.yarn_classic.workspaces._get_workspace_paths")
+def test_packages_with_workspaces_outside_source_dir_are_rejected(
+    mock_get_ws_paths: mock.Mock,
+    mock_read_package_from: mock.Mock,
+) -> None:
+    package = YarnClassicPackageInput(type="yarn-classic", path=".")
+    mock_read_package_from.return_value = {"workspaces": ["../../usr"]}
+    mock_get_ws_paths.return_value = [Path("/tmp/foo/bar"), Path("/usr")]
+    source_dir = RootedPath("/tmp/foo")
+
+    with pytest.raises(PackageRejected):
+        extract_workspace_metadata(package, source_dir=source_dir)
+
+
+@mock.patch("cachi2.core.package_managers.yarn_classic.workspaces._read_package_from")
+@mock.patch("cachi2.core.package_managers.yarn_classic.workspaces._get_workspace_paths")
+@mock.patch(
+    "cachi2.core.package_managers.yarn_classic.workspaces._ensure_workspaces_are_well_formed"
+)
+def test_workspaces_could_be_parsed(
+    mock_workspaces_ok: mock.Mock,
+    mock_get_ws_paths: mock.Mock,
+    mock_read_package_from: mock.Mock,
+) -> None:
+    package = YarnClassicPackageInput(type="yarn-classic", path=".")
+    mock_read_package_from.side_effect = [{"workspaces": ["quux"]}, {"name": "inner_package"}]
+    mock_get_ws_paths.return_value = [Path("/tmp/foo/bar")]
+    source_dir = RootedPath("/tmp/foo")
+
+    expected_result = [
+        Workspace(
+            path="/tmp/foo/bar",
+            package=YarnClassicPackageInput(type="yarn-classic", path=Path("bar")),
+            package_contents={"name": "inner_package"},
+        ),
+    ]
+    result = extract_workspace_metadata(package, source_dir=source_dir)
+
+    assert result == expected_result
+
+
+def test_extracting_workspace_globs_works_with_globs_deined_in_list() -> None:
+    package = {"workspaces": ["foo"]}
+
+    expected = ["foo"]
+    result = _extract_workspaces_globs(package)
+
+    assert expected == result
+
+
+def test_workspace_paths_could_be_resolved(rooted_tmp_path: RootedPath) -> None:
+    expected = rooted_tmp_path.path / "foo"
+    expected.mkdir()
+
+    result = list(_get_workspace_paths(["foo"], rooted_tmp_path))
+
+    assert result == [expected]