diff --git a/cycode/cli/files_collector/sca/npm/restore_bun_dependencies.py b/cycode/cli/files_collector/sca/npm/restore_bun_dependencies.py index 2bf0d647..72c65c21 100644 --- a/cycode/cli/files_collector/sca/npm/restore_bun_dependencies.py +++ b/cycode/cli/files_collector/sca/npm/restore_bun_dependencies.py @@ -6,15 +6,19 @@ import typer from cycode.cli.files_collector.sca.base_restore_dependencies import BaseRestoreDependencies, build_dep_tree_path +from cycode.cli.files_collector.sca.npm.workspace import ( + BUN_LOCK_FILE_NAME, + MANIFEST_FILE_NAME, + is_covered_workspace_member, +) from cycode.cli.models import Document -from cycode.cli.utils.path_utils import get_file_content +from cycode.cli.utils.path_utils import get_file_content, get_scan_roots_from_context from cycode.cli.utils.shell_executor import shell from cycode.logger import get_logger logger = get_logger('Bun Restore Dependencies') -BUN_MANIFEST_FILE_NAME = 'package.json' -BUN_LOCK_FILE_NAME = 'bun.lock' +BUN_MANIFEST_FILE_NAME = MANIFEST_FILE_NAME # Only Bun >=1.2 produces the text-based `bun.lock` lockfile that we parse. # Older Bun versions emit a binary `bun.lockb`, which is not supported. @@ -61,6 +65,9 @@ def is_project(self, document: Document) -> bool: if manifest_dir and (Path(manifest_dir) / BUN_LOCK_FILE_NAME).is_file(): return True + if is_covered_workspace_member(manifest_dir, document.path, get_scan_roots_from_context(self.ctx)): + return False + return _indicates_bun(document.content) def _is_supported_bun_version(self) -> bool: diff --git a/cycode/cli/files_collector/sca/npm/restore_deno_dependencies.py b/cycode/cli/files_collector/sca/npm/restore_deno_dependencies.py index d3aeb5e5..d81b7a9d 100644 --- a/cycode/cli/files_collector/sca/npm/restore_deno_dependencies.py +++ b/cycode/cli/files_collector/sca/npm/restore_deno_dependencies.py @@ -4,6 +4,7 @@ import typer from cycode.cli.files_collector.sca.base_restore_dependencies import BaseRestoreDependencies, build_dep_tree_path +from cycode.cli.files_collector.sca.npm.workspace import DENO_LOCK_FILE_NAME from cycode.cli.models import Document from cycode.cli.utils.path_utils import get_file_content from cycode.logger import get_logger @@ -11,7 +12,6 @@ logger = get_logger('Deno Restore Dependencies') DENO_MANIFEST_FILE_NAMES = ('deno.json', 'deno.jsonc') -DENO_LOCK_FILE_NAME = 'deno.lock' class RestoreDenoDependencies(BaseRestoreDependencies): diff --git a/cycode/cli/files_collector/sca/npm/restore_npm_dependencies.py b/cycode/cli/files_collector/sca/npm/restore_npm_dependencies.py index 9416f58c..ab15d2d3 100644 --- a/cycode/cli/files_collector/sca/npm/restore_npm_dependencies.py +++ b/cycode/cli/files_collector/sca/npm/restore_npm_dependencies.py @@ -3,15 +3,25 @@ import typer from cycode.cli.files_collector.sca.base_restore_dependencies import BaseRestoreDependencies +from cycode.cli.files_collector.sca.npm.workspace import ( + BUN_LOCK_FILE_NAME, + DENO_LOCK_FILE_NAME, + MANIFEST_FILE_NAME, + NPM_LOCK_FILE_NAME, + NPM_SHRINKWRAP_FILE_NAME, + PNPM_LOCK_FILE_NAME, + YARN_LOCK_FILE_NAME, + is_covered_workspace_member, +) from cycode.cli.models import Document +from cycode.cli.utils.path_utils import get_scan_roots_from_context from cycode.logger import get_logger logger = get_logger('NPM Restore Dependencies') -NPM_MANIFEST_FILE_NAME = 'package.json' -NPM_LOCK_FILE_NAME = 'package-lock.json' +NPM_MANIFEST_FILE_NAME = MANIFEST_FILE_NAME # These lockfiles indicate another package manager owns the project — NPM should not run -_ALTERNATIVE_LOCK_FILES = ('yarn.lock', 'pnpm-lock.yaml', 'deno.lock', 'bun.lock') +_ALTERNATIVE_LOCK_FILES = (YARN_LOCK_FILE_NAME, PNPM_LOCK_FILE_NAME, DENO_LOCK_FILE_NAME, BUN_LOCK_FILE_NAME) class RestoreNpmDependencies(BaseRestoreDependencies): @@ -37,16 +47,18 @@ def is_project(self, document: Document) -> bool: return False manifest_dir = self.get_manifest_dir(document) - if manifest_dir: - for lock_file in _ALTERNATIVE_LOCK_FILES: - if (Path(manifest_dir) / lock_file).is_file(): - logger.debug( - 'Skipping npm restore: alternative lockfile detected, %s', - {'path': document.path, 'lockfile': lock_file}, - ) - return False + if not manifest_dir: + return True - return True + for lock_file in _ALTERNATIVE_LOCK_FILES: + if (Path(manifest_dir) / lock_file).is_file(): + logger.debug( + 'Skipping npm restore: alternative lockfile detected, %s', + {'path': document.path, 'lockfile': lock_file}, + ) + return False + + return not is_covered_workspace_member(manifest_dir, document.path, get_scan_roots_from_context(self.ctx)) def get_commands(self, manifest_file_path: str) -> list[list[str]]: return [ @@ -65,7 +77,11 @@ def get_lock_file_name(self) -> str: return NPM_LOCK_FILE_NAME def get_lock_file_names(self) -> list[str]: - return [NPM_LOCK_FILE_NAME] + return [NPM_LOCK_FILE_NAME, NPM_SHRINKWRAP_FILE_NAME] + + def get_restored_lock_file_name(self, restore_file_path: str) -> str: + name = Path(restore_file_path).name + return name if name in self.get_lock_file_names() else self.get_lock_file_name() @staticmethod def prepare_manifest_file_path_for_command(manifest_file_path: str) -> str: diff --git a/cycode/cli/files_collector/sca/npm/restore_pnpm_dependencies.py b/cycode/cli/files_collector/sca/npm/restore_pnpm_dependencies.py index bce7eff6..159ac506 100644 --- a/cycode/cli/files_collector/sca/npm/restore_pnpm_dependencies.py +++ b/cycode/cli/files_collector/sca/npm/restore_pnpm_dependencies.py @@ -5,14 +5,18 @@ import typer from cycode.cli.files_collector.sca.base_restore_dependencies import BaseRestoreDependencies, build_dep_tree_path +from cycode.cli.files_collector.sca.npm.workspace import ( + MANIFEST_FILE_NAME, + PNPM_LOCK_FILE_NAME, + is_covered_workspace_member, +) from cycode.cli.models import Document -from cycode.cli.utils.path_utils import get_file_content +from cycode.cli.utils.path_utils import get_file_content, get_scan_roots_from_context from cycode.logger import get_logger logger = get_logger('Pnpm Restore Dependencies') -PNPM_MANIFEST_FILE_NAME = 'package.json' -PNPM_LOCK_FILE_NAME = 'pnpm-lock.yaml' +PNPM_MANIFEST_FILE_NAME = MANIFEST_FILE_NAME def _indicates_pnpm(package_json_content: Optional[str]) -> bool: @@ -44,6 +48,9 @@ def is_project(self, document: Document) -> bool: if manifest_dir and (Path(manifest_dir) / PNPM_LOCK_FILE_NAME).is_file(): return True + if is_covered_workspace_member(manifest_dir, document.path, get_scan_roots_from_context(self.ctx)): + return False + return _indicates_pnpm(document.content) def try_restore_dependencies(self, document: Document) -> Optional[Document]: diff --git a/cycode/cli/files_collector/sca/npm/restore_yarn_dependencies.py b/cycode/cli/files_collector/sca/npm/restore_yarn_dependencies.py index 79b0c4ec..9b4cbb11 100644 --- a/cycode/cli/files_collector/sca/npm/restore_yarn_dependencies.py +++ b/cycode/cli/files_collector/sca/npm/restore_yarn_dependencies.py @@ -5,14 +5,18 @@ import typer from cycode.cli.files_collector.sca.base_restore_dependencies import BaseRestoreDependencies, build_dep_tree_path +from cycode.cli.files_collector.sca.npm.workspace import ( + MANIFEST_FILE_NAME, + YARN_LOCK_FILE_NAME, + is_covered_workspace_member, +) from cycode.cli.models import Document -from cycode.cli.utils.path_utils import get_file_content +from cycode.cli.utils.path_utils import get_file_content, get_scan_roots_from_context from cycode.logger import get_logger logger = get_logger('Yarn Restore Dependencies') -YARN_MANIFEST_FILE_NAME = 'package.json' -YARN_LOCK_FILE_NAME = 'yarn.lock' +YARN_MANIFEST_FILE_NAME = MANIFEST_FILE_NAME def _indicates_yarn(package_json_content: Optional[str]) -> bool: @@ -44,6 +48,9 @@ def is_project(self, document: Document) -> bool: if manifest_dir and (Path(manifest_dir) / YARN_LOCK_FILE_NAME).is_file(): return True + if is_covered_workspace_member(manifest_dir, document.path, get_scan_roots_from_context(self.ctx)): + return False + return _indicates_yarn(document.content) def try_restore_dependencies(self, document: Document) -> Optional[Document]: diff --git a/cycode/cli/files_collector/sca/npm/workspace/__init__.py b/cycode/cli/files_collector/sca/npm/workspace/__init__.py new file mode 100644 index 00000000..0e96d2a1 --- /dev/null +++ b/cycode/cli/files_collector/sca/npm/workspace/__init__.py @@ -0,0 +1,59 @@ +"""Deciding whether a lockfile above a manifest already resolves that manifest's dependencies. + +The lockfile is the authority wherever it can name its members; the manifest's workspaces globs +are a fallback for the formats that cannot. See resolvers.py for the per-format readers. +""" + +from cycode.cli.files_collector.sca.npm.workspace import coverage as _coverage +from cycode.cli.files_collector.sca.npm.workspace import globs as _globs +from cycode.cli.files_collector.sca.npm.workspace import resolvers as _resolvers +from cycode.cli.files_collector.sca.npm.workspace.coverage import ( + WorkspaceCoverage, + find_covering_workspace, + is_covered_workspace_member, +) +from cycode.cli.files_collector.sca.npm.workspace.names import ( + BUN_BINARY_LOCK_FILE_NAME, + BUN_LOCK_FILE_NAME, + BUN_PACKAGE_MANAGER, + DENO_LOCK_FILE_NAME, + DENO_PACKAGE_MANAGER, + MANIFEST_FILE_NAME, + NPM_LOCK_FILE_NAME, + NPM_PACKAGE_MANAGER, + NPM_SHRINKWRAP_FILE_NAME, + PNPM_LOCK_FILE_NAME, + PNPM_PACKAGE_MANAGER, + YARN_LOCK_FILE_NAME, + YARN_PACKAGE_MANAGER, +) +from cycode.cli.files_collector.sca.npm.workspace.resolvers import MEMBER_RESOLVERS + + +def clear_cache() -> None: + """Drop every memo, so one scan never inherits another scan's view of the filesystem.""" + _resolvers.clear_cache() + _globs.clear_cache() + _coverage.clear_cache() + + +__all__ = [ + 'BUN_BINARY_LOCK_FILE_NAME', + 'BUN_LOCK_FILE_NAME', + 'BUN_PACKAGE_MANAGER', + 'DENO_LOCK_FILE_NAME', + 'DENO_PACKAGE_MANAGER', + 'MANIFEST_FILE_NAME', + 'MEMBER_RESOLVERS', + 'NPM_LOCK_FILE_NAME', + 'NPM_PACKAGE_MANAGER', + 'NPM_SHRINKWRAP_FILE_NAME', + 'PNPM_LOCK_FILE_NAME', + 'PNPM_PACKAGE_MANAGER', + 'YARN_LOCK_FILE_NAME', + 'YARN_PACKAGE_MANAGER', + 'WorkspaceCoverage', + 'clear_cache', + 'find_covering_workspace', + 'is_covered_workspace_member', +] diff --git a/cycode/cli/files_collector/sca/npm/workspace/coverage.py b/cycode/cli/files_collector/sca/npm/workspace/coverage.py new file mode 100644 index 00000000..7512cb60 --- /dev/null +++ b/cycode/cli/files_collector/sca/npm/workspace/coverage.py @@ -0,0 +1,142 @@ +"""Walking up from a manifest to the lockfile that already resolves it.""" + +import os +from pathlib import Path +from typing import TYPE_CHECKING, NamedTuple, Optional + +from cycode.cli.files_collector.sca.npm.workspace.files import logger, resolved_path +from cycode.cli.files_collector.sca.npm.workspace.globs import declares_workspace_member +from cycode.cli.files_collector.sca.npm.workspace.resolvers import MEMBER_RESOLVERS +from cycode.cli.utils.path_utils import is_sub_path + +if TYPE_CHECKING: + from collections.abc import Iterator + +_GIT_DIR_NAME = '.git' + +_reported_unscanned_roots: set = set() + + +def clear_cache() -> None: + _reported_unscanned_roots.clear() + + +class WorkspaceCoverage(NamedTuple): + package_manager: str + lock_file: Path + + +def _scan_root_directories(scan_roots: tuple) -> list: + """The directory each scanned path stands for; scanning a file scans its directory.""" + directories = [] + for scan_root in scan_roots: + resolved = resolved_path(scan_root) + if os.path.isfile(resolved): + resolved = os.path.dirname(resolved) + + if resolved: + directories.append(resolved) + + return directories + + +def _containing_scan_roots(manifest_dir: Path, scan_roots: tuple) -> list: + resolved_manifest_dir = resolved_path(manifest_dir) + return [ + Path(directory) + for directory in _scan_root_directories(scan_roots) + if is_sub_path(directory, resolved_manifest_dir) + ] + + +def _resolve_walk_boundary(manifest_dir: Path, scan_roots: tuple) -> Optional[Path]: + for root_dir in manifest_dir.parents: + if (root_dir / _GIT_DIR_NAME).exists(): + return root_dir + + containing = _containing_scan_roots(manifest_dir, scan_roots) + if containing: + return min(containing, key=lambda scan_root: len(scan_root.parts)) + + if scan_roots: + return manifest_dir + + return None + + +def _workspace_root_candidates(manifest_dir: Path, scan_roots: tuple) -> 'Iterator[Path]': + boundary = _resolve_walk_boundary(manifest_dir, scan_roots) + resolved_boundary = resolved_path(boundary) if boundary is not None else None + if resolved_boundary == resolved_path(manifest_dir): + return + + for root_dir in manifest_dir.parents: + yield root_dir + if resolved_boundary is not None and resolved_path(root_dir) == resolved_boundary: + return + + +def _find_covering_workspace(manifest_dir: Path, scan_roots: tuple) -> Optional[WorkspaceCoverage]: + for root_dir in _workspace_root_candidates(manifest_dir, scan_roots): + member_path = manifest_dir.relative_to(root_dir).as_posix() + + for resolver in MEMBER_RESOLVERS: + for lock_file_name in resolver.lock_file_names: + lock_file = root_dir / lock_file_name + if not lock_file.is_file(): + continue + + member_names = resolver.resolve(lock_file) + if member_names is not None: + if member_path in member_names: + return WorkspaceCoverage(resolver.package_manager, lock_file) + continue + + if resolver.may_use_workspace_globs and declares_workspace_member(root_dir, member_path): + return WorkspaceCoverage(resolver.package_manager, lock_file) + + return None + + +def find_covering_workspace(manifest_dir: Optional[str], scan_roots: tuple = ()) -> Optional[WorkspaceCoverage]: + if not manifest_dir: + return None + + return _find_covering_workspace(Path(manifest_dir), scan_roots) + + +def _is_inside_scanned_paths(scan_roots: tuple, root_dir: Path) -> bool: + directories = _scan_root_directories(scan_roots) + if not directories: + logger.debug('No scanned paths in context; treating the workspace root as scanned, %s', {'root': str(root_dir)}) + return True + + resolved_root_dir = resolved_path(root_dir) + return any(is_sub_path(directory, resolved_root_dir) for directory in directories) + + +def is_covered_workspace_member(manifest_dir: Optional[str], document_path: str, scan_roots: tuple = ()) -> bool: + coverage = find_covering_workspace(manifest_dir, scan_roots) + if coverage is None: + return False + + details = { + 'path': document_path, + 'root_lockfile': str(coverage.lock_file), + 'workspace': coverage.package_manager, + } + if _is_inside_scanned_paths(scan_roots, coverage.lock_file.parent): + logger.debug('Skipping restore: the workspace root lockfile already covers this member, %s', details) + return True + + report_key = (document_path, str(coverage.lock_file)) + if report_key not in _reported_unscanned_roots: + _reported_unscanned_roots.add(report_key) + logger.warning( + 'The workspace root lockfile is outside the scanned path and will not be collected, ' + 'so this member is restored on its own. Scan the workspace root for the versions it ' + 'actually installs, %s', + details, + ) + + return False diff --git a/cycode/cli/files_collector/sca/npm/workspace/files.py b/cycode/cli/files_collector/sca/npm/workspace/files.py new file mode 100644 index 00000000..e51485da --- /dev/null +++ b/cycode/cli/files_collector/sca/npm/workspace/files.py @@ -0,0 +1,38 @@ +"""Filesystem access shared by the readers: identity for caching, and tolerant parsing.""" + +import json +import os +from pathlib import Path +from typing import Optional + +from cycode.cli.utils.path_utils import get_absolute_path +from cycode.logger import get_logger + +logger = get_logger('SCA NPM Workspace') + +FileStamp = tuple[str, int, int] + + +def resolved_path(path: object) -> str: + return os.path.realpath(get_absolute_path(str(path))) + + +def file_stamp(path: Path) -> Optional[FileStamp]: + try: + stat_result = path.stat() + except OSError: + return None + + return str(path), stat_result.st_mtime_ns, stat_result.st_size + + +def read_json_object(path: Path) -> Optional[dict]: + try: + content = json.loads(path.read_text(encoding='UTF-8')) + except FileNotFoundError: + return None + except (OSError, ValueError) as e: + logger.debug('Could not read an npm workspace file, %s', {'path': str(path), 'error': e}) + return None + + return content if isinstance(content, dict) else None diff --git a/cycode/cli/files_collector/sca/npm/workspace/globs.py b/cycode/cli/files_collector/sca/npm/workspace/globs.py new file mode 100644 index 00000000..65f8e187 --- /dev/null +++ b/cycode/cli/files_collector/sca/npm/workspace/globs.py @@ -0,0 +1,138 @@ +"""Matching a member against the manifest's workspaces globs. + +Only reached for lockfiles that cannot name their own members: classic yarn.lock and the +binary bun.lockb. Every other format answers from the lockfile itself. +""" + +import re +from pathlib import Path +from typing import NamedTuple + +from cycode.cli.files_collector.sca.npm.workspace.files import FileStamp, file_stamp, read_json_object +from cycode.cli.files_collector.sca.npm.workspace.names import MANIFEST_FILE_NAME + +_MANIFEST_WORKSPACES_SECTION = 'workspaces' +_MANIFEST_WORKSPACE_PACKAGES_SECTION = 'packages' +_NEGATION_PREFIX = '!' +_GLOBSTAR_SUFFIX = '/**' +_GLOBSTAR_PREFIX = '**/' + + +class _WorkspacePatterns(NamedTuple): + included: tuple[str, ...] + excluded: tuple[str, ...] + + +_EMPTY_WORKSPACE_PATTERNS = _WorkspacePatterns((), ()) + +_workspace_patterns_cache: dict[FileStamp, _WorkspacePatterns] = {} +_workspace_pattern_regex_cache: dict[str, 're.Pattern[str]'] = {} + + +def clear_cache() -> None: + _workspace_patterns_cache.clear() + _workspace_pattern_regex_cache.clear() + + +def _workspace_pattern_body(pattern: str) -> str: + parts = [] + index = 0 + while index < len(pattern): + character = pattern[index] + if character == '*' and pattern[index + 1 : index + 2] == '*': + parts.append('.*') + index += 2 + elif character == '*': + parts.append('[^/]*') + index += 1 + elif character == '?': + parts.append('[^/]') + index += 1 + else: + parts.append(re.escape(character)) + index += 1 + + return ''.join(parts) + + +def _compile_workspace_pattern(pattern: str) -> 're.Pattern[str]': + """Translate a workspace glob, where ** spans zero or more path segments. + + Workspace members are discovered by globbing /package.json, so src/app/** + matches src/app itself as well as anything beneath it. pnpm records exactly that in its + lockfile importers, and treating the trailing separator as mandatory would miss the member. + """ + compiled = _workspace_pattern_regex_cache.get(pattern) + if compiled is not None: + return compiled + + body = pattern + matches_anything_below = body.endswith(_GLOBSTAR_SUFFIX) + if matches_anything_below: + body = body[: -len(_GLOBSTAR_SUFFIX)] + + matches_anything_above = body.startswith(_GLOBSTAR_PREFIX) + if matches_anything_above: + body = body[len(_GLOBSTAR_PREFIX) :] + + expression = '' + if matches_anything_above: + expression += '(?:.*/)?' + expression += _workspace_pattern_body(body) + if matches_anything_below: + expression += '(?:/.*)?' + + compiled = re.compile(expression) + _workspace_pattern_regex_cache[pattern] = compiled + return compiled + + +def _split_workspace_patterns(declared: list) -> _WorkspacePatterns: + included = [] + excluded = [] + for entry in declared: + stripped = entry.strip() + is_excluded = stripped.startswith(_NEGATION_PREFIX) + normalized = (stripped[1:] if is_excluded else stripped).strip() + if normalized.startswith('./'): + normalized = normalized[2:] + + normalized = normalized.rstrip('/') + if not normalized: + continue + + if is_excluded: + excluded.append(normalized) + else: + included.append(normalized) + + return _WorkspacePatterns(tuple(included), tuple(excluded)) + + +def _read_manifest_workspace_patterns(root_dir: Path) -> _WorkspacePatterns: + manifest = root_dir / MANIFEST_FILE_NAME + stamp = file_stamp(manifest) + if stamp is None: + return _EMPTY_WORKSPACE_PATTERNS + + cached = _workspace_patterns_cache.get(stamp) + if cached is not None: + return cached + + content = read_json_object(manifest) + workspaces = content.get(_MANIFEST_WORKSPACES_SECTION) if content is not None else None + if isinstance(workspaces, dict): + workspaces = workspaces.get(_MANIFEST_WORKSPACE_PACKAGES_SECTION) + + declared = [entry for entry in workspaces if isinstance(entry, str)] if isinstance(workspaces, list) else [] + patterns = _split_workspace_patterns(declared) + _workspace_patterns_cache[stamp] = patterns + return patterns + + +def declares_workspace_member(root_dir: Path, member_path: str) -> bool: + patterns = _read_manifest_workspace_patterns(root_dir) + if not any(_compile_workspace_pattern(pattern).fullmatch(member_path) for pattern in patterns.included): + return False + + return not any(_compile_workspace_pattern(pattern).fullmatch(member_path) for pattern in patterns.excluded) diff --git a/cycode/cli/files_collector/sca/npm/workspace/names.py b/cycode/cli/files_collector/sca/npm/workspace/names.py new file mode 100644 index 00000000..f9cec37f --- /dev/null +++ b/cycode/cli/files_collector/sca/npm/workspace/names.py @@ -0,0 +1,17 @@ +"""File names and package-manager identifiers shared across the workspace package.""" + +MANIFEST_FILE_NAME = 'package.json' + +NPM_PACKAGE_MANAGER = 'npm' +YARN_PACKAGE_MANAGER = 'yarn' +PNPM_PACKAGE_MANAGER = 'pnpm' +BUN_PACKAGE_MANAGER = 'bun' +DENO_PACKAGE_MANAGER = 'deno' + +NPM_LOCK_FILE_NAME = 'package-lock.json' +NPM_SHRINKWRAP_FILE_NAME = 'npm-shrinkwrap.json' +YARN_LOCK_FILE_NAME = 'yarn.lock' +PNPM_LOCK_FILE_NAME = 'pnpm-lock.yaml' +BUN_LOCK_FILE_NAME = 'bun.lock' +BUN_BINARY_LOCK_FILE_NAME = 'bun.lockb' +DENO_LOCK_FILE_NAME = 'deno.lock' diff --git a/cycode/cli/files_collector/sca/npm/workspace/resolvers.py b/cycode/cli/files_collector/sca/npm/workspace/resolvers.py new file mode 100644 index 00000000..c4a186fc --- /dev/null +++ b/cycode/cli/files_collector/sca/npm/workspace/resolvers.py @@ -0,0 +1,251 @@ +"""Reading the members a root lockfile resolves, one reader per lockfile format. + +A reader that returns a set is the authority on that lockfile, including when the set is empty. +None means the format cannot name its members at all, and only then does may_use_workspace_globs +decide whether the manifest globs get a say. +""" + +import re +from abc import ABC, abstractmethod +from pathlib import Path +from typing import Optional + +import yaml + +from cycode.cli.files_collector.sca.npm.workspace.files import FileStamp, file_stamp, logger, read_json_object +from cycode.cli.files_collector.sca.npm.workspace.names import ( + BUN_BINARY_LOCK_FILE_NAME, + BUN_LOCK_FILE_NAME, + BUN_PACKAGE_MANAGER, + DENO_LOCK_FILE_NAME, + DENO_PACKAGE_MANAGER, + NPM_LOCK_FILE_NAME, + NPM_PACKAGE_MANAGER, + NPM_SHRINKWRAP_FILE_NAME, + PNPM_LOCK_FILE_NAME, + PNPM_PACKAGE_MANAGER, + YARN_LOCK_FILE_NAME, + YARN_PACKAGE_MANAGER, +) + +_LOCKFILE_PACKAGES_SECTION = 'packages' +_NODE_MODULES_SEPARATOR = 'node_modules/' +_PNPM_LOCKFILE_IMPORTERS_SECTION = 'importers' +_PNPM_LOCKFILE_ROOT_IMPORTER = '.' +_YAML_COMMENT_PREFIX = '#' +_YARN_BERRY_MARKER = '__metadata' +_YARN_RESOLUTION_PREFIX = 'resolution:' +_YARN_WORKSPACE_PROTOCOL = re.compile(r'@workspace:([^"\',\s]+)') + +_member_names_cache: dict[FileStamp, Optional[frozenset[str]]] = {} + + +def clear_cache() -> None: + _member_names_cache.clear() + + +def _normalize_member_path(member_path: str) -> Optional[str]: + normalized = member_path.strip() + if normalized.startswith('./'): + normalized = normalized[2:] + + normalized = normalized.rstrip('/') + if not normalized or normalized == _PNPM_LOCKFILE_ROOT_IMPORTER: + return None + + return normalized + + +def _npm_lockfile_member_names(lock_file: Path) -> Optional[frozenset]: + content = read_json_object(lock_file) + packages = content.get(_LOCKFILE_PACKAGES_SECTION) if content is not None else None + if not isinstance(packages, dict): + return None + + return frozenset(name for name in packages if name and _NODE_MODULES_SEPARATOR not in name) + + +def _read_pnpm_importers_section(lock_file: Path) -> str: + """Slice out the top-level importers block so a large lockfile is not parsed in full.""" + section = [] + inside = False + try: + with lock_file.open(encoding='UTF-8') as lock_file_lines: + for raw_line in lock_file_lines: + line = raw_line.rstrip('\n').rstrip('\r') + if not inside: + if line.startswith(f'{_PNPM_LOCKFILE_IMPORTERS_SECTION}:'): + inside = True + section.append(line) + continue + + if line.startswith(_YAML_COMMENT_PREFIX): + continue + + if line and not line[0].isspace(): + break + + section.append(line) + except FileNotFoundError: + return '' + except (OSError, ValueError) as e: + logger.debug('Could not read a pnpm lockfile, %s', {'path': str(lock_file), 'error': e}) + return '' + + return '\n'.join(section) + + +def _pnpm_lockfile_member_names(lock_file: Path) -> Optional[frozenset]: + section = _read_pnpm_importers_section(lock_file) + if not section: + return None + + try: + content = yaml.safe_load(section) + except (ValueError, yaml.YAMLError) as e: + logger.debug('Could not read a pnpm lockfile, %s', {'path': str(lock_file), 'error': e}) + return None + + importers = content.get(_PNPM_LOCKFILE_IMPORTERS_SECTION) if isinstance(content, dict) else None + if not isinstance(importers, dict): + return None + + member_names = set() + for name in importers: + if not isinstance(name, str): + continue + + normalized = _normalize_member_path(name) + if normalized: + member_names.add(normalized) + + # a lockfile whose only importer is the root describes a single package, not a workspace + return frozenset(member_names) or None + + +def _yarn_lockfile_member_names(lock_file: Path) -> Optional[frozenset]: + """Yarn berry records every member as "@workspace:"; classic yarn records nothing.""" + try: + text = lock_file.read_text(encoding='UTF-8', errors='replace') + except FileNotFoundError: + return None + except (OSError, ValueError) as e: + logger.debug('Could not read a yarn lockfile, %s', {'path': str(lock_file), 'error': e}) + return None + + if _YARN_BERRY_MARKER not in text: + return None # classic: flat, so the manifest globs are the only remaining source + + member_names = set() + for line in text.splitlines(): + # every entry carries a resolution naming its real path; the dependency entries carry + # ranges instead (workspace:^, workspace:*), which are not paths + if not line.strip().startswith(_YARN_RESOLUTION_PREFIX): + continue + + for declared_path in _YARN_WORKSPACE_PROTOCOL.findall(line): + normalized = _normalize_member_path(declared_path) + if normalized: + member_names.add(normalized) + + # berry always records what it installed, so an empty result means this is not a workspace + return frozenset(member_names) + + +class WorkspaceMemberResolver(ABC): + """Reads, from one root lockfile, the member directories that lockfile resolves.""" + + @property + @abstractmethod + def package_manager(self) -> str: ... + + @property + @abstractmethod + def lock_file_names(self) -> tuple: ... + + @property + def may_use_workspace_globs(self) -> bool: + """Whether an unanswerable lockfile may defer to the manifest's workspaces globs. + + False means the silence is itself an answer: a v1 package-lock predates workspaces, so + its root is simply not one. True means the format has workspaces but does not record + them, leaving the globs as the only remaining source. + """ + return False + + def resolve(self, lock_file: Path) -> Optional[frozenset]: + """Member paths this lockfile resolves, or None when it cannot name them. + + Caches on the file's identity so one scan parses each root lockfile once. + """ + stamp = file_stamp(lock_file) + if stamp is None: + return None + + if stamp in _member_names_cache: + return _member_names_cache[stamp] + + member_names = self._read_member_names(lock_file) + _member_names_cache[stamp] = member_names + return member_names + + @abstractmethod + def _read_member_names(self, lock_file: Path) -> Optional[frozenset]: ... + + +class NpmLockfileResolver(WorkspaceMemberResolver): + package_manager = NPM_PACKAGE_MANAGER + lock_file_names = (NPM_LOCK_FILE_NAME, NPM_SHRINKWRAP_FILE_NAME) + + def _read_member_names(self, lock_file: Path) -> Optional[frozenset]: + return _npm_lockfile_member_names(lock_file) + + +class PnpmLockfileResolver(WorkspaceMemberResolver): + package_manager = PNPM_PACKAGE_MANAGER + lock_file_names = (PNPM_LOCK_FILE_NAME,) + + def _read_member_names(self, lock_file: Path) -> Optional[frozenset]: + return _pnpm_lockfile_member_names(lock_file) + + +class YarnLockfileResolver(WorkspaceMemberResolver): + """Berry names every member; classic is flat and names none, so only classic needs the globs.""" + + package_manager = YARN_PACKAGE_MANAGER + lock_file_names = (YARN_LOCK_FILE_NAME,) + may_use_workspace_globs = True + + def _read_member_names(self, lock_file: Path) -> Optional[frozenset]: + return _yarn_lockfile_member_names(lock_file) + + +class OpaqueLockfileResolver(WorkspaceMemberResolver): + """A lockfile we cannot read members from at all, so the manifest globs decide.""" + + may_use_workspace_globs = True + + def __init__(self, package_manager: str, lock_file_names: tuple) -> None: + self._package_manager = package_manager + self._lock_file_names = lock_file_names + + @property + def package_manager(self) -> str: + return self._package_manager + + @property + def lock_file_names(self) -> tuple: + return self._lock_file_names + + def _read_member_names(self, lock_file: Path) -> Optional[frozenset]: + return None + + +# Order is precedence: the first lockfile that resolves the member wins. +MEMBER_RESOLVERS = ( + NpmLockfileResolver(), + YarnLockfileResolver(), + PnpmLockfileResolver(), + OpaqueLockfileResolver(BUN_PACKAGE_MANAGER, (BUN_LOCK_FILE_NAME, BUN_BINARY_LOCK_FILE_NAME)), + OpaqueLockfileResolver(DENO_PACKAGE_MANAGER, (DENO_LOCK_FILE_NAME,)), +) diff --git a/cycode/cli/files_collector/sca/sca_file_collector.py b/cycode/cli/files_collector/sca/sca_file_collector.py index 4db5cd04..bcc7e095 100644 --- a/cycode/cli/files_collector/sca/sca_file_collector.py +++ b/cycode/cli/files_collector/sca/sca_file_collector.py @@ -15,6 +15,7 @@ from cycode.cli.files_collector.sca.npm.restore_npm_dependencies import RestoreNpmDependencies from cycode.cli.files_collector.sca.npm.restore_pnpm_dependencies import RestorePnpmDependencies from cycode.cli.files_collector.sca.npm.restore_yarn_dependencies import RestoreYarnDependencies +from cycode.cli.files_collector.sca.npm.workspace import clear_cache as clear_npm_workspace_cache from cycode.cli.files_collector.sca.nuget.restore_nuget_dependencies import RestoreNugetDependencies from cycode.cli.files_collector.sca.php.restore_composer_dependencies import RestoreComposerDependencies from cycode.cli.files_collector.sca.python.restore_pip_dependencies import RestorePipDependencies @@ -179,6 +180,8 @@ def _add_dependencies_tree_documents( {'documents_count': len(documents_to_scan), 'is_git_diff': is_git_diff}, ) + clear_npm_workspace_cache() + documents_to_add: dict[str, Document] = {document.path: document for document in documents_to_scan} restore_dependencies_list = _get_restore_handlers(ctx, is_git_diff) diff --git a/cycode/cli/utils/path_utils.py b/cycode/cli/utils/path_utils.py index ea6d2f38..57bb3fc5 100644 --- a/cycode/cli/utils/path_utils.py +++ b/cycode/cli/utils/path_utils.py @@ -133,11 +133,38 @@ def concat_unique_id(filename: str, unique_id: str) -> str: return os.path.join(unique_id, filename) +def _as_scan_root(value: object) -> Optional[str]: + if isinstance(value, (str, os.PathLike)): + return os.fspath(value) or None + + return None + + +def get_scan_roots_from_context(ctx: typer.Context) -> tuple[str, ...]: + """Every path the user asked to scan, in the order they were given. + + Typer declares these arguments as Path, so callers must not assume str. + """ + params = getattr(ctx, 'params', None) + if not isinstance(params, dict): + return () + + scan_roots = [] + + single_root = _as_scan_root(params.get('path')) + if single_root: + scan_roots.append(single_root) + + paths = params.get('paths') + if isinstance(paths, (list, tuple)): + scan_roots.extend(root for root in (_as_scan_root(entry) for entry in paths) if root) + + return tuple(dict.fromkeys(scan_roots)) + + def get_path_from_context(ctx: typer.Context) -> Optional[str]: - path = ctx.params.get('path') - if path is None and 'paths' in ctx.params: - path = ctx.params['paths'][0] - return path + scan_roots = get_scan_roots_from_context(ctx) + return scan_roots[0] if scan_roots else None def normalize_file_path(path: str) -> str: diff --git a/tests/cli/files_collector/sca/npm/test_restore_bun_dependencies.py b/tests/cli/files_collector/sca/npm/test_restore_bun_dependencies.py index 17f189df..e9a64c3c 100644 --- a/tests/cli/files_collector/sca/npm/test_restore_bun_dependencies.py +++ b/tests/cli/files_collector/sca/npm/test_restore_bun_dependencies.py @@ -203,3 +203,64 @@ def test_preexisting_lockfile_is_not_deleted(self, restore_bun: RestoreBunDepend assert result is not None assert lock_path.exists(), f'Pre-existing {BUN_LOCK_FILE_NAME} must not be deleted' + + +class TestIsProjectInWorkspace: + """A bun workspace keeps the only lockfile at the root, so the member folder holds just a manifest.""" + + @staticmethod + def _member_document(member_dir: Path) -> Document: + manifest = member_dir / 'package.json' + return Document(str(manifest), manifest.read_text(), absolute_path=str(manifest)) + + @staticmethod + def _write_workspace_root(root: Path, *, with_lockfile: bool = True) -> None: + (root / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + if with_lockfile: + (root / 'bun.lock').write_text('{"lockfileVersion": 1}') + + def test_member_covered_by_the_root_lockfile_does_not_match( + self, restore_bun: RestoreBunDependencies, tmp_path: Path + ) -> None: + """The root lockfile resolves the member, so restoring it separately would be wrong.""" + self._write_workspace_root(tmp_path) + member_dir = tmp_path / 'packages' / 'app' + member_dir.mkdir(parents=True) + (member_dir / 'package.json').write_text('{"name": "app"}') + + assert restore_bun.is_project(self._member_document(member_dir)) is False + + def test_member_of_a_workspace_without_a_lockfile_falls_back_to_the_signal( + self, restore_bun: RestoreBunDependencies, tmp_path: Path + ) -> None: + """Nothing resolves the member yet, so the packageManager signal still decides.""" + self._write_workspace_root(tmp_path, with_lockfile=False) + member_dir = tmp_path / 'packages' / 'app' + member_dir.mkdir(parents=True) + (member_dir / 'package.json').write_text('{"name": "app", "packageManager": "bun@1.2.3"}') + + assert restore_bun.is_project(self._member_document(member_dir)) is True + + def test_member_with_its_own_lockfile_still_matches( + self, restore_bun: RestoreBunDependencies, tmp_path: Path + ) -> None: + """A lockfile inside the member is authoritative for that member.""" + self._write_workspace_root(tmp_path) + member_dir = tmp_path / 'packages' / 'app' + member_dir.mkdir(parents=True) + (member_dir / 'package.json').write_text('{"name": "app"}') + (member_dir / 'bun.lock').write_text('{"lockfileVersion": 1}') + + assert restore_bun.is_project(self._member_document(member_dir)) is True + + def test_nested_package_that_is_not_a_workspace_member_still_matches( + self, restore_bun: RestoreBunDependencies, tmp_path: Path + ) -> None: + """Without a matching workspaces pattern the root lockfile does not resolve this package.""" + (tmp_path / 'package.json').write_text('{"name": "root"}') + (tmp_path / 'bun.lock').write_text('{"lockfileVersion": 1}') + member_dir = tmp_path / 'nested' + member_dir.mkdir() + (member_dir / 'package.json').write_text('{"name": "nested", "packageManager": "bun@1.2.3"}') + + assert restore_bun.is_project(self._member_document(member_dir)) is True diff --git a/tests/cli/files_collector/sca/npm/test_restore_npm_dependencies.py b/tests/cli/files_collector/sca/npm/test_restore_npm_dependencies.py index 95f94da0..821a7fb1 100644 --- a/tests/cli/files_collector/sca/npm/test_restore_npm_dependencies.py +++ b/tests/cli/files_collector/sca/npm/test_restore_npm_dependencies.py @@ -1,3 +1,4 @@ +import json from pathlib import Path from typing import Optional from unittest.mock import MagicMock, patch @@ -7,6 +8,7 @@ from cycode.cli.files_collector.sca.npm.restore_npm_dependencies import ( NPM_LOCK_FILE_NAME, + NPM_SHRINKWRAP_FILE_NAME, RestoreNpmDependencies, ) from cycode.cli.models import Document @@ -58,6 +60,19 @@ def test_package_json_with_bun_lock_does_not_match( doc = Document(str(tmp_path / 'package.json'), '{"name": "test"}', absolute_path=str(tmp_path / 'package.json')) assert restore_npm.is_project(doc) is False + def test_package_json_with_only_a_binary_bun_lock_still_matches( + self, restore_npm: RestoreNpmDependencies, tmp_path: Path + ) -> None: + """Bun restores only from a text bun.lock, so npm must stay the fallback for a Bun <1.2 project. + + Excluding bun.lockb here would leave such a project with no handler at all and no collected + dependencies, which is worse than an npm-resolved lockfile. + """ + (tmp_path / 'package.json').write_text('{"name": "test"}') + (tmp_path / 'bun.lockb').write_bytes(b'\x00bun-binary-lockfile') + doc = Document(str(tmp_path / 'package.json'), '{"name": "test"}', absolute_path=str(tmp_path / 'package.json')) + assert restore_npm.is_project(doc) is True + def test_tsconfig_json_does_not_match(self, restore_npm: RestoreNpmDependencies) -> None: doc = Document('tsconfig.json', '{}') assert restore_npm.is_project(doc) is False @@ -105,8 +120,20 @@ class TestGetLockFileName: def test_get_lock_file_name(self, restore_npm: RestoreNpmDependencies) -> None: assert restore_npm.get_lock_file_name() == NPM_LOCK_FILE_NAME - def test_get_lock_file_names_contains_only_npm_lock(self, restore_npm: RestoreNpmDependencies) -> None: - assert restore_npm.get_lock_file_names() == [NPM_LOCK_FILE_NAME] + def test_get_lock_file_names_contains_both_npm_lockfile_spellings( + self, restore_npm: RestoreNpmDependencies + ) -> None: + """A project may commit npm-shrinkwrap.json instead of package-lock.json; both must be honoured.""" + assert restore_npm.get_lock_file_names() == [NPM_LOCK_FILE_NAME, NPM_SHRINKWRAP_FILE_NAME] + + def test_restored_name_keeps_the_shrinkwrap_spelling(self, restore_npm: RestoreNpmDependencies) -> None: + """The collected document must report the file we actually read, not a renamed copy.""" + path = str(Path('/repo/npm-shrinkwrap.json')) + assert restore_npm.get_restored_lock_file_name(path) == NPM_SHRINKWRAP_FILE_NAME + + def test_restored_name_defaults_to_package_lock(self, restore_npm: RestoreNpmDependencies) -> None: + path = str(Path('/repo/package-lock.json')) + assert restore_npm.get_restored_lock_file_name(path) == NPM_LOCK_FILE_NAME _BASE_MODULE = 'cycode.cli.files_collector.sca.base_restore_dependencies' @@ -135,6 +162,24 @@ def side_effect( assert result is not None assert not lock_path.exists(), f'{NPM_LOCK_FILE_NAME} must be deleted after restore' + def test_committed_shrinkwrap_is_used_instead_of_regenerating( + self, restore_npm: RestoreNpmDependencies, tmp_path: Path + ) -> None: + """A project may ship npm-shrinkwrap.json; regenerating would re-resolve it against the registry.""" + (tmp_path / 'package.json').write_text('{"name": "test"}') + shrinkwrap_path = tmp_path / NPM_SHRINKWRAP_FILE_NAME + shrinkwrap_path.write_text('{"lockfileVersion": 3, "packages": {"": {}}}') + doc = Document(str(tmp_path / 'package.json'), '{"name": "test"}', absolute_path=str(tmp_path / 'package.json')) + + with patch(f'{_BASE_MODULE}.execute_commands') as mock_execute: + result = restore_npm.try_restore_dependencies(doc) + + mock_execute.assert_not_called() + assert result is not None + assert result.content == shrinkwrap_path.read_text() + assert Path(result.path).name == NPM_SHRINKWRAP_FILE_NAME + assert shrinkwrap_path.exists(), 'A committed lockfile must not be deleted' + def test_preexisting_lockfile_is_not_deleted(self, restore_npm: RestoreNpmDependencies, tmp_path: Path) -> None: (tmp_path / 'package.json').write_text('{"name": "test"}') lock_path = tmp_path / NPM_LOCK_FILE_NAME @@ -159,3 +204,190 @@ def test_package_json_in_cwd_returns_empty_string(self, restore_npm: RestoreNpmD def test_non_package_json_path_returned_unchanged(self, restore_npm: RestoreNpmDependencies) -> None: path = str(Path('/path/to/')) assert restore_npm.prepare_manifest_file_path_for_command(path) == path + + +class TestIsProjectInNpmWorkspace: + @staticmethod + def _member_document(member_dir: Path) -> Document: + manifest = member_dir / 'package.json' + return Document(str(manifest), manifest.read_text(), absolute_path=str(manifest)) + + @staticmethod + def _write_workspace_root(root: Path, *, lockfile_members: Optional[list] = None) -> None: + (root / 'package.json').write_text('{"name": "root", "workspaces": ["frontend"]}') + if lockfile_members is None: + return + + packages = {'': {}} + for member in lockfile_members: + packages[member] = {} + (root / NPM_LOCK_FILE_NAME).write_text(json.dumps({'lockfileVersion': 3, 'packages': packages})) + + def test_workspace_member_covered_by_the_root_lockfile_does_not_match( + self, restore_npm: RestoreNpmDependencies, tmp_path: Path + ) -> None: + """npm installs a workspace from the root lockfile, so a member lockfile is never used.""" + self._write_workspace_root(tmp_path, lockfile_members=['frontend']) + member_dir = tmp_path / 'frontend' + member_dir.mkdir() + (member_dir / 'package.json').write_text('{"name": "frontend"}') + + assert restore_npm.is_project(self._member_document(member_dir)) is False + + def test_workspace_member_not_listed_in_the_root_lockfile_matches( + self, restore_npm: RestoreNpmDependencies, tmp_path: Path + ) -> None: + """The root lockfile covers other members, so this one still needs its own.""" + self._write_workspace_root(tmp_path, lockfile_members=['other']) + member_dir = tmp_path / 'frontend' + member_dir.mkdir() + (member_dir / 'package.json').write_text('{"name": "frontend"}') + + assert restore_npm.is_project(self._member_document(member_dir)) is True + + def test_workspace_root_without_a_lockfile_still_matches( + self, restore_npm: RestoreNpmDependencies, tmp_path: Path + ) -> None: + """Nothing covers the member yet, so the restore must still run.""" + self._write_workspace_root(tmp_path) + member_dir = tmp_path / 'frontend' + member_dir.mkdir() + (member_dir / 'package.json').write_text('{"name": "frontend"}') + + assert restore_npm.is_project(self._member_document(member_dir)) is True + + def test_nested_project_that_is_not_a_workspace_member_matches( + self, restore_npm: RestoreNpmDependencies, tmp_path: Path + ) -> None: + """A monorepo of independent packages: each one needs its own lockfile.""" + (tmp_path / 'package.json').write_text('{"name": "outer"}') + (tmp_path / NPM_LOCK_FILE_NAME).write_text(json.dumps({'lockfileVersion': 3, 'packages': {'': {}}})) + member_dir = tmp_path / 'nested' + member_dir.mkdir() + (member_dir / 'package.json').write_text('{"name": "nested"}') + + assert restore_npm.is_project(self._member_document(member_dir)) is True + + def test_workspace_root_itself_still_matches(self, restore_npm: RestoreNpmDependencies, tmp_path: Path) -> None: + """The root owns the lockfile; try_restore_dependencies reads it instead of regenerating.""" + self._write_workspace_root(tmp_path, lockfile_members=['frontend']) + + manifest = tmp_path / 'package.json' + doc = Document(str(manifest), manifest.read_text(), absolute_path=str(manifest)) + + assert restore_npm.is_project(doc) is True + + def test_root_lockfile_that_is_not_a_json_object_matches( + self, restore_npm: RestoreNpmDependencies, tmp_path: Path + ) -> None: + """A lockfile whose JSON root is not an object must not abort the scan.""" + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["frontend"]}') + (tmp_path / NPM_LOCK_FILE_NAME).write_text('[1, 2, 3]') + member_dir = tmp_path / 'frontend' + member_dir.mkdir() + (member_dir / 'package.json').write_text('{"name": "frontend"}') + + assert restore_npm.is_project(self._member_document(member_dir)) is True + + def test_malformed_root_lockfile_matches(self, restore_npm: RestoreNpmDependencies, tmp_path: Path) -> None: + """Unparseable lockfile: fall back to generating rather than failing the scan.""" + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["frontend"]}') + (tmp_path / NPM_LOCK_FILE_NAME).write_text('this is not json') + member_dir = tmp_path / 'frontend' + member_dir.mkdir() + (member_dir / 'package.json').write_text('{"name": "frontend"}') + + assert restore_npm.is_project(self._member_document(member_dir)) is True + + def test_workspace_member_covered_by_a_root_shrinkwrap_does_not_match( + self, restore_npm: RestoreNpmDependencies, tmp_path: Path + ) -> None: + """npm shrinkwrap just renames the lockfile, so it resolves the workspace the same way.""" + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["frontend"]}') + (tmp_path / NPM_SHRINKWRAP_FILE_NAME).write_text( + json.dumps({'lockfileVersion': 3, 'packages': {'': {}, 'frontend': {}}}) + ) + member_dir = tmp_path / 'frontend' + member_dir.mkdir() + (member_dir / 'package.json').write_text('{"name": "frontend"}') + + assert restore_npm.is_project(self._member_document(member_dir)) is False + + def test_workspace_member_covered_by_a_lockfile_version_2_root_does_not_match( + self, restore_npm: RestoreNpmDependencies, tmp_path: Path + ) -> None: + """npm 7 writes lockfileVersion 2, which carries both "packages" and the legacy "dependencies".""" + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["frontend"]}') + (tmp_path / NPM_LOCK_FILE_NAME).write_text( + json.dumps({'lockfileVersion': 2, 'packages': {'': {}, 'frontend': {}}, 'dependencies': {}}) + ) + member_dir = tmp_path / 'frontend' + member_dir.mkdir() + (member_dir / 'package.json').write_text('{"name": "frontend"}') + + assert restore_npm.is_project(self._member_document(member_dir)) is False + + def test_lockfile_version_1_root_still_matches(self, restore_npm: RestoreNpmDependencies, tmp_path: Path) -> None: + """Workspaces arrived in npm 7 with lockfileVersion 2, so a v1 lockfile never describes one.""" + (tmp_path / 'package.json').write_text('{"name": "root"}') + (tmp_path / NPM_LOCK_FILE_NAME).write_text( + json.dumps({'lockfileVersion': 1, 'dependencies': {'y18n': {'version': '5.0.0'}}}) + ) + member_dir = tmp_path / 'frontend' + member_dir.mkdir() + (member_dir / 'package.json').write_text('{"name": "frontend"}') + + assert restore_npm.is_project(self._member_document(member_dir)) is True + + def test_file_dependency_directory_does_not_match( + self, restore_npm: RestoreNpmDependencies, tmp_path: Path + ) -> None: + """The root lockfile resolves a file: target's dependencies, so a second lockfile would only drift.""" + (tmp_path / 'package.json').write_text('{"name": "root", "dependencies": {"local-lib": "file:local-lib"}}') + (tmp_path / NPM_LOCK_FILE_NAME).write_text( + json.dumps({'lockfileVersion': 3, 'packages': {'': {}, 'local-lib': {}}}) + ) + member_dir = tmp_path / 'local-lib' + member_dir.mkdir() + (member_dir / 'package.json').write_text('{"name": "local-lib"}') + + assert restore_npm.is_project(self._member_document(member_dir)) is False + + def test_a_member_absent_from_the_root_lockfile_still_matches( + self, restore_npm: RestoreNpmDependencies, tmp_path: Path + ) -> None: + """The lockfile names what it resolves; a directory it omits still needs its own lockfile.""" + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + (tmp_path / NPM_LOCK_FILE_NAME).write_text( + json.dumps({'lockfileVersion': 3, 'packages': {'': {}, 'packages/other': {}}}) + ) + member_dir = tmp_path / 'packages' / 'a' / 'b' + member_dir.mkdir(parents=True) + (member_dir / 'package.json').write_text('{"name": "deep"}') + + assert restore_npm.is_project(self._member_document(member_dir)) is True + + def test_workspace_glob_star_matches_a_direct_child( + self, restore_npm: RestoreNpmDependencies, tmp_path: Path + ) -> None: + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + (tmp_path / NPM_LOCK_FILE_NAME).write_text( + json.dumps({'lockfileVersion': 3, 'packages': {'': {}, 'packages/member': {}}}) + ) + member_dir = tmp_path / 'packages' / 'member' + member_dir.mkdir(parents=True) + (member_dir / 'package.json').write_text('{"name": "member"}') + + assert restore_npm.is_project(self._member_document(member_dir)) is False + + def test_workspaces_object_form_is_honoured(self, restore_npm: RestoreNpmDependencies, tmp_path: Path) -> None: + """npm also accepts {"workspaces": {"packages": [...]}}.""" + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": {"packages": ["packages/member"]}}') + (tmp_path / NPM_LOCK_FILE_NAME).write_text( + json.dumps({'lockfileVersion': 3, 'packages': {'': {}, 'packages/member': {}}}) + ) + member_dir = tmp_path / 'packages' / 'member' + member_dir.mkdir(parents=True) + (member_dir / 'package.json').write_text('{"name": "member"}') + + assert restore_npm.is_project(self._member_document(member_dir)) is False diff --git a/tests/cli/files_collector/sca/npm/test_restore_pnpm_dependencies.py b/tests/cli/files_collector/sca/npm/test_restore_pnpm_dependencies.py index 88502578..5f6d1f17 100644 --- a/tests/cli/files_collector/sca/npm/test_restore_pnpm_dependencies.py +++ b/tests/cli/files_collector/sca/npm/test_restore_pnpm_dependencies.py @@ -131,3 +131,64 @@ def test_preexisting_lockfile_is_not_deleted(self, restore_pnpm: RestorePnpmDepe assert result is not None assert lock_path.exists(), f'Pre-existing {PNPM_LOCK_FILE_NAME} must not be deleted' + + +class TestIsProjectInWorkspace: + """A pnpm workspace keeps the only lockfile at the root, so the member folder holds just a manifest.""" + + @staticmethod + def _member_document(member_dir: Path) -> Document: + manifest = member_dir / 'package.json' + return Document(str(manifest), manifest.read_text(), absolute_path=str(manifest)) + + @staticmethod + def _write_workspace_root(root: Path, *, with_lockfile: bool = True) -> None: + (root / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + if with_lockfile: + (root / 'pnpm-lock.yaml').write_text("lockfileVersion: '9.0'\n") + + def test_member_covered_by_the_root_lockfile_does_not_match( + self, restore_pnpm: RestorePnpmDependencies, tmp_path: Path + ) -> None: + """The root lockfile resolves the member, so restoring it separately would be wrong.""" + self._write_workspace_root(tmp_path) + member_dir = tmp_path / 'packages' / 'app' + member_dir.mkdir(parents=True) + (member_dir / 'package.json').write_text('{"name": "app"}') + + assert restore_pnpm.is_project(self._member_document(member_dir)) is False + + def test_member_of_a_workspace_without_a_lockfile_falls_back_to_the_signal( + self, restore_pnpm: RestorePnpmDependencies, tmp_path: Path + ) -> None: + """Nothing resolves the member yet, so the packageManager signal still decides.""" + self._write_workspace_root(tmp_path, with_lockfile=False) + member_dir = tmp_path / 'packages' / 'app' + member_dir.mkdir(parents=True) + (member_dir / 'package.json').write_text('{"name": "app", "packageManager": "pnpm@1.2.3"}') + + assert restore_pnpm.is_project(self._member_document(member_dir)) is True + + def test_member_with_its_own_lockfile_still_matches( + self, restore_pnpm: RestorePnpmDependencies, tmp_path: Path + ) -> None: + """A lockfile inside the member is authoritative for that member.""" + self._write_workspace_root(tmp_path) + member_dir = tmp_path / 'packages' / 'app' + member_dir.mkdir(parents=True) + (member_dir / 'package.json').write_text('{"name": "app"}') + (member_dir / 'pnpm-lock.yaml').write_text("lockfileVersion: '9.0'\n") + + assert restore_pnpm.is_project(self._member_document(member_dir)) is True + + def test_nested_package_that_is_not_a_workspace_member_still_matches( + self, restore_pnpm: RestorePnpmDependencies, tmp_path: Path + ) -> None: + """Without a matching workspaces pattern the root lockfile does not resolve this package.""" + (tmp_path / 'package.json').write_text('{"name": "root"}') + (tmp_path / 'pnpm-lock.yaml').write_text("lockfileVersion: '9.0'\n") + member_dir = tmp_path / 'nested' + member_dir.mkdir() + (member_dir / 'package.json').write_text('{"name": "nested", "packageManager": "pnpm@1.2.3"}') + + assert restore_pnpm.is_project(self._member_document(member_dir)) is True diff --git a/tests/cli/files_collector/sca/npm/test_restore_yarn_dependencies.py b/tests/cli/files_collector/sca/npm/test_restore_yarn_dependencies.py index 88175031..0b1856bd 100644 --- a/tests/cli/files_collector/sca/npm/test_restore_yarn_dependencies.py +++ b/tests/cli/files_collector/sca/npm/test_restore_yarn_dependencies.py @@ -131,3 +131,64 @@ def test_preexisting_lockfile_is_not_deleted(self, restore_yarn: RestoreYarnDepe assert result is not None assert lock_path.exists(), f'Pre-existing {YARN_LOCK_FILE_NAME} must not be deleted' + + +class TestIsProjectInWorkspace: + """A yarn workspace keeps the only lockfile at the root, so the member folder holds just a manifest.""" + + @staticmethod + def _member_document(member_dir: Path) -> Document: + manifest = member_dir / 'package.json' + return Document(str(manifest), manifest.read_text(), absolute_path=str(manifest)) + + @staticmethod + def _write_workspace_root(root: Path, *, with_lockfile: bool = True) -> None: + (root / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + if with_lockfile: + (root / 'yarn.lock').write_text('# yarn lockfile v1\n') + + def test_member_covered_by_the_root_lockfile_does_not_match( + self, restore_yarn: RestoreYarnDependencies, tmp_path: Path + ) -> None: + """The root lockfile resolves the member, so restoring it separately would be wrong.""" + self._write_workspace_root(tmp_path) + member_dir = tmp_path / 'packages' / 'app' + member_dir.mkdir(parents=True) + (member_dir / 'package.json').write_text('{"name": "app"}') + + assert restore_yarn.is_project(self._member_document(member_dir)) is False + + def test_member_of_a_workspace_without_a_lockfile_falls_back_to_the_signal( + self, restore_yarn: RestoreYarnDependencies, tmp_path: Path + ) -> None: + """Nothing resolves the member yet, so the packageManager signal still decides.""" + self._write_workspace_root(tmp_path, with_lockfile=False) + member_dir = tmp_path / 'packages' / 'app' + member_dir.mkdir(parents=True) + (member_dir / 'package.json').write_text('{"name": "app", "packageManager": "yarn@1.2.3"}') + + assert restore_yarn.is_project(self._member_document(member_dir)) is True + + def test_member_with_its_own_lockfile_still_matches( + self, restore_yarn: RestoreYarnDependencies, tmp_path: Path + ) -> None: + """A lockfile inside the member is authoritative for that member.""" + self._write_workspace_root(tmp_path) + member_dir = tmp_path / 'packages' / 'app' + member_dir.mkdir(parents=True) + (member_dir / 'package.json').write_text('{"name": "app"}') + (member_dir / 'yarn.lock').write_text('# yarn lockfile v1\n') + + assert restore_yarn.is_project(self._member_document(member_dir)) is True + + def test_nested_package_that_is_not_a_workspace_member_still_matches( + self, restore_yarn: RestoreYarnDependencies, tmp_path: Path + ) -> None: + """Without a matching workspaces pattern the root lockfile does not resolve this package.""" + (tmp_path / 'package.json').write_text('{"name": "root"}') + (tmp_path / 'yarn.lock').write_text('# yarn lockfile v1\n') + member_dir = tmp_path / 'nested' + member_dir.mkdir() + (member_dir / 'package.json').write_text('{"name": "nested", "packageManager": "yarn@1.2.3"}') + + assert restore_yarn.is_project(self._member_document(member_dir)) is True diff --git a/tests/cli/files_collector/sca/npm/test_workspace.py b/tests/cli/files_collector/sca/npm/test_workspace.py new file mode 100644 index 00000000..460cd13d --- /dev/null +++ b/tests/cli/files_collector/sca/npm/test_workspace.py @@ -0,0 +1,1152 @@ +import contextlib +import json +import logging +from pathlib import Path +from typing import Optional +from unittest.mock import MagicMock + +import pytest +import typer + +from cycode.cli.files_collector.sca.npm import workspace +from cycode.cli.files_collector.sca.npm.restore_bun_dependencies import RestoreBunDependencies +from cycode.cli.files_collector.sca.npm.restore_npm_dependencies import RestoreNpmDependencies +from cycode.cli.files_collector.sca.npm.restore_pnpm_dependencies import RestorePnpmDependencies +from cycode.cli.files_collector.sca.npm.restore_yarn_dependencies import RestoreYarnDependencies +from cycode.cli.files_collector.sca.npm.workspace import ( + clear_cache, + find_covering_workspace, + is_covered_workspace_member, +) +from cycode.cli.files_collector.sca.npm.workspace import resolvers as workspace_resolvers +from cycode.cli.models import Document +from cycode.cli.utils.path_utils import get_scan_roots_from_context + + +@pytest.fixture(autouse=True) +def _clear_workspace_cache() -> None: + """Every test must see the filesystem it just built, not a previous test's memo.""" + clear_cache() + + +_WORKSPACE_LOGGER_NAME = 'SCA NPM Workspace' + + +@contextlib.contextmanager +def _counting_lines(handle: object, consumed: list) -> object: + """Yields the handle's lines while recording how many bytes the caller actually consumed.""" + + def lines() -> object: + for line in handle: + consumed.append(len(line)) + yield line + + try: + yield lines() + finally: + handle.close() + + +def _write_member(root: Path, relative_dir: str, name: str = 'member') -> Path: + member_dir = root / relative_dir + member_dir.mkdir(parents=True, exist_ok=True) + (member_dir / 'package.json').write_text(json.dumps({'name': name})) + return member_dir + + +def _write_pnpm_lockfile(root: Path, members: list) -> None: + lines = ["lockfileVersion: '9.0'", 'importers:', ' .:', ' dependencies: {}'] + for member in members: + lines.append(f' {member}:') + lines.append(' dependencies: {}') + lines.append('packages: {}') + (root / 'pnpm-lock.yaml').write_text('\n'.join(lines) + '\n') + + +def _write_yarn_classic_lockfile(root: Path) -> None: + """Classic yarn records no members, so a root carrying it falls back to the manifest globs.""" + (root / 'yarn.lock').write_text('# yarn lockfile v1\n') + + +def _write_npm_lockfile(root: Path, members: list, file_name: str = 'package-lock.json') -> None: + packages = {'': {'name': 'root'}} + for member in members: + packages[member] = {'name': member} + (root / file_name).write_text(json.dumps({'lockfileVersion': 3, 'packages': packages})) + + +class TestNpmWorkspaceCoverage: + def test_member_covered_by_the_root_lockfile(self, tmp_path: Path) -> None: + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + _write_npm_lockfile(tmp_path, ['packages/app']) + member_dir = _write_member(tmp_path, 'packages/app') + + coverage = find_covering_workspace(str(member_dir)) + + assert coverage is not None + assert coverage.package_manager == 'npm' + assert coverage.lock_file == tmp_path / 'package-lock.json' + + def test_member_covered_by_a_root_shrinkwrap(self, tmp_path: Path) -> None: + """npm shrinkwrap only renames the lockfile, so it resolves the workspace the same way.""" + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + _write_npm_lockfile(tmp_path, ['packages/app'], file_name='npm-shrinkwrap.json') + member_dir = _write_member(tmp_path, 'packages/app') + + coverage = find_covering_workspace(str(member_dir)) + + assert coverage is not None + assert coverage.lock_file.name == 'npm-shrinkwrap.json' + + def test_stale_root_lockfile_missing_the_member_is_not_coverage(self, tmp_path: Path) -> None: + """A root lockfile written before the member existed cannot resolve it, so it must still restore.""" + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + _write_npm_lockfile(tmp_path, ['packages/other']) + member_dir = _write_member(tmp_path, 'packages/app') + + assert find_covering_workspace(str(member_dir)) is None + + def test_a_file_dependency_target_is_covered(self, tmp_path: Path) -> None: + """The root lockfile resolves a file: target's dependencies exactly as it does a member's. + + Verified against npm 11: both get a packages entry and their ranges pinned under the root + node_modules, so generating a second lockfile inside the target can only drift from it. + """ + (tmp_path / 'package.json').write_text('{"name": "root", "dependencies": {"lib": "file:lib"}}') + _write_npm_lockfile(tmp_path, ['lib']) + member_dir = _write_member(tmp_path, 'lib') + + assert find_covering_workspace(str(member_dir)) is not None + + def test_lockfile_version_1_root_is_not_coverage(self, tmp_path: Path) -> None: + """Workspaces arrived in npm 7 with lockfileVersion 2, so a v1 lockfile never describes one.""" + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + (tmp_path / 'package-lock.json').write_text(json.dumps({'lockfileVersion': 1, 'dependencies': {}})) + member_dir = _write_member(tmp_path, 'packages/app') + + assert find_covering_workspace(str(member_dir)) is None + + def test_malformed_root_lockfile_is_not_coverage(self, tmp_path: Path) -> None: + """An unparseable lockfile must fall back to restoring rather than failing the scan.""" + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + (tmp_path / 'package-lock.json').write_text('this is not json') + member_dir = _write_member(tmp_path, 'packages/app') + + assert find_covering_workspace(str(member_dir)) is None + + +class TestNonNpmWorkspaceCoverage: + @pytest.mark.parametrize( + ('lock_file_name', 'lock_file_content', 'expected_package_manager'), + [ + ('yarn.lock', '# yarn lockfile v1\n', 'yarn'), + ('bun.lock', '{"lockfileVersion": 1}', 'bun'), + ('deno.lock', '{"version": "4"}', 'deno'), + ], + ) + def test_member_covered_by_a_root_lockfile_of_another_package_manager( + self, tmp_path: Path, lock_file_name: str, lock_file_content: str, expected_package_manager: str + ) -> None: + """The member folder holds no lockfile, so without this npm would claim it and generate the wrong one.""" + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + (tmp_path / lock_file_name).write_text(lock_file_content) + member_dir = _write_member(tmp_path, 'packages/app') + + coverage = find_covering_workspace(str(member_dir)) + + assert coverage is not None + assert coverage.package_manager == expected_package_manager + + def test_member_covered_by_a_binary_bun_lockfile(self, tmp_path: Path) -> None: + """Bun <1.2 writes a binary bun.lockb; it still proves Bun owns the workspace.""" + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + (tmp_path / 'bun.lockb').write_bytes(b'\x00bun-binary-lockfile') + member_dir = _write_member(tmp_path, 'packages/app') + + coverage = find_covering_workspace(str(member_dir)) + + assert coverage is not None + assert coverage.package_manager == 'bun' + + def test_pnpm_declares_its_workspace_outside_package_json(self, tmp_path: Path) -> None: + """pnpm lists members in pnpm-workspace.yaml, so the package.json "workspaces" field is absent.""" + (tmp_path / 'package.json').write_text('{"name": "root", "private": true}') + (tmp_path / 'pnpm-workspace.yaml').write_text('packages:\n - "packages/*"\n') + _write_pnpm_lockfile(tmp_path, ['packages/app']) + member_dir = _write_member(tmp_path, 'packages/app') + + coverage = find_covering_workspace(str(member_dir)) + + assert coverage is not None + assert coverage.package_manager == 'pnpm' + + def test_malformed_pnpm_workspace_file_is_not_coverage(self, tmp_path: Path) -> None: + (tmp_path / 'package.json').write_text('{"name": "root", "private": true}') + (tmp_path / 'pnpm-workspace.yaml').write_text('packages: [unclosed\n - "oops"\n') + (tmp_path / 'pnpm-lock.yaml').write_text("lockfileVersion: '9.0'\n") + member_dir = _write_member(tmp_path, 'packages/app') + + assert find_covering_workspace(str(member_dir)) is None + + def test_root_without_any_lockfile_is_not_coverage(self, tmp_path: Path) -> None: + """Nothing resolves the member yet, so the restore must still run.""" + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + member_dir = _write_member(tmp_path, 'packages/app') + + assert find_covering_workspace(str(member_dir)) is None + + +class TestWorkspacePatternMatching: + def test_single_star_stops_at_a_path_separator(self, tmp_path: Path) -> None: + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + _write_yarn_classic_lockfile(tmp_path) + member_dir = _write_member(tmp_path, 'packages/a/b') + + assert find_covering_workspace(str(member_dir)) is None + + def test_double_star_crosses_a_path_separator(self, tmp_path: Path) -> None: + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/**"]}') + _write_yarn_classic_lockfile(tmp_path) + member_dir = _write_member(tmp_path, 'packages/a/b') + + assert find_covering_workspace(str(member_dir)) is not None + + @pytest.mark.parametrize( + ('pattern', 'member_path', 'matches'), + [ + # a trailing globstar covers the directory itself, because workspace discovery globs + # /package.json and ** spans zero or more segments + ('src/app/**', 'src/app', True), + ('src/app/**', 'src/app/nested', True), + ('src/app/**', 'src/apple', False), + ('packages/**', 'packages', True), + ('packages/**', 'packages/a/b', True), + ('packages/*', 'packages/a', True), + ('packages/*', 'packages/a/b', False), + ('**/src/test/**', 'src/test', True), + ('**/src/test/**', 'a/src/test/b', True), + ], + ) + def test_globstar_spans_zero_or_more_segments( + self, tmp_path: Path, pattern: str, member_path: str, matches: bool + ) -> None: + """pnpm records src/app as a member of src/app/**, so the separator cannot be mandatory.""" + (tmp_path / 'package.json').write_text(json.dumps({'name': 'root', 'workspaces': [pattern]})) + _write_yarn_classic_lockfile(tmp_path) + member_dir = _write_member(tmp_path, member_path) + + assert (find_covering_workspace(str(member_dir)) is not None) is matches + + def test_workspaces_object_form_is_honoured(self, tmp_path: Path) -> None: + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": {"packages": ["packages/*"]}}') + _write_yarn_classic_lockfile(tmp_path) + member_dir = _write_member(tmp_path, 'packages/app') + + assert find_covering_workspace(str(member_dir)) is not None + + def test_leading_dot_slash_and_trailing_slash_are_normalized(self, tmp_path: Path) -> None: + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["./packages/app/"]}') + _write_yarn_classic_lockfile(tmp_path) + member_dir = _write_member(tmp_path, 'packages/app') + + assert find_covering_workspace(str(member_dir)) is not None + + def test_negated_pattern_excludes_a_member(self, tmp_path: Path) -> None: + """packages/* matches, but the exclusion wins — treating "!" as a literal would lose this project.""" + (tmp_path / 'package.json').write_text( + '{"name": "root", "workspaces": ["packages/*", "!packages/legacy"]}', + ) + _write_yarn_classic_lockfile(tmp_path) + member_dir = _write_member(tmp_path, 'packages/legacy') + + assert find_covering_workspace(str(member_dir)) is None + + def test_negated_pattern_leaves_other_members_covered(self, tmp_path: Path) -> None: + (tmp_path / 'package.json').write_text( + '{"name": "root", "workspaces": ["packages/*", "!packages/legacy"]}', + ) + _write_yarn_classic_lockfile(tmp_path) + member_dir = _write_member(tmp_path, 'packages/app') + + assert find_covering_workspace(str(member_dir)) is not None + + def test_non_string_workspace_entries_are_ignored(self, tmp_path: Path) -> None: + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": [null, 7, "packages/*"]}') + _write_yarn_classic_lockfile(tmp_path) + member_dir = _write_member(tmp_path, 'packages/app') + + assert find_covering_workspace(str(member_dir)) is not None + + +class TestWorkspaceRootWalk: + def test_a_workspace_nested_inside_another_workspace(self, tmp_path: Path) -> None: + """The inner root owns the member; the walk must stop at the nearest declaring ancestor.""" + (tmp_path / 'package.json').write_text('{"name": "outer", "workspaces": ["apps/*"]}') + (tmp_path / 'yarn.lock').write_text('# yarn lockfile v1\n') + + inner_root = tmp_path / 'apps' / 'inner' + inner_root.mkdir(parents=True) + (inner_root / 'package.json').write_text('{"name": "inner", "workspaces": ["packages/*"]}') + _write_npm_lockfile(inner_root, ['packages/app']) + member_dir = _write_member(inner_root, 'packages/app') + + coverage = find_covering_workspace(str(member_dir)) + + assert coverage is not None + assert coverage.package_manager == 'npm' + assert coverage.lock_file == inner_root / 'package-lock.json' + + def test_outer_workspace_covers_a_member_the_inner_root_does_not_declare(self, tmp_path: Path) -> None: + """The walk continues past an ancestor that declares no matching pattern.""" + (tmp_path / 'package.json').write_text('{"name": "outer", "workspaces": ["apps/**"]}') + (tmp_path / 'yarn.lock').write_text('# yarn lockfile v1\n') + + inner_root = tmp_path / 'apps' / 'inner' + inner_root.mkdir(parents=True) + (inner_root / 'package.json').write_text('{"name": "inner"}') + member_dir = _write_member(inner_root, 'packages/app') + + coverage = find_covering_workspace(str(member_dir)) + + assert coverage is not None + assert coverage.package_manager == 'yarn' + + def test_the_walk_stops_at_the_git_repository_root(self, tmp_path: Path) -> None: + """A manifest above the repository must never suppress a project inside it.""" + repo_root = tmp_path / 'repo' + repo_root.mkdir() + (repo_root / '.git').mkdir() + (repo_root / 'package.json').write_text('{"name": "repo"}') + + (tmp_path / 'package.json').write_text('{"name": "outside", "workspaces": ["repo/**"]}') + _write_npm_lockfile(tmp_path, ['repo/packages/app']) + + member_dir = _write_member(repo_root, 'packages/app') + + assert find_covering_workspace(str(member_dir)) is None + + def test_a_git_file_also_marks_the_repository_root(self, tmp_path: Path) -> None: + """Worktrees and submodules carry a .git file rather than a directory.""" + repo_root = tmp_path / 'repo' + repo_root.mkdir() + (repo_root / '.git').write_text('gitdir: ../.git/worktrees/repo\n') + (repo_root / 'package.json').write_text('{"name": "repo"}') + + (tmp_path / 'package.json').write_text('{"name": "outside", "workspaces": ["repo/**"]}') + _write_npm_lockfile(tmp_path, ['repo/packages/app']) + + member_dir = _write_member(repo_root, 'packages/app') + + assert find_covering_workspace(str(member_dir)) is None + + def test_no_manifest_dir_is_not_coverage(self) -> None: + assert find_covering_workspace(None) is None + assert find_covering_workspace('') is None + + +class TestIsCoveredWorkspaceMember: + def test_a_root_outside_the_scanned_path_is_not_coverage( + self, tmp_path: Path, caplog: pytest.LogCaptureFixture + ) -> None: + """The root lockfile will not be collected, so treating it as coverage yields no data at all. + + Scanning only the member folder used to report clean with nothing but a log line to say why. + Restoring the member gives versions that may drift from the root, which still beats silence. + """ + (tmp_path / '.git').mkdir() + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + _write_npm_lockfile(tmp_path, ['packages/app']) + member_dir = _write_member(tmp_path, 'packages/app') + + with caplog.at_level(logging.WARNING, logger=_WORKSPACE_LOGGER_NAME): + covered = is_covered_workspace_member(str(member_dir), 'packages/app/package.json', (str(member_dir),)) + + assert covered is False + assert 'outside the scanned path' in caplog.text + + def test_does_not_warn_when_the_workspace_root_is_scanned_too( + self, tmp_path: Path, caplog: pytest.LogCaptureFixture + ) -> None: + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + _write_npm_lockfile(tmp_path, ['packages/app']) + member_dir = _write_member(tmp_path, 'packages/app') + + with caplog.at_level(logging.WARNING, logger=_WORKSPACE_LOGGER_NAME): + covered = is_covered_workspace_member(str(member_dir), 'packages/app/package.json', (str(tmp_path),)) + + assert covered is True + assert 'outside the scanned path' not in caplog.text + + def test_uncovered_member_is_not_reported(self, tmp_path: Path) -> None: + member_dir = _write_member(tmp_path, 'packages/app') + + assert is_covered_workspace_member(str(member_dir), 'packages/app/package.json', (str(tmp_path),)) is False + + +class TestCaching: + def test_the_root_lockfile_is_parsed_once_for_all_members(self, tmp_path: Path) -> None: + """200 members must not mean 200 parses of the same multi-megabyte lockfile.""" + members = [f'packages/m{index}' for index in range(25)] + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + _write_npm_lockfile(tmp_path, members) + member_dirs = [_write_member(tmp_path, member) for member in members] + + parsed_paths = [] + original_read_json_object = workspace_resolvers.read_json_object + + def counting_read_json_object(path: Path) -> object: + parsed_paths.append(str(path)) + return original_read_json_object(path) + + workspace_resolvers.read_json_object = counting_read_json_object + try: + coverages = [find_covering_workspace(str(member_dir)) for member_dir in member_dirs] + finally: + workspace_resolvers.read_json_object = original_read_json_object + + assert all(coverage is not None for coverage in coverages) + assert parsed_paths.count(str(tmp_path / 'package-lock.json')) == 1 + + def test_clearing_the_cache_picks_up_an_edited_lockfile(self, tmp_path: Path) -> None: + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + _write_npm_lockfile(tmp_path, ['packages/other']) + member_dir = _write_member(tmp_path, 'packages/app') + + assert find_covering_workspace(str(member_dir)) is None + + _write_npm_lockfile(tmp_path, ['packages/app']) + clear_cache() + + assert find_covering_workspace(str(member_dir)) is not None + + +class TestNoHandlerClaimsACoveredMember: + """The npm fallback only helps if every dedicated handler declines a member the root already resolves. + + Before this, a yarn/pnpm/bun workspace member was claimed by npm: the dedicated handler + declined because the member folder holds no lockfile, and npm generated a package-lock.json + for a project that does not install with npm. + """ + + @staticmethod + def _handlers(tmp_path: Path) -> dict: + ctx = MagicMock(spec=typer.Context) + ctx.obj = {'monitor': False} + ctx.params = {'path': str(tmp_path)} + return { + 'yarn': RestoreYarnDependencies(ctx, is_git_diff=False, command_timeout=30), + 'pnpm': RestorePnpmDependencies(ctx, is_git_diff=False, command_timeout=30), + 'bun': RestoreBunDependencies(ctx, is_git_diff=False, command_timeout=30), + 'npm': RestoreNpmDependencies(ctx, is_git_diff=False, command_timeout=30), + } + + def _claimants(self, tmp_path: Path, member_dir: Path) -> list: + manifest = member_dir / 'package.json' + document = Document(str(manifest), manifest.read_text(), absolute_path=str(manifest)) + return [name for name, handler in self._handlers(tmp_path).items() if handler.is_project(document)] + + @pytest.mark.parametrize( + ('lock_file_name', 'lock_file_content'), + [ + ('yarn.lock', '# yarn lockfile v1\n'), + ('bun.lock', '{"lockfileVersion": 1}'), + ('bun.lockb', '\x00binary'), + ], + ) + def test_no_handler_claims_a_member_of_a_non_npm_workspace( + self, tmp_path: Path, lock_file_name: str, lock_file_content: str + ) -> None: + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + (tmp_path / lock_file_name).write_text(lock_file_content) + member_dir = _write_member(tmp_path, 'packages/app') + + assert self._claimants(tmp_path, member_dir) == [] + + def test_no_handler_claims_a_member_of_a_pnpm_workspace(self, tmp_path: Path) -> None: + (tmp_path / 'package.json').write_text('{"name": "root", "private": true}') + (tmp_path / 'pnpm-workspace.yaml').write_text('packages:\n - "packages/*"\n') + _write_pnpm_lockfile(tmp_path, ['packages/app']) + member_dir = _write_member(tmp_path, 'packages/app') + + assert self._claimants(tmp_path, member_dir) == [] + + def test_no_handler_claims_a_member_of_an_npm_workspace(self, tmp_path: Path) -> None: + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + _write_npm_lockfile(tmp_path, ['packages/app']) + member_dir = _write_member(tmp_path, 'packages/app') + + assert self._claimants(tmp_path, member_dir) == [] + + def test_a_member_declaring_yarn_is_still_skipped_when_the_root_covers_it(self, tmp_path: Path) -> None: + """The packageManager signal must not resurrect a member the root lockfile already resolves.""" + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + (tmp_path / 'yarn.lock').write_text('# yarn lockfile v1\n') + member_dir = tmp_path / 'packages' / 'app' + member_dir.mkdir(parents=True) + (member_dir / 'package.json').write_text('{"name": "app", "packageManager": "yarn@4.0.2"}') + + assert self._claimants(tmp_path, member_dir) == [] + + def test_a_member_with_its_own_lockfile_is_still_claimed(self, tmp_path: Path) -> None: + """A lockfile inside the member is authoritative for that member, workspace or not.""" + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + (tmp_path / 'yarn.lock').write_text('# yarn lockfile v1\n') + member_dir = _write_member(tmp_path, 'packages/app') + (member_dir / 'yarn.lock').write_text('# yarn lockfile v1\n') + + assert self._claimants(tmp_path, member_dir) == ['yarn'] + + def test_an_independent_nested_package_is_still_claimed_by_npm(self, tmp_path: Path) -> None: + """A monorepo of unrelated packages is not a workspace; each one still needs its own lockfile.""" + (tmp_path / 'package.json').write_text('{"name": "root"}') + (tmp_path / 'yarn.lock').write_text('# yarn lockfile v1\n') + member_dir = _write_member(tmp_path, 'nested') + + assert self._claimants(tmp_path, member_dir) == ['npm'] + + def test_scanning_only_the_member_folder_still_collects_something(self, tmp_path: Path) -> None: + """Nothing else in this scan carries the member's dependencies, so a handler must take it.""" + (tmp_path / '.git').mkdir() + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + (tmp_path / 'yarn.lock').write_text('# yarn lockfile v1\n') + member_dir = _write_member(tmp_path, 'packages/app') + + manifest = member_dir / 'package.json' + ctx = MagicMock(spec=typer.Context) + ctx.obj = {'monitor': False} + ctx.params = {'path': str(member_dir)} + document = Document(str(manifest), manifest.read_text(), absolute_path=str(manifest)) + + npm = RestoreNpmDependencies(ctx, is_git_diff=False, command_timeout=30) + assert npm.is_project(document) is True + + def test_scanning_the_workspace_root_still_skips_the_member(self, tmp_path: Path) -> None: + """The root lockfile is collected by this scan, so the member must not be restored as well.""" + (tmp_path / '.git').mkdir() + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + (tmp_path / 'yarn.lock').write_text('# yarn lockfile v1\n') + member_dir = _write_member(tmp_path, 'packages/app') + + manifest = member_dir / 'package.json' + ctx = MagicMock(spec=typer.Context) + ctx.obj = {'monitor': False} + ctx.params = {'paths': [tmp_path]} + document = Document(str(manifest), manifest.read_text(), absolute_path=str(manifest)) + + npm = RestoreNpmDependencies(ctx, is_git_diff=False, command_timeout=30) + assert npm.is_project(document) is False + + +class TestDeclarationSourceIsNotShared: + """pnpm reads only pnpm-workspace.yaml; npm/yarn/bun read only package.json "workspaces".""" + + def test_pnpm_lockfile_does_not_cover_a_member_declared_only_in_package_json(self, tmp_path: Path) -> None: + """A repo migrated to pnpm may still carry a stale workspaces array; it does not make a pnpm member.""" + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + (tmp_path / 'pnpm-lock.yaml').write_text("lockfileVersion: '9.0'\n") + member_dir = _write_member(tmp_path, 'packages/app') + + assert find_covering_workspace(str(member_dir)) is None + + def test_pnpm_workspace_yaml_does_not_cover_a_yarn_member(self, tmp_path: Path) -> None: + """yarn.lock coverage must come from package.json, not from a leftover pnpm-workspace.yaml.""" + (tmp_path / 'package.json').write_text('{"name": "root", "private": true}') + (tmp_path / 'pnpm-workspace.yaml').write_text('packages:\n - "packages/*"\n') + (tmp_path / 'yarn.lock').write_text('# yarn lockfile v1\n') + member_dir = _write_member(tmp_path, 'packages/app') + + assert find_covering_workspace(str(member_dir)) is None + + def test_a_pnpm_member_excluded_in_yaml_is_not_covered(self, tmp_path: Path) -> None: + (tmp_path / 'package.json').write_text('{"name": "root", "private": true}') + (tmp_path / 'pnpm-workspace.yaml').write_text('packages:\n - "packages/*"\n - "!packages/legacy"\n') + (tmp_path / 'pnpm-lock.yaml').write_text("lockfileVersion: '9.0'\n") + member_dir = _write_member(tmp_path, 'packages/legacy') + + assert find_covering_workspace(str(member_dir)) is None + + +class TestWalkBoundary: + def test_without_a_git_root_the_walk_stops_at_the_scanned_path(self, tmp_path: Path) -> None: + """An extracted tarball has no .git; a manifest above the scanned path must not suppress a project.""" + scanned = tmp_path / 'checkout' + scanned.mkdir() + (scanned / 'package.json').write_text('{"name": "checkout"}') + + (tmp_path / 'package.json').write_text('{"name": "stray", "workspaces": ["**"]}') + _write_npm_lockfile(tmp_path, ['checkout/packages/app']) + + member_dir = _write_member(scanned, 'packages/app') + + assert find_covering_workspace(str(member_dir), (str(scanned),)) is None + + def test_without_a_git_root_the_walk_still_reaches_the_scanned_path(self, tmp_path: Path) -> None: + """Stopping at the scanned path must not stop before it.""" + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + _write_npm_lockfile(tmp_path, ['packages/app']) + member_dir = _write_member(tmp_path, 'packages/app') + + assert find_covering_workspace(str(member_dir), (str(tmp_path),)) is not None + + def test_a_git_root_lets_the_walk_pass_above_the_scanned_path(self, tmp_path: Path) -> None: + """Scanning one member of a real repository must still find the workspace root above it.""" + (tmp_path / '.git').mkdir() + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + _write_npm_lockfile(tmp_path, ['packages/app']) + member_dir = _write_member(tmp_path, 'packages/app') + + assert find_covering_workspace(str(member_dir), (str(member_dir),)) is not None + + +class TestUnscannedRootWarning: + def test_the_warning_is_emitted_once_even_though_four_handlers_ask( + self, tmp_path: Path, caplog: pytest.LogCaptureFixture + ) -> None: + """All four handlers consult the same check; the user must not see the same skip four times.""" + (tmp_path / '.git').mkdir() + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + _write_npm_lockfile(tmp_path, ['packages/app']) + member_dir = _write_member(tmp_path, 'packages/app') + + with caplog.at_level(logging.WARNING, logger=_WORKSPACE_LOGGER_NAME): + for _ in range(4): + is_covered_workspace_member(str(member_dir), 'packages/app/package.json', (str(member_dir),)) + + assert caplog.text.count('outside the scanned path') == 1 + + def test_a_second_scan_root_containing_the_workspace_suppresses_the_warning( + self, tmp_path: Path, caplog: pytest.LogCaptureFixture + ) -> None: + """cycode scan path ./frontend ./repo-root scans the root too, so there is nothing to warn about.""" + (tmp_path / '.git').mkdir() + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + _write_npm_lockfile(tmp_path, ['packages/app']) + member_dir = _write_member(tmp_path, 'packages/app') + + with caplog.at_level(logging.WARNING, logger=_WORKSPACE_LOGGER_NAME): + covered = is_covered_workspace_member( + str(member_dir), 'packages/app/package.json', (str(member_dir), str(tmp_path)) + ) + + assert covered is True + assert 'outside the scanned path' not in caplog.text + + +class TestLogNoise: + def test_absent_ancestor_manifests_are_not_logged(self, tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None: + """The walk visits every ancestor; an expected absence must not look like a parse failure.""" + member_dir = _write_member(tmp_path, 'a/b/c') + + with caplog.at_level(logging.DEBUG, logger=_WORKSPACE_LOGGER_NAME): + find_covering_workspace(str(member_dir), (str(tmp_path),)) + + assert 'Could not read' not in caplog.text + + def test_a_genuine_parse_failure_is_still_logged(self, tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None: + """Silencing the expected absences must not also silence a real malformed file.""" + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + (tmp_path / 'package-lock.json').write_text('this is not json') + member_dir = _write_member(tmp_path, 'packages/app') + + with caplog.at_level(logging.DEBUG, logger=_WORKSPACE_LOGGER_NAME): + find_covering_workspace(str(member_dir), (str(tmp_path),)) + + assert 'Could not read' in caplog.text + + +class TestFileNamesHaveASingleSource: + """workspace.py owns every file name in the npm module. + + Each handler previously declared its own copy, so the workspace coverage table and the + handler that consumes it could drift apart over time without anything failing. + """ + + def test_each_handler_reuses_the_shared_name(self) -> None: + from cycode.cli.files_collector.sca.npm import ( + restore_bun_dependencies, + restore_deno_dependencies, + restore_npm_dependencies, + restore_pnpm_dependencies, + restore_yarn_dependencies, + ) + + assert restore_yarn_dependencies.YARN_LOCK_FILE_NAME is workspace.YARN_LOCK_FILE_NAME + assert restore_pnpm_dependencies.PNPM_LOCK_FILE_NAME is workspace.PNPM_LOCK_FILE_NAME + assert restore_bun_dependencies.BUN_LOCK_FILE_NAME is workspace.BUN_LOCK_FILE_NAME + assert restore_deno_dependencies.DENO_LOCK_FILE_NAME is workspace.DENO_LOCK_FILE_NAME + assert restore_npm_dependencies.NPM_LOCK_FILE_NAME is workspace.NPM_LOCK_FILE_NAME + assert restore_npm_dependencies.NPM_SHRINKWRAP_FILE_NAME is workspace.NPM_SHRINKWRAP_FILE_NAME + + for module in ( + restore_npm_dependencies.NPM_MANIFEST_FILE_NAME, + restore_yarn_dependencies.YARN_MANIFEST_FILE_NAME, + restore_pnpm_dependencies.PNPM_MANIFEST_FILE_NAME, + restore_bun_dependencies.BUN_MANIFEST_FILE_NAME, + ): + assert module is workspace.MANIFEST_FILE_NAME + + def test_every_name_is_declared_only_in_workspace(self) -> None: + """A new literal anywhere else in the module reintroduces exactly the drift this prevents.""" + module_dir = Path(workspace.__file__).parent.parent + names = ( + 'package.json', + 'package-lock.json', + 'npm-shrinkwrap.json', + 'yarn.lock', + 'pnpm-lock.yaml', + 'pnpm-workspace.yaml', + 'bun.lock', + 'bun.lockb', + 'deno.lock', + ) + + single_source = Path(workspace.__file__).parent / 'names.py' + offenders = {} + for source in sorted(module_dir.rglob('*.py')): + if source == single_source: + continue + + text = source.read_text(encoding='UTF-8') + declared = [name for name in names if f"'{name}'" in text] + if declared: + offenders[str(source.relative_to(module_dir))] = declared + + assert offenders == {}, f'file names must come from workspace/names.py, but found literals in: {offenders}' + + def test_the_alternative_lockfiles_come_from_the_shared_table(self) -> None: + """npm declines a project owned by another package manager; bun.lockb is the deliberate exception.""" + from cycode.cli.files_collector.sca.npm.restore_npm_dependencies import _ALTERNATIVE_LOCK_FILES + + non_npm_names = { + lock_file_name + for resolver in workspace.MEMBER_RESOLVERS + if resolver.package_manager != workspace.NPM_PACKAGE_MANAGER + for lock_file_name in resolver.lock_file_names + } + + assert set(_ALTERNATIVE_LOCK_FILES) <= non_npm_names + assert non_npm_names - set(_ALTERNATIVE_LOCK_FILES) == {workspace.BUN_BINARY_LOCK_FILE_NAME} + + +class TestScanRootsFromContext: + """Typer delivers Path objects, so a str-only filter silently discards every real scan root.""" + + @staticmethod + def _ctx(params: dict) -> typer.Context: + ctx = MagicMock(spec=typer.Context) + ctx.params = params + return ctx + + @pytest.mark.parametrize( + ('command', 'params'), + [ + ('scan path', {'paths': [Path('/repo/member')]}), + ('scan path (tuple)', {'paths': (Path('/repo/member'),)}), + ('scan repository', {'path': Path('/repo/member')}), + ('report sbom path', {'path': Path('/repo/member')}), + ], + ) + def test_path_objects_are_accepted(self, command: str, params: dict) -> None: + assert get_scan_roots_from_context(self._ctx(params)) == (str(Path('/repo/member')),), command + + def test_strings_are_still_accepted(self) -> None: + assert get_scan_roots_from_context(self._ctx({'path': '/repo/member'})) == ('/repo/member',) + + def test_every_scanned_path_is_returned(self) -> None: + """cycode scan path ./a ./b scans both, so both must count as scan roots.""" + params = {'paths': [Path('/repo/a'), Path('/repo/b')]} + + assert get_scan_roots_from_context(self._ctx(params)) == (str(Path('/repo/a')), str(Path('/repo/b'))) + + def test_empty_and_missing_params_are_safe(self) -> None: + assert get_scan_roots_from_context(self._ctx({})) == () + assert get_scan_roots_from_context(self._ctx({'path': None, 'paths': None})) == () + assert get_scan_roots_from_context(MagicMock(spec=typer.Context)) == () + + +class TestSymlinkedScanRoot: + def test_a_scan_root_reached_through_a_symlink_is_recognised( + self, tmp_path: Path, caplog: pytest.LogCaptureFixture + ) -> None: + """typer resolves the scan root but the walk does not, so both sides must be resolved. + + On macOS /tmp and /var are symlinks, so this is the ordinary case rather than an exotic one. + """ + real_repo = tmp_path / 'real' + real_repo.mkdir() + (real_repo / '.git').mkdir() + (real_repo / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + _write_npm_lockfile(real_repo, ['packages/app']) + member_dir = _write_member(real_repo, 'packages/app') + + linked_repo = tmp_path / 'link' + linked_repo.symlink_to(real_repo) + + with caplog.at_level(logging.WARNING, logger=_WORKSPACE_LOGGER_NAME): + covered = is_covered_workspace_member(str(member_dir), 'packages/app/package.json', (str(linked_repo),)) + + assert covered is True + assert 'outside the scanned path' not in caplog.text + + +class TestPnpmLockfileMembership: + """pnpm records its members under importers:, so a stale lockfile must not claim coverage.""" + + def test_a_member_listed_in_importers_is_covered(self, tmp_path: Path) -> None: + (tmp_path / 'package.json').write_text('{"name": "root", "private": true}') + (tmp_path / 'pnpm-workspace.yaml').write_text('packages:\n - "packages/*"\n') + _write_pnpm_lockfile(tmp_path, ['packages/app']) + member_dir = _write_member(tmp_path, 'packages/app') + + coverage = find_covering_workspace(str(member_dir)) + + assert coverage is not None + assert coverage.package_manager == 'pnpm' + + def test_a_member_missing_from_a_stale_lockfile_is_not_covered(self, tmp_path: Path) -> None: + """Declared in pnpm-workspace.yaml but never installed: the lockfile cannot resolve it.""" + (tmp_path / 'package.json').write_text('{"name": "root", "private": true}') + (tmp_path / 'pnpm-workspace.yaml').write_text('packages:\n - "packages/*"\n') + _write_pnpm_lockfile(tmp_path, ['packages/other']) + member_dir = _write_member(tmp_path, 'packages/app') + + assert find_covering_workspace(str(member_dir)) is None + + def test_the_root_importer_is_not_a_member(self, tmp_path: Path) -> None: + """pnpm keys the workspace root as '.', which must never match a member path.""" + (tmp_path / 'package.json').write_text('{"name": "root", "private": true}') + (tmp_path / 'pnpm-workspace.yaml').write_text('packages:\n - "."\n') + _write_pnpm_lockfile(tmp_path, []) + member_dir = _write_member(tmp_path, 'packages/app') + + assert find_covering_workspace(str(member_dir)) is None + + def test_only_the_importers_block_is_parsed(self, tmp_path: Path) -> None: + """A large packages: block must not be parsed just to read the member list.""" + (tmp_path / 'package.json').write_text('{"name": "root", "private": true}') + (tmp_path / 'pnpm-workspace.yaml').write_text('packages:\n - "packages/*"\n') + + lines = ["lockfileVersion: '9.0'", 'importers:', ' .:', ' dependencies: {}'] + lines += [' packages/app:', ' dependencies: {}', 'packages:'] + lines += [f' dep{index}@1.0.0: {{resolution: {{integrity: sha512-x}}}}' for index in range(2000)] + (tmp_path / 'pnpm-lock.yaml').write_text('\n'.join(lines) + '\n') + + section = workspace_resolvers._read_pnpm_importers_section(tmp_path / 'pnpm-lock.yaml') + + assert 'packages/app' in section + assert 'dep0@1.0.0' not in section + + member_dir = _write_member(tmp_path, 'packages/app') + assert find_covering_workspace(str(member_dir)) is not None + + def test_a_malformed_importers_block_is_not_coverage(self, tmp_path: Path) -> None: + (tmp_path / 'package.json').write_text('{"name": "root", "private": true}') + (tmp_path / 'pnpm-workspace.yaml').write_text('packages:\n - "packages/*"\n') + (tmp_path / 'pnpm-lock.yaml').write_text("lockfileVersion: '9.0'\nimporters:\n [unclosed\n") + member_dir = _write_member(tmp_path, 'packages/app') + + assert find_covering_workspace(str(member_dir)) is None + + +class TestPnpmImportersSlicing: + """The importers block is sliced by line, so anything that looks like a top-level key matters.""" + + @staticmethod + def _members(tmp_path: Path, lockfile_text: str) -> frozenset: + lock_file = tmp_path / 'pnpm-lock.yaml' + lock_file.write_text(lockfile_text) + return workspace_resolvers._pnpm_lockfile_member_names(lock_file) + + def test_a_comment_at_column_zero_does_not_end_the_block(self, tmp_path: Path) -> None: + """Truncating here would drop every later member and hand a pnpm project to npm.""" + text = ( + "lockfileVersion: '9.0'\n" + 'importers:\n' + ' .:\n' + ' dependencies: {}\n' + '# a comment written by hand\n' + ' packages/app:\n' + ' dependencies: {}\n' + 'packages: {}\n' + ) + + assert self._members(tmp_path, text) == frozenset({'packages/app'}) + + def test_a_real_top_level_key_still_ends_the_block(self, tmp_path: Path) -> None: + text = ( + "lockfileVersion: '9.0'\n" + 'importers:\n' + ' packages/app:\n' + ' dependencies: {}\n' + 'packages:\n' + ' not-a-member@1.0.0:\n' + ' resolution: {integrity: sha512-x}\n' + ) + + assert self._members(tmp_path, text) == frozenset({'packages/app'}) + + @pytest.mark.parametrize( + ('label', 'text', 'expected'), + [ + ('flow style', "lockfileVersion: '9.0'\nimporters: {.: {}, packages/app: {}}\n", {'packages/app'}), + ( + 'crlf', + "lockfileVersion: '9.0'\r\nimporters:\r\n packages/app:\r\n dependencies: {}\r\n", + {'packages/app'}, + ), + ('dot slash prefix', "lockfileVersion: '9.0'\nimporters:\n ./packages/app: {}\n", {'packages/app'}), + ('no trailing newline', "lockfileVersion: '9.0'\nimporters:\n packages/app: {}", {'packages/app'}), + ('no importers', "lockfileVersion: '9.0'\npackages: {}\n", None), + ('root importer only', "lockfileVersion: '9.0'\nimporters:\n .: {}\n", None), + ], + ) + def test_lockfile_shapes(self, tmp_path: Path, label: str, text: str, expected: Optional[set]) -> None: + got = self._members(tmp_path, text) + assert got == (None if expected is None else frozenset(expected)), label + + def test_a_huge_packages_block_is_never_read(self, tmp_path: Path) -> None: + """Reading the whole lockfile would undo the point of slicing out importers.""" + lock_file = tmp_path / 'pnpm-lock.yaml' + lines = ["lockfileVersion: '9.0'", 'importers:', ' packages/app:', ' dependencies: {}', 'packages:'] + lines += [f' dep{index}@1.0.0: {{resolution: {{integrity: sha512-x}}}}' for index in range(50000)] + lock_file.write_text('\n'.join(lines)) + + consumed = [] + real_open = Path.open + + def counting_open(path: Path, *args: object, **kwargs: object) -> object: + handle = real_open(path, *args, **kwargs) + if path != lock_file: + return handle + + return _counting_lines(handle, consumed) + + Path.open = counting_open + try: + section = workspace_resolvers._read_pnpm_importers_section(lock_file) + finally: + Path.open = real_open + + assert 'packages/app' in section + assert sum(consumed) < lock_file.stat().st_size / 10, ( + f'read {sum(consumed)} of {lock_file.stat().st_size} bytes; the packages block should stop the read' + ) + + +class TestWalkNeverEscapesTheScannedTree: + """Without a .git ancestor the walk must still stop somewhere inside what was scanned.""" + + def test_a_file_scan_root_stands_for_its_directory(self, tmp_path: Path) -> None: + """cycode scan path .../package.json scans that file; the directory is what contains the member.""" + checkout = tmp_path / 'checkout' + member_dir = _write_member(checkout, 'packages/app') + + (tmp_path / 'package.json').write_text('{"name": "stray", "workspaces": ["**"]}') + _write_npm_lockfile(tmp_path, ['checkout/packages/app']) + + scanned_file = member_dir / 'package.json' + + assert find_covering_workspace(str(member_dir), (str(scanned_file),)) is None + + def test_an_unrelated_ancestor_manifest_cannot_cover_a_member(self, tmp_path: Path) -> None: + """A stray manifest above the scanned tree must never suppress a project inside it.""" + checkout = tmp_path / 'checkout' + member_dir = _write_member(checkout, 'packages/app') + + (tmp_path / 'package.json').write_text('{"name": "stray", "workspaces": ["**"]}') + _write_npm_lockfile(tmp_path, ['checkout/packages/app']) + + assert find_covering_workspace(str(member_dir), (str(checkout),)) is None + + def test_a_workspace_root_inside_the_scanned_tree_is_still_found(self, tmp_path: Path) -> None: + """Bounding the walk must not stop it before the real root.""" + checkout = tmp_path / 'checkout' + checkout.mkdir() + (checkout / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + _write_npm_lockfile(checkout, ['packages/app']) + member_dir = _write_member(checkout, 'packages/app') + + assert find_covering_workspace(str(member_dir), (str(checkout),)) is not None + + def test_a_scan_root_that_does_not_contain_the_member_stops_the_walk(self, tmp_path: Path) -> None: + """Known scan roots that are unrelated to this manifest must not license a walk to /.""" + elsewhere = tmp_path / 'elsewhere' + elsewhere.mkdir() + member_dir = _write_member(tmp_path / 'checkout', 'packages/app') + + (tmp_path / 'package.json').write_text('{"name": "stray", "workspaces": ["**"]}') + _write_npm_lockfile(tmp_path, ['checkout/packages/app']) + + assert find_covering_workspace(str(member_dir), (str(elsewhere),)) is None + + def test_the_containment_check_also_treats_a_file_as_its_directory(self, tmp_path: Path) -> None: + """Scanning the root's own package.json means the root lockfile is collected, so the member is covered.""" + (tmp_path / '.git').mkdir() + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + _write_npm_lockfile(tmp_path, ['packages/app']) + member_dir = _write_member(tmp_path, 'packages/app') + + covered = is_covered_workspace_member( + str(member_dir), 'packages/app/package.json', (str(tmp_path / 'package.json'),) + ) + + assert covered is True + + +class TestYarnLockfileMembership: + """Yarn berry names its members; classic yarn does not, and only then do the globs apply.""" + + @staticmethod + def _write_berry_lockfile(root: Path, members: list) -> None: + lines = ['__metadata:', ' version: 8', '', '"root@workspace:.":', ' resolution: "root@workspace:."', ''] + for member in members: + lines += [ + f'"@scope/{Path(member).name}@workspace:{member}":', + f' resolution: "@scope/{Path(member).name}@workspace:{member}"', + '', + ] + (root / 'yarn.lock').write_text('\n'.join(lines)) + + def test_a_member_named_by_a_berry_lockfile_is_covered(self, tmp_path: Path) -> None: + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + self._write_berry_lockfile(tmp_path, ['packages/app']) + member_dir = _write_member(tmp_path, 'packages/app') + + coverage = find_covering_workspace(str(member_dir)) + + assert coverage is not None + assert coverage.package_manager == 'yarn' + + def test_a_member_missing_from_a_stale_berry_lockfile_is_not_covered(self, tmp_path: Path) -> None: + """The globs would claim it, but berry records what it installed and this is not in it.""" + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + self._write_berry_lockfile(tmp_path, ['packages/other']) + member_dir = _write_member(tmp_path, 'packages/app') + + assert find_covering_workspace(str(member_dir)) is None + + def test_a_workspace_range_is_not_mistaken_for_a_member_path(self, tmp_path: Path) -> None: + """Berry writes sibling dependencies as workspace:^ or workspace:*, which are ranges, not paths.""" + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + (tmp_path / 'yarn.lock').write_text( + '__metadata:\n version: 8\n\n' + '"@scope/app@workspace:packages/app":\n' + ' resolution: "@scope/app@workspace:packages/app"\n' + ' dependencies:\n' + ' "@scope/common": "workspace:^"\n\n' + '"@scope/common@workspace:*, @scope/common@workspace:packages/common":\n' + ' resolution: "@scope/common@workspace:packages/common"\n' + ) + + assert workspace_resolvers._yarn_lockfile_member_names(tmp_path / 'yarn.lock') == frozenset( + {'packages/app', 'packages/common'} + ) + + def test_classic_yarn_falls_back_to_the_manifest_globs(self, tmp_path: Path) -> None: + """A v1 lockfile is flat, so the workspaces field is the only remaining source.""" + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + _write_yarn_classic_lockfile(tmp_path) + member_dir = _write_member(tmp_path, 'packages/app') + + coverage = find_covering_workspace(str(member_dir)) + + assert coverage is not None + assert coverage.package_manager == 'yarn' + + def test_the_berry_root_importer_is_not_a_member(self, tmp_path: Path) -> None: + """root@workspace:. and the @workspace:* self-reference must never match a member path.""" + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + (tmp_path / 'yarn.lock').write_text( + '__metadata:\n version: 8\n\n"root@workspace:*, root@workspace:.":\n resolution: "root@workspace:."\n' + ) + member_dir = _write_member(tmp_path, 'packages/app') + + assert find_covering_workspace(str(member_dir)) is None + + +class TestLockfileIsTheAuthority: + """Where a lockfile can name its members, the workspaces globs are not consulted at all.""" + + def test_an_npm_member_is_covered_without_any_workspaces_field(self, tmp_path: Path) -> None: + (tmp_path / 'package.json').write_text('{"name": "root"}') + _write_npm_lockfile(tmp_path, ['packages/app']) + member_dir = _write_member(tmp_path, 'packages/app') + + assert find_covering_workspace(str(member_dir)) is not None + + def test_a_pnpm_member_is_covered_without_pnpm_workspace_yaml(self, tmp_path: Path) -> None: + (tmp_path / 'package.json').write_text('{"name": "root", "private": true}') + _write_pnpm_lockfile(tmp_path, ['packages/app']) + member_dir = _write_member(tmp_path, 'packages/app') + + assert find_covering_workspace(str(member_dir)) is not None + + def test_globs_cannot_rescue_a_member_the_npm_lockfile_omits(self, tmp_path: Path) -> None: + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + _write_npm_lockfile(tmp_path, ['packages/other']) + member_dir = _write_member(tmp_path, 'packages/app') + + assert find_covering_workspace(str(member_dir)) is None + + def test_a_version_1_lockfile_is_not_a_workspace_root(self, tmp_path: Path) -> None: + """Workspaces arrived with npm 7 and lockfileVersion 2, so v1 can never resolve a member.""" + (tmp_path / 'package.json').write_text('{"name": "root", "workspaces": ["packages/*"]}') + (tmp_path / 'package-lock.json').write_text(json.dumps({'lockfileVersion': 1, 'dependencies': {}})) + member_dir = _write_member(tmp_path, 'packages/app') + + assert find_covering_workspace(str(member_dir)) is None + + +class TestMemberResolvers: + """Each resolver owns its lockfile format and what its silence means.""" + + def test_every_resolver_declares_what_it_handles(self) -> None: + for resolver in workspace.MEMBER_RESOLVERS: + assert resolver.package_manager, type(resolver).__name__ + assert resolver.lock_file_names, type(resolver).__name__ + + def test_no_lockfile_is_claimed_by_two_resolvers(self) -> None: + """Precedence is the resolver order, so a name appearing twice would be ambiguous.""" + seen = [name for resolver in workspace.MEMBER_RESOLVERS for name in resolver.lock_file_names] + + assert len(seen) == len(set(seen)), seen + + @pytest.mark.parametrize( + ('package_manager', 'may_use_globs'), + [('npm', False), ('pnpm', False), ('yarn', True), ('bun', True), ('deno', True)], + ) + def test_only_the_blind_formats_may_fall_back_to_globs(self, package_manager: str, may_use_globs: bool) -> None: + """npm and pnpm always name their members, so their silence means "not a workspace".""" + resolver = next(r for r in workspace.MEMBER_RESOLVERS if r.package_manager == package_manager) + + assert resolver.may_use_workspace_globs is may_use_globs + + def test_an_opaque_resolver_never_names_members(self, tmp_path: Path) -> None: + resolver = next(r for r in workspace.MEMBER_RESOLVERS if r.package_manager == 'deno') + lock_file = tmp_path / 'deno.lock' + lock_file.write_text('{"version": "4"}') + + assert resolver.resolve(lock_file) is None + + def test_the_base_class_caches_so_a_root_is_parsed_once(self, tmp_path: Path) -> None: + """resolve() is the template method; subclasses only parse.""" + resolver = next(r for r in workspace.MEMBER_RESOLVERS if r.package_manager == 'npm') + _write_npm_lockfile(tmp_path, ['packages/app']) + lock_file = tmp_path / 'package-lock.json' + + reads = [] + original = workspace_resolvers.read_json_object + + def counting(path: Path) -> object: + reads.append(str(path)) + return original(path) + + workspace_resolvers.read_json_object = counting + try: + first = resolver.resolve(lock_file) + second = resolver.resolve(lock_file) + finally: + workspace_resolvers.read_json_object = original + + assert first == second == frozenset({'packages/app'}) + assert reads.count(str(lock_file)) == 1 diff --git a/tests/cli/files_collector/sca/test_sca_file_collector.py b/tests/cli/files_collector/sca/test_sca_file_collector.py index f645283d..13688f0a 100644 --- a/tests/cli/files_collector/sca/test_sca_file_collector.py +++ b/tests/cli/files_collector/sca/test_sca_file_collector.py @@ -1,3 +1,5 @@ +import json +from pathlib import Path from unittest.mock import MagicMock import click @@ -5,7 +7,11 @@ import typer from cycode.cli.exceptions.custom_exceptions import FileCollectionError -from cycode.cli.files_collector.sca.sca_file_collector import _try_restore_dependencies +from cycode.cli.files_collector.sca.npm import workspace +from cycode.cli.files_collector.sca.sca_file_collector import ( + _add_dependencies_tree_documents, + _try_restore_dependencies, +) from cycode.cli.models import Document @@ -77,3 +83,26 @@ def test_sets_empty_content_when_restore_returns_document_with_none_content(self assert result is not None assert result.content == '' + + +class TestNpmWorkspaceCacheLifetime: + def test_each_scan_starts_with_a_cleared_npm_workspace_cache(self, tmp_path: Path) -> None: + """The cache memoises root lockfiles by path; a later scan must not inherit a previous scan's view.""" + root_manifest = tmp_path / 'package.json' + root_manifest.write_text('{"name": "root", "workspaces": ["packages/*"]}') + lock_file = tmp_path / 'package-lock.json' + lock_file.write_text(json.dumps({'lockfileVersion': 3, 'packages': {'': {}, 'packages/other': {}}})) + + member_dir = tmp_path / 'packages' / 'app' + member_dir.mkdir(parents=True) + manifest = member_dir / 'package.json' + manifest.write_text('{"name": "app"}') + + assert workspace.find_covering_workspace(str(member_dir)) is None + + lock_file.write_text(json.dumps({'lockfileVersion': 3, 'packages': {'': {}, 'packages/app': {}}})) + + ctx = _make_ctx() + _add_dependencies_tree_documents(ctx, [Document(str(manifest), manifest.read_text())]) + + assert workspace.find_covering_workspace(str(member_dir)) is not None diff --git a/tests/cli/files_collector/sca/test_scan_roots_from_context.py b/tests/cli/files_collector/sca/test_scan_roots_from_context.py new file mode 100644 index 00000000..570cc04b --- /dev/null +++ b/tests/cli/files_collector/sca/test_scan_roots_from_context.py @@ -0,0 +1,45 @@ +from pathlib import Path +from unittest.mock import MagicMock + +import pytest +import typer + +from cycode.cli.utils.path_utils import get_path_from_context, get_scan_roots_from_context + + +def _ctx(params: dict) -> typer.Context: + ctx = MagicMock(spec=typer.Context) + ctx.params = params + return ctx + + +class TestGetPathFromContextDelegates: + """Both lookups read the same parameters, so they must not drift apart.""" + + @pytest.mark.parametrize( + ('command', 'params'), + [ + ('scan path', {'paths': [Path('/repo/a')]}), + ('scan repository', {'path': Path('/repo/a')}), + ('report sbom path', {'path': Path('/repo/a')}), + ], + ) + def test_it_returns_the_first_scan_root_as_a_string(self, command: str, params: dict) -> None: + """It is annotated Optional[str] but used to hand back the PosixPath typer produced.""" + result = get_path_from_context(_ctx(params)) + + assert result == str(Path('/repo/a')), command + assert isinstance(result, str), command + + def test_an_empty_paths_list_returns_none(self) -> None: + """Indexing [0] used to raise IndexError here.""" + assert get_path_from_context(_ctx({'paths': []})) is None + + def test_it_agrees_with_the_full_scan_root_list(self) -> None: + params = {'paths': [Path('/repo/a'), Path('/repo/b')]} + + assert get_path_from_context(_ctx(params)) == get_scan_roots_from_context(_ctx(params))[0] + + def test_no_parameters_returns_none(self) -> None: + assert get_path_from_context(_ctx({})) is None + assert get_path_from_context(MagicMock(spec=typer.Context)) is None