diff --git a/docs/built-in-pipelines.rst b/docs/built-in-pipelines.rst index 13f8ffda71..67c42d26f4 100644 --- a/docs/built-in-pipelines.rst +++ b/docs/built-in-pipelines.rst @@ -282,3 +282,36 @@ Fetch Scores (addon) .. autoclass:: scanpipe.pipelines.fetch_scores.FetchScores() :members: :member-order: bysource + + +.. _pipeline_npm_health: + +NPM Health (addon) +------------------ + +The ``npm_health`` pipeline analyzes one exact versioned npm ``Project.purl``. +It retrieves npm registry metadata, records repository and package tarball +locations, normalizes health signals, computes a weighted score from 0 to 100, +and stores a reusable snapshot in ``Project.extra_data["npm_health"]``. + +.. autoclass:: scanpipe.pipelines.npm_health.NpmHealth() + :members: + :member-order: bysource + + +NPM Health configuration +~~~~~~~~~~~~~~~~~~~~~~~~ + +Fresh snapshots are reused for 90 days by default. Override the age with +``npm_health_cache_max_age_days`` in project pipeline settings. + +Optional GrimoireLab or other project-health metrics can be connected through +``npm_health_metrics_command``. The command is parsed to an argument list and +executed without a shell. It may use ``{purl}``, ``{repository_url}``, +``{tarball_url}``, and ``{output}`` placeholders. The collector must write JSON +metrics to ``{output}``, either directly or under a top-level ``metrics`` key. +Metric values may be fractions from 0 to 1 or percentages from 0 to 100. + +The optional ``npm_health_metric_weights`` mapping customizes scoring weights. +Only metrics present in the current analysis participate in the denominator. +The pipeline also writes a timestamped ``npm-health-*.json`` project output. diff --git a/pyproject.toml b/pyproject.toml index 350c59a0b7..9dbe3bfe47 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -158,6 +158,7 @@ collect_symbols_pygments = "scanpipe.pipelines.collect_symbols_pygments:CollectS collect_symbols_tree_sitter = "scanpipe.pipelines.collect_symbols_tree_sitter:CollectSymbolsTreeSitter" enrich_with_purldb = "scanpipe.pipelines.enrich_with_purldb:EnrichWithPurlDB" fetch_scores = "scanpipe.pipelines.fetch_scores:FetchScores" +npm_health = "scanpipe.pipelines.npm_health:NpmHealth" find_vulnerabilities = "scanpipe.pipelines.find_vulnerabilities:FindVulnerabilities" inspect_elf_binaries = "scanpipe.pipelines.inspect_elf_binaries:InspectELFBinaries" inspect_packages = "scanpipe.pipelines.inspect_packages:InspectPackages" diff --git a/scanpipe/pipelines/npm_health.py b/scanpipe/pipelines/npm_health.py new file mode 100644 index 0000000000..94efa25067 --- /dev/null +++ b/scanpipe/pipelines/npm_health.py @@ -0,0 +1,124 @@ +# SPDX-License-Identifier: Apache-2.0 +# +# http://nexb.com and https://github.com/aboutcode-org/scancode.io +# The ScanCode.io software is licensed under the Apache License version 2.0. +# Data generated with ScanCode.io is provided as-is without warranties. +# ScanCode is a trademark of nexB Inc. +# +# You may not use this software except in compliance with the License. +# You may obtain a copy of the License at: http://apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software distributed +# under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR +# CONDITIONS OF ANY KIND, either express or implied. See the License for the +# specific language governing permissions and limitations under the License. + +"""Collect and score npm package health information.""" + +from scanpipe.pipelines import Pipeline +from scanpipe.pipes import npm_health + + +class NpmHealth(Pipeline): + """Collect reusable npm package health metrics for one project PURL.""" + + download_inputs = False + is_addon = True + results_url = "/project/{slug}/" + + + def validate_project_purl(self): + """Validate and parse the project's versioned npm PURL.""" + self.package = npm_health.validate_npm_package_url(self.project.purl) + + + def load_cached_snapshot(self): + """Reuse a fresh project snapshot when one is already available.""" + self.cached_snapshot = None + snapshot = npm_health.get_cached_snapshot(self.project) + max_age = int( + self.env.get( + "npm_health_cache_max_age_days", + npm_health.DEFAULT_CACHE_MAX_AGE_DAYS, + ) + ) + if snapshot and not npm_health.is_stale(snapshot, max_age_days=max_age): + self.cached_snapshot = snapshot + self.snapshot = snapshot + self.log("Using fresh cached npm-health analysis.") + + + def fetch_package_metadata(self): + """Fetch exact npm registry metadata unless a fresh cache is used.""" + if self.cached_snapshot: + return + self.metadata = npm_health.fetch_registry_metadata(self.package) + + + def collect_package_metrics(self): + """Collect built-in registry signals and optional external metrics.""" + if self.cached_snapshot: + return + + baseline = npm_health.collect_registry_metrics(self.metadata) + external = {} + command_template = self.env.get("npm_health_metrics_command") + if command_template: + output = self.project.tmp_path / "npm-health-external-metrics.json" + external = npm_health.collect_external_metrics( + command_template=command_template, + purl=self.project.purl, + metadata=self.metadata, + output=output, + cwd=self.project.tmp_path, + ) + self.metrics = npm_health.merge_metrics(baseline, external) + + + def compute_package_health_score(self): + """Compute the normalized package health score.""" + if self.cached_snapshot: + return + weights = self.env.get("npm_health_metric_weights") + self.score = npm_health.compute_health_score(self.metrics, weights=weights) + + + def build_result_snapshot(self): + """Build the reusable snapshot stored on the project.""" + if self.cached_snapshot: + return + self.snapshot = npm_health.build_snapshot( + purl=self.project.purl, + metadata=self.metadata, + metrics=self.metrics, + score=self.score, + ) + + + def persist_results(self): + """Persist fresh results to extra_data and a portable JSON output.""" + if self.cached_snapshot: + return + + npm_health.cache_snapshot(self.project, self.snapshot) + output = npm_health.write_snapshot(self.project, self.snapshot) + self.project.add_info( + model="npm_health", + description=( + f"npm-health score: {self.snapshot['score']} " + f"({self.snapshot['classification']})" + ), + details={"output": output.name}, + ) + + + @classmethod + def steps(cls): + return ( + cls.validate_project_purl, + cls.load_cached_snapshot, + cls.fetch_package_metadata, + cls.collect_package_metrics, + cls.compute_package_health_score, + cls.build_result_snapshot, + cls.persist_results, + ) diff --git a/scanpipe/pipes/npm_health.py b/scanpipe/pipes/npm_health.py new file mode 100644 index 0000000000..059c297d13 --- /dev/null +++ b/scanpipe/pipes/npm_health.py @@ -0,0 +1,442 @@ +# SPDX-License-Identifier: Apache-2.0 +# +# http://nexb.com and https://github.com/aboutcode-org/scancode.io +# The ScanCode.io software is licensed under the Apache License version 2.0. +# Data generated with ScanCode.io is provided as-is without warranties. +# ScanCode is a trademark of nexB Inc. +# +# You may not use this software except in compliance with the License. +# You may obtain a copy of the License at: http://apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software distributed +# under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR +# CONDITIONS OF ANY KIND, either express or implied. See the License for the +# specific language governing permissions and limitations under the License. + +"""Utilities for the npm-health ScanCode.io pipeline.""" + +import json +import shlex +import subprocess +from datetime import UTC +from datetime import datetime +from datetime import timedelta +from pathlib import Path +from urllib.parse import quote +from urllib.parse import urlparse +from urllib.parse import urlunparse + +import requests +from packageurl import PackageURL + + +NPM_HEALTH_EXTRA_DATA_KEY = "npm_health" +DEFAULT_CACHE_MAX_AGE_DAYS = 90 +DEFAULT_REQUEST_TIMEOUT = 30 +DEFAULT_COMMAND_TIMEOUT = 900 + +DEFAULT_METRIC_WEIGHTS = { + "activity": 1.0, + "community": 1.0, + "maintenance": 1.0, + "documentation": 0.75, + "security": 1.0, + "metadata_completeness": 0.5, + "maintainer_presence": 0.5, + "dependency_simplicity": 0.25, +} + + +class NpmHealthError(Exception): + """Base npm-health error.""" + + +class NpmHealthPayloadError(NpmHealthError): + """Invalid PURL, registry data, or metrics payload.""" + + +class NpmHealthCommandError(NpmHealthError): + """External metrics collector failure.""" + + + +def parse_package_url(value): + """Return a PackageURL parsed from ``value``.""" + if not value or not isinstance(value, str): + raise NpmHealthPayloadError("A project PURL is required.") + try: + return PackageURL.from_string(value) + except ValueError as error: + raise NpmHealthPayloadError(f"Invalid package URL: {value}") from error + + + +def validate_npm_package_url(value): + """Return a versioned npm PackageURL or raise a descriptive error.""" + package = parse_package_url(value) + if package.type != "npm": + raise NpmHealthPayloadError( + f"npm-health requires an npm PURL, not {package.type!r}." + ) + if not package.name: + raise NpmHealthPayloadError("npm-health requires a package name.") + if not package.version: + raise NpmHealthPayloadError("npm-health requires a package version.") + return package + + + +def get_package_name(package): + """Return the npm registry name for a PackageURL.""" + if not package.namespace: + return package.name + namespace = package.namespace + if not namespace.startswith("@"): + namespace = f"@{namespace}" + return f"{namespace}/{package.name}" + + + +def get_registry_metadata_url(package): + """Return the npm registry URL for one exact package version.""" + name = quote(get_package_name(package), safe="@") + version = quote(package.version, safe="") + return f"https://registry.npmjs.org/{name}/{version}" + + + +def normalize_repository_url(repository): + """Return a normalized HTTP(S) repository URL.""" + if isinstance(repository, dict): + repository = repository.get("url") + if not repository or not isinstance(repository, str): + return "" + + value = repository.strip() + if value.startswith("git+"): + value = value[4:] + if value.startswith("git@github.com:"): + value = "https://github.com/" + value.removeprefix("git@github.com:") + if value.startswith("git://github.com/"): + value = "https://github.com/" + value.removeprefix("git://github.com/") + + parsed = urlparse(value) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + return "" + path = parsed.path.removesuffix(".git").rstrip("/") + return urlunparse((parsed.scheme, parsed.netloc, path, "", "", "")) + + + +def get_repository_url(metadata): + """Return a normalized repository URL from npm metadata.""" + return normalize_repository_url(metadata.get("repository")) + + + +def get_tarball_url(metadata): + """Return the npm distribution tarball URL.""" + dist = metadata.get("dist") or {} + value = dist.get("tarball") + return value if isinstance(value, str) else "" + + + +def get_homepage_url(metadata): + """Return the package homepage URL when available.""" + value = metadata.get("homepage") + return value if isinstance(value, str) else "" + + + +def get_license(metadata): + """Return a compact license value from npm metadata.""" + value = metadata.get("license") + if isinstance(value, str): + return value + if isinstance(value, dict) and isinstance(value.get("type"), str): + return value["type"] + return "" + + + +def get_maintainer_count(metadata): + """Return the number of maintainers declared by npm.""" + maintainers = metadata.get("maintainers") or [] + return len(maintainers) if isinstance(maintainers, list) else 0 + + + +def get_dependency_count(metadata): + """Return the number of runtime dependencies in npm metadata.""" + dependencies = metadata.get("dependencies") or {} + return len(dependencies) if isinstance(dependencies, dict) else 0 + + + +def fetch_registry_metadata( + package, + session=requests, + timeout=DEFAULT_REQUEST_TIMEOUT, +): + """Fetch and return npm registry metadata for ``package``.""" + response = session.get(get_registry_metadata_url(package), timeout=timeout) + response.raise_for_status() + data = response.json() + if not isinstance(data, dict): + raise NpmHealthPayloadError("The npm registry returned a non-object payload.") + return data + + + +def clamp(value, minimum=0.0, maximum=1.0): + """Clamp a numeric value to an inclusive range.""" + return max(minimum, min(maximum, value)) + + + +def normalize_metric_value(value): + """Normalize bool, 0..1, or percentage metric values to 0..1.""" + if isinstance(value, bool): + return float(value) + if not isinstance(value, int | float): + raise NpmHealthPayloadError(f"Unsupported metric value: {value!r}") + normalized = float(value) + if normalized > 1.0: + normalized /= 100.0 + return clamp(normalized) + + + +def normalize_metrics(payload): + """Return normalized metrics from a direct or nested collector payload.""" + if not isinstance(payload, dict): + raise NpmHealthPayloadError("Metrics payload must be a JSON object.") + values = payload.get("metrics", payload) + if not isinstance(values, dict): + raise NpmHealthPayloadError("Metrics must be a JSON object.") + return { + name: normalize_metric_value(value) + for name, value in values.items() + if isinstance(name, str) + } + + + +def collect_registry_metrics(metadata): + """Return baseline health signals available from npm registry metadata.""" + completeness = sum( + ( + bool(get_repository_url(metadata)), + bool(get_homepage_url(metadata)), + bool(get_license(metadata)), + ) + ) / 3 + return { + "metadata_completeness": completeness, + "maintainer_presence": clamp(get_maintainer_count(metadata) / 3), + "dependency_simplicity": 1.0 - clamp(get_dependency_count(metadata) / 50), + } + + + +def merge_metrics(*metric_sets): + """Merge normalized metric mappings from left to right.""" + merged = {} + for metrics in metric_sets: + if metrics: + merged.update(normalize_metrics(metrics)) + return merged + + + +def normalize_weights(weights=None): + """Return positive numeric scoring weights.""" + weights = weights or DEFAULT_METRIC_WEIGHTS + return { + name: float(value) + for name, value in weights.items() + if isinstance(name, str) and isinstance(value, int | float) and value > 0 + } + + + +def compute_health_score(metrics, weights=None): + """Return a weighted package health score from 0 to 100.""" + metrics = normalize_metrics(metrics) + weights = normalize_weights(weights) + weighted = [ + (value, weights[name]) + for name, value in metrics.items() + if name in weights + ] + denominator = sum(weight for _, weight in weighted) + if not denominator: + return 0.0 + numerator = sum(value * weight for value, weight in weighted) + return round((numerator / denominator) * 100, 2) + + + +def classify_health_score(score): + """Return a qualitative classification for a numeric health score.""" + if score >= 80: + return "excellent" + if score >= 60: + return "good" + if score >= 40: + return "needs-attention" + return "high-risk" + + + +def parse_timestamp(value): + """Return an aware UTC datetime parsed from an ISO timestamp.""" + if not value or not isinstance(value, str): + return None + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=UTC) + return parsed.astimezone(UTC) + + + +def is_stale(snapshot, max_age_days=DEFAULT_CACHE_MAX_AGE_DAYS, now=None): + """Return True when a cached npm-health snapshot is missing or too old.""" + if not isinstance(snapshot, dict): + return True + collected_at = parse_timestamp(snapshot.get("collected_at")) + if not collected_at: + return True + now = now or datetime.now(UTC) + return now - collected_at > timedelta(days=max_age_days) + + + +def build_collection_targets(metadata): + """Return source locations useful to external health collectors.""" + return { + "repository_url": get_repository_url(metadata), + "tarball_url": get_tarball_url(metadata), + "homepage_url": get_homepage_url(metadata), + } + + + +def build_command_context(purl, metadata, output): + """Return safe substitutions for an external collector command.""" + targets = build_collection_targets(metadata) + return { + "purl": purl, + "repository_url": targets["repository_url"], + "tarball_url": targets["tarball_url"], + "output": str(output), + } + + + +def render_metrics_command(command_template, context): + """Render a command template to an argument list without a shell.""" + if not command_template or not isinstance(command_template, str): + raise NpmHealthCommandError("npm_health_metrics_command is not configured.") + try: + rendered = command_template.format_map(context) + except KeyError as error: + raise NpmHealthCommandError( + f"Unknown npm-health command placeholder: {error.args[0]}" + ) from error + args = shlex.split(rendered) + if not args: + raise NpmHealthCommandError("The rendered metrics command is empty.") + return args + + + +def run_metrics_command(args, cwd=None, timeout=DEFAULT_COMMAND_TIMEOUT): + """Run an external metrics collector and return its completed process.""" + completed = subprocess.run( # noqa: S603 + args, + cwd=cwd, + check=False, + capture_output=True, + text=True, + timeout=timeout, + ) + if completed.returncode: + details = (completed.stderr or completed.stdout or "").strip() + raise NpmHealthCommandError( + f"External metrics collector failed ({completed.returncode}): {details}" + ) + return completed + + + +def load_metrics_json(location): + """Load and normalize metrics from a JSON output file.""" + location = Path(location) + if not location.is_file(): + raise NpmHealthCommandError( + f"External metrics output was not created: {location}" + ) + try: + payload = json.loads(location.read_text(encoding="utf-8")) + except json.JSONDecodeError as error: + message = "External metrics output is invalid JSON." + raise NpmHealthPayloadError(message) from error + return normalize_metrics(payload) + + + +def collect_external_metrics( + command_template, + purl, + metadata, + output, + cwd=None, +): + """Run a configured external collector and return normalized metrics.""" + context = build_command_context(purl, metadata, output) + args = render_metrics_command(command_template, context) + run_metrics_command(args=args, cwd=cwd) + return load_metrics_json(output) + + + +def build_snapshot(purl, metadata, metrics, score, collected_at=None): + """Return the persisted npm-health result structure.""" + return { + "purl": purl, + "collected_at": collected_at or datetime.now(UTC).isoformat(), + "score": score, + "classification": classify_health_score(score), + "metrics": normalize_metrics(metrics), + "sources": build_collection_targets(metadata), + } + + + +def get_cached_snapshot(project): + """Return the cached npm-health snapshot from project.extra_data.""" + data = project.extra_data or {} + snapshot = data.get(NPM_HEALTH_EXTRA_DATA_KEY) + return snapshot if isinstance(snapshot, dict) else None + + + +def cache_snapshot(project, snapshot): + """Persist one npm-health snapshot in Project.extra_data.""" + project.update_extra_data({NPM_HEALTH_EXTRA_DATA_KEY: snapshot}) + return snapshot + + + +def write_snapshot(project, snapshot): + """Write one npm-health JSON result into the project output directory.""" + output = project.get_output_file_path("npm-health", "json") + output.write_text( + json.dumps(snapshot, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + return output diff --git a/scanpipe/tests/pipes/test_npm_health.py b/scanpipe/tests/pipes/test_npm_health.py new file mode 100644 index 0000000000..63766dec9a --- /dev/null +++ b/scanpipe/tests/pipes/test_npm_health.py @@ -0,0 +1,220 @@ +# SPDX-License-Identifier: Apache-2.0 +# +# http://nexb.com and https://github.com/aboutcode-org/scancode.io +# The ScanCode.io software is licensed under the Apache License version 2.0. +# Data generated with ScanCode.io is provided as-is without warranties. +# ScanCode is a trademark of nexB Inc. +# +# You may not use this software except in compliance with the License. +# You may obtain a copy of the License at: http://apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software distributed +# under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR +# CONDITIONS OF ANY KIND, either express or implied. See the License for the +# specific language governing permissions and limitations under the License. + +"""Tests for npm-health helper utilities.""" + +from datetime import UTC +from datetime import datetime +from pathlib import Path +from tempfile import TemporaryDirectory +from unittest import mock + +from django.test import SimpleTestCase + +from scanpipe.pipes import npm_health + + +class NpmHealthPipesTest(SimpleTestCase): + + def test_validate_npm_package_url(self): + package = npm_health.validate_npm_package_url("pkg:npm/lodash@4.17.21") + self.assertEqual("npm", package.type) + self.assertEqual("lodash", package.name) + self.assertEqual("4.17.21", package.version) + + + def test_parse_package_url_rejects_empty_value(self): + with self.assertRaises(npm_health.NpmHealthPayloadError): + npm_health.parse_package_url("") + + + def test_validate_npm_package_url_rejects_other_types(self): + with self.assertRaises(npm_health.NpmHealthPayloadError): + npm_health.validate_npm_package_url("pkg:pypi/django@5.2") + + + def test_validate_npm_package_url_requires_version(self): + with self.assertRaises(npm_health.NpmHealthPayloadError): + npm_health.validate_npm_package_url("pkg:npm/lodash") + + + def test_get_package_name_supports_scopes(self): + package = npm_health.validate_npm_package_url( + "pkg:npm/%40babel/core@7.28.0" + ) + self.assertEqual("@babel/core", npm_health.get_package_name(package)) + + + def test_get_registry_metadata_url_for_scoped_package(self): + package = npm_health.validate_npm_package_url( + "pkg:npm/%40babel/core@7.28.0" + ) + self.assertEqual( + "https://registry.npmjs.org/@babel%2Fcore/7.28.0", + npm_health.get_registry_metadata_url(package), + ) + + + def test_normalize_repository_url_dict(self): + repository = {"type": "git", "url": "git+https://github.com/a/b.git"} + self.assertEqual( + "https://github.com/a/b", + npm_health.normalize_repository_url(repository), + ) + + + def test_normalize_repository_url_git_transports(self): + self.assertEqual( + "https://github.com/a/b", + npm_health.normalize_repository_url("git://github.com/a/b.git"), + ) + self.assertEqual( + "https://github.com/a/b", + npm_health.normalize_repository_url("git@github.com:a/b.git"), + ) + + + def test_build_collection_targets(self): + metadata = { + "repository": {"url": "https://github.com/a/b.git"}, + "homepage": "https://example.com", + "dist": {"tarball": "https://registry.example/a.tgz"}, + } + self.assertEqual( + { + "repository_url": "https://github.com/a/b", + "tarball_url": "https://registry.example/a.tgz", + "homepage_url": "https://example.com", + }, + npm_health.build_collection_targets(metadata), + ) + + + def test_normalize_metric_value_fraction(self): + self.assertEqual(0.75, npm_health.normalize_metric_value(0.75)) + + + def test_normalize_metric_value_percentage(self): + self.assertEqual(0.75, npm_health.normalize_metric_value(75)) + self.assertEqual(1.0, npm_health.normalize_metric_value(150)) + + + def test_normalize_metrics_nested_payload(self): + self.assertEqual( + {"activity": 0.8, "security": 1.0}, + npm_health.normalize_metrics( + {"metrics": {"activity": 80, "security": True}} + ), + ) + + + def test_collect_registry_metrics(self): + metadata = { + "repository": {"url": "https://github.com/a/b.git"}, + "homepage": "https://example.com", + "license": "MIT", + "maintainers": [{"name": "a"}, {"name": "b"}, {"name": "c"}], + "dependencies": {"one": "1", "two": "2"}, + } + metrics = npm_health.collect_registry_metrics(metadata) + self.assertEqual(1.0, metrics["metadata_completeness"]) + self.assertEqual(1.0, metrics["maintainer_presence"]) + self.assertGreater(metrics["dependency_simplicity"], 0.9) + + + def test_merge_metrics_external_values_override_baseline(self): + merged = npm_health.merge_metrics( + {"activity": 0.2, "security": 0.5}, + {"activity": 90}, + ) + self.assertEqual({"activity": 0.9, "security": 0.5}, merged) + + + def test_normalize_weights_ignores_non_positive_values(self): + self.assertEqual( + {"activity": 2.0}, + npm_health.normalize_weights({"activity": 2, "security": 0}), + ) + + + def test_compute_health_score(self): + score = npm_health.compute_health_score( + {"activity": 1.0, "security": 0.5}, + {"activity": 1.0, "security": 1.0}, + ) + self.assertEqual(75.0, score) + + + def test_classify_health_score(self): + self.assertEqual("excellent", npm_health.classify_health_score(90)) + self.assertEqual("good", npm_health.classify_health_score(70)) + self.assertEqual("needs-attention", npm_health.classify_health_score(50)) + self.assertEqual("high-risk", npm_health.classify_health_score(20)) + + + def test_is_stale_keeps_fresh_snapshot(self): + now = datetime(2026, 8, 17, tzinfo=UTC) + snapshot = {"collected_at": "2026-08-16T00:00:00+00:00"} + self.assertFalse(npm_health.is_stale(snapshot, max_age_days=90, now=now)) + + + def test_is_stale_expires_old_snapshot(self): + now = datetime(2026, 8, 17, tzinfo=UTC) + snapshot = {"collected_at": "2026-01-01T00:00:00+00:00"} + self.assertTrue(npm_health.is_stale(snapshot, max_age_days=90, now=now)) + + + def test_render_metrics_command(self): + context = { + "purl": "pkg:npm/lodash@4.17.21", + "repository_url": "https://github.com/lodash/lodash", + "tarball_url": "https://example.com/lodash.tgz", + "output": "/tmp/result.json", + } + args = npm_health.render_metrics_command( + "collector --repo {repository_url} --output {output}", + context, + ) + self.assertEqual( + [ + "collector", + "--repo", + "https://github.com/lodash/lodash", + "--output", + "/tmp/result.json", + ], + args, + ) + + + def test_load_metrics_json_rejects_invalid_json(self): + with TemporaryDirectory() as temp_dir: + location = Path(temp_dir) / "metrics.json" + location.write_text("{", encoding="utf-8") + with self.assertRaises(npm_health.NpmHealthPayloadError): + npm_health.load_metrics_json(location) + + + def test_get_cached_snapshot(self): + project = mock.Mock(extra_data={"npm_health": {"score": 88}}) + self.assertEqual({"score": 88}, npm_health.get_cached_snapshot(project)) + + + def test_cache_snapshot(self): + project = mock.Mock() + snapshot = {"purl": "pkg:npm/lodash@4.17.21", "score": 88} + self.assertEqual(snapshot, npm_health.cache_snapshot(project, snapshot)) + project.update_extra_data.assert_called_once_with( + {npm_health.NPM_HEALTH_EXTRA_DATA_KEY: snapshot} + ) diff --git a/scanpipe/tests/test_npm_health_pipeline.py b/scanpipe/tests/test_npm_health_pipeline.py new file mode 100644 index 0000000000..65f8ac94a5 --- /dev/null +++ b/scanpipe/tests/test_npm_health_pipeline.py @@ -0,0 +1,40 @@ +# SPDX-License-Identifier: Apache-2.0 +# +# http://nexb.com and https://github.com/aboutcode-org/scancode.io +# The ScanCode.io software is licensed under the Apache License version 2.0. +# Data generated with ScanCode.io is provided as-is without warranties. +# ScanCode is a trademark of nexB Inc. +# +# You may not use this software except in compliance with the License. +# You may obtain a copy of the License at: http://apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software distributed +# under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR +# CONDITIONS OF ANY KIND, either express or implied. See the License for the +# specific language governing permissions and limitations under the License. + +"""Tests for the npm-health pipeline definition.""" + +from django.test import SimpleTestCase + +from scanpipe.pipelines.npm_health import NpmHealth + + +class NpmHealthPipelineTest(SimpleTestCase): + def test_pipeline_flags_and_results_url(self): + self.assertFalse(NpmHealth.download_inputs) + self.assertTrue(NpmHealth.is_addon) + self.assertEqual("/project/{slug}/", NpmHealth.results_url) + + def test_pipeline_steps(self): + self.assertEqual( + [ + "validate_project_purl", + "load_cached_snapshot", + "fetch_package_metadata", + "collect_package_metrics", + "compute_package_health_score", + "build_result_snapshot", + "persist_results", + ], + [step.__name__ for step in NpmHealth.steps()], + )