Adding Yarn v1.x workspaces metadata extractor

In Yarn v1.x workspaces metadata is stored outside of yarn.lock in
package.json. This commit inroduces functions for workspaces metadata
extraction and a dataclass to represent workspaces.

Signed-off-by: Alexey Ovchinnikov <[email protected]>

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__)
@@ -22,11 +23,17 @@ def _ensure_mirror_dir_exists(output_dir: RootedPath) -> None:
         output_dir.join_within_root(MIRROR_DIR).path.mkdir(parents=True, exist_ok=True)
 
     for package in request.yarn_classic_packages:
+        print("::: ", package)
         path = request.source_dir.join_within_root(package.path)
         _ensure_mirror_dir_exists(request.output_dir)
         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,98 @@
+import json
+from contextlib import suppress
+from itertools import chain
+from pathlib import Path
+from typing import Any, Iterable
+
+import pydantic
+
+from cachi2.core.errors import PackageRejected
+from cachi2.core.models.input import YarnClassicPackageInput
+from cachi2.core.rooted_path import RootedPath
+
+
+class Workspace(pydantic.BaseModel):
+    """Workspace model."""
+
+    path: Path  # path to a workspace.
+    package_contents: dict  # package data extracted from path/"package.json".
+    # package reference for potential nested workspace extraction:
+    package: YarnClassicPackageInput
+
+
+def ensure_no_path_leads_out(
+    paths: Iterable[Path],
+    source_dir: RootedPath,
+) -> None:
+    """Ensure no path leads out of source directpry.
+
+    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:
+        if not path.is_relative_to(source_dir.path):
+            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 get_workspace_paths(
+    workspaces_globs: list[str],
+    source_dir: RootedPath,
+) -> Iterable[Path]:
+    """Resolve globs within source directory."""
+
+    def all_paths_matching(glob: str) -> list[Path]:
+        return [pth.resolve() for pth in source_dir.path.glob(glob)]
+
+    return chain.from_iterable(all_paths_matching(g) for g in 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", [])
+    # This couls be a list or a list in a dictionary. If it is not a dictionary
+    # then it is already a list that we need:
+    with suppress(AttributeError):
+        workspaces_globs = workspaces_globs.get("packages", [])
+    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.
+
+    Currently does not deal with nested workspaces, however the way the code
+    is structured it woould be trivial to make component geneartion recursive.
+    It is left non-recursive until it is clear that nested workspaces appear in
+    the wild.
+    """
+    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)
+    parsed_workspaces = []
+    for wp in workspaces_paths:
+        parsed_workspaces.append(
+            Workspace(
+                path=wp,
+                package=YarnClassicPackageInput(
+                    type="yarn-classic", path=wp.relative_to(source_dir.path)
+                ),
+                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,79 @@
+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_workspace_metadata,
+    extract_workspaces_globs,
+    get_workspace_paths,
+)
+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")
+def test_workspaces_could_be_parsed(
+    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_extracting_workspace_globs_works_with_glons_defined_in_dict() -> None:
+    package = {"workspaces": {"packages": ["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]