Create a ScanCode.io pipeline to integrate Grimoire

pyproject.toml

@@ -171,6 +171,7 @@ resolve_dependencies = "scanpipe.pipelines.resolve_dependencies:ResolveDependenc
 scan_codebase = "scanpipe.pipelines.scan_codebase:ScanCodebase"
 scan_for_virus = "scanpipe.pipelines.scan_for_virus:ScanForVirus"
 scan_single_package = "scanpipe.pipelines.scan_single_package:ScanSinglePackage"
+scan_repo_grimoirelab = "scanpipe.pipelines.scan_repo_grimoirelab:ScanRepoGrimoirelab"
 
 [tool.setuptools.packages.find]
 where = ["."]

scancodeio/settings.py

@@ -362,3 +362,14 @@
 FEDERATEDCODE_GIT_SERVICE_TOKEN = env.str("FEDERATEDCODE_GIT_SERVICE_TOKEN", default="")
 FEDERATEDCODE_GIT_SERVICE_NAME = env.str("FEDERATEDCODE_GIT_SERVICE_NAME", default="")
 FEDERATEDCODE_GIT_SERVICE_EMAIL = env.str("FEDERATEDCODE_GIT_SERVICE_EMAIL", default="")
+
+# GrimoireLab integration
+
+GRIMOIRELAB_METRICS_EXECUTABLE = env.str("GRIMOIRELAB_METRICS_EXECUTABLE", default="")
+GRIMOIRELAB_URL = env.str("GRIMOIRELAB_URL", default="")
+GRIMOIRELAB_USERNAME = env.str("GRIMOIRELAB_USERNAME", default="")
+GRIMOIRELAB_PASSWORD = env.str("GRIMOIRELAB_PASSWORD", default="")
+GRIMOIRELAB_OPENSEARCH_URL = env.str("GRIMOIRELAB_OPENSEARCH_URL", default="")
+GRIMOIRELAB_OPENSEARCH_INDEX = env.str("GRIMOIRELAB_OPENSEARCH_INDEX", default="")
+GRIMOIRELAB_OPENSEARCH_USERNAME = env.str("GRIMOIRELAB_OPENSEARCH_USERNAME", default="")
+GRIMOIRELAB_OPENSEARCH_PASSWORD = env.str("GRIMOIRELAB_OPENSEARCH_PASSWORD", default="")

scanpipe/pipelines/scan_repo_grimoirelab.py

@@ -0,0 +1,141 @@
+# 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.
+#
+# Data Generated with ScanCode.io is provided on an "AS IS" BASIS, WITHOUT WARRANTIES
+# OR CONDITIONS OF ANY KIND, either express or implied. No content created from
+# ScanCode.io should be considered or used as legal advice. Consult an Attorney
+# for any legal advice.
+#
+# ScanCode.io is a free software code scanning tool from nexB Inc. and others.
+# Visit https://github.com/aboutcode-org/scancode.io for support and download.
+import json
+import subprocess
+import urllib.parse
+
+from scancodeio.settings import GRIMOIRELAB_METRICS_EXECUTABLE
+from scancodeio.settings import GRIMOIRELAB_OPENSEARCH_INDEX
+from scancodeio.settings import GRIMOIRELAB_OPENSEARCH_PASSWORD
+from scancodeio.settings import GRIMOIRELAB_OPENSEARCH_URL
+from scancodeio.settings import GRIMOIRELAB_OPENSEARCH_USERNAME
+from scancodeio.settings import GRIMOIRELAB_PASSWORD
+from scancodeio.settings import GRIMOIRELAB_URL
+from scancodeio.settings import GRIMOIRELAB_USERNAME
+from scanpipe.pipelines import Pipeline
+from scanpipe.pipes import run_command_safely
+
+
+class ScanRepoGrimoirelab(Pipeline):
+    """Run a GrimoireLab scan to extract repository metrics and health score."""
+
+    results_url = "/project/{slug}/resources/?extra_data=grimoire_data"
+    download_inputs = False
+
+    @classmethod
+    def steps(cls):
+        return (
+            cls.collect_and_store_grimoire_metric,
+            cls.format_metrics_output,
+        )
+
+    def collect_and_store_grimoire_metric(self):
+        """
+        Run the grimoirelab-metrics command against the input source.
+        Save the generated metrics JSON to the project output directory.
+        """
+        if len(self.project.input_sources) != 1:
+            raise ValueError("Expected exactly one input source")
+
+        repo_url = self.project.input_sources[0]["download_url"]
+        if not is_valid_vcs_url(repo_url):
+            raise ValueError(
+                "Invalid input source: the pipeline accepts only a valid repository URL"
+            )
+
+        repo_url = repo_url.replace("git://", "https://")
+        self.metrics_output_path = self.project.get_output_file_path("metrics", "json")
+        command_args = [
+            GRIMOIRELAB_METRICS_EXECUTABLE,
+            repo_url,
+            "--grimoirelab-url",
+            GRIMOIRELAB_URL,
+            "--grimoirelab-user",
+            GRIMOIRELAB_USERNAME,
+            "--grimoirelab-password",
+            GRIMOIRELAB_PASSWORD,
+            "--opensearch-url",
+            GRIMOIRELAB_OPENSEARCH_URL,
+            "--opensearch-index",
+            GRIMOIRELAB_OPENSEARCH_INDEX,
+            "--opensearch-user",
+            GRIMOIRELAB_OPENSEARCH_USERNAME,
+            "--opensearch-password",
+            GRIMOIRELAB_OPENSEARCH_PASSWORD,
+            "--output",
+            str(self.metrics_output_path),
+        ]
+
+        try:
+            run_command_safely(command_args=command_args)
+            self.log("GrimoireLab metrics pipeline completed successfully")
+        except subprocess.SubprocessError:
+            raise RuntimeError("Grimoirelab-metrics pipeline failed")
+        except FileNotFoundError:
+            raise FileNotFoundError(
+                "Grimoirelab-metrics not found. "
+                "Please ensure grimoirelab-metrics is correctly configured."
+            )
+
+    def format_metrics_output(self):
+        """
+        Format the GrimoireLab metrics output by extracting the repository URL,
+        score, and metrics from the generated JSON and overwriting it with a
+        simplified structure.
+        """
+        with open(self.metrics_output_path) as f:
+            data = json.load(f)
+
+        package = list(data["packages"].values())[0]
+
+        repository = package["repository"]
+        score = package["score"]
+        metrics = package["metrics"]
+
+        result = {
+            "repository": repository,
+            "npm_health_score": score,
+            "metrics": metrics,
+        }
+
+        with open(self.metrics_output_path, "w") as f:
+            json.dump(result, f)
+
+
+def is_valid_vcs_url(url):
+    """Determine whether the URL is a valid VCS repository URL."""
+    if not isinstance(url, str) or not url:
+        return False
+
+    if any(char.isspace() for char in url):
+        return False
+
+    forbidden_chars = ["|", ";", "&", "`", "$(", ">", "<", "&&", "||"]
+    if any(char in url for char in forbidden_chars):
+        return False
+
+    parsed = urllib.parse.urlparse(url)
+    valid_schemes = {"https", "http", "git"}
+    if parsed.scheme in valid_schemes and parsed.netloc:
+        return True
+
+    return False

scanpipe/tests/data/grimorielab/expected-metrics.json

@@ -0,0 +1,38 @@
+{
+  "metrics": {
+    "total_commits": 260,
+    "total_contributors": 33,
+    "total_organizations": 16,
+    "pony_factor": 2,
+    "elephant_factor": 1,
+    "recent_organizations": 4,
+    "recent_contributors": 6,
+    "recent_commits": 69,
+    "contributor_growth": 7,
+    "contributor_growth_rate": 0.4666666666666667,
+    "active_branches": 4,
+    "days_since_last_commit": 0,
+    "casual_regular_contributors_rate": 0.65,
+    "returning_contributors": 4,
+    "commits_over_periods_rate": 0.2653846153846154,
+    "coefficient_of_variation": 0.745479473205275,
+    "file_types_code": 143,
+    "file_types_binary": 0,
+    "file_types_other": 2124,
+    "commit_size_added_lines": 38313,
+    "commit_size_removed_lines": 62036,
+    "message_size_total": 35352,
+    "message_size_mean": 135.96923076923076,
+    "message_size_median": 88,
+    "developer_categories_core": 1,
+    "developer_categories_regular": 19,
+    "developer_categories_casual": 13,
+    "commits_per_week": 1.4387351778656126,
+    "commits_per_month": 6.16600790513834,
+    "commits_per_year": 75.0197628458498,
+    "found_file_license": 1,
+    "found_file_adopters": 1
+  },
+  "repository": "https://github.com/chaoss/grimoirelab.git",
+  "npm_health_score": 2.361341589536065e-72
+}

scanpipe/tests/data/grimorielab/metrics.json

@@ -0,0 +1,65 @@
+{
+    "packages": {
+        "SPDXRef-Package-grimoirelab": {
+            "metrics": {
+                "total_commits": 260,
+                "total_contributors": 33,
+                "total_organizations": 16,
+                "pony_factor": 2,
+                "elephant_factor": 1,
+                "recent_organizations": 4,
+                "recent_contributors": 6,
+                "recent_commits": 69,
+                "contributor_growth": 7,
+                "contributor_growth_rate": 0.4666666666666667,
+                "active_branches": 4,
+                "days_since_last_commit": 0,
+                "casual_regular_contributors_rate": 0.65,
+                "returning_contributors": 4,
+                "commits_over_periods_rate": 0.2653846153846154,
+                "coefficient_of_variation": 0.745479473205275,
+                "file_types_code": 143,
+                "file_types_binary": 0,
+                "file_types_other": 2124,
+                "commit_size_added_lines": 38313,
+                "commit_size_removed_lines": 62036,
+                "message_size_total": 35352,
+                "message_size_mean": 135.96923076923076,
+                "message_size_median": 88,
+                "developer_categories_core": 1,
+                "developer_categories_regular": 19,
+                "developer_categories_casual": 13,
+                "commits_per_week": 1.4387351778656126,
+                "commits_per_month": 6.16600790513834,
+                "commits_per_year": 75.0197628458498,
+                "found_file_license": 1,
+                "found_file_adopters": 1
+            },
+            "metadata": {
+                "first_commit": "fc8754a69b50d1bb6a7a30fafc4fe69bda6e3a10",
+                "last_commit": "c51f5deb930371f512394af3d95a1a27383a14c2",
+                "first_commit_date": "2023-01-10T09:54:25+01:00",
+                "last_commit_date": "2026-06-18T09:38:47+02:00"
+            },
+            "repository": "https://github.com/chaoss/grimoirelab.git",
+            "score": 2.361341589536065e-72
+        }
+    },
+    "metadata": {
+        "version": "0.1.0",
+        "started_at": "2026-08-01T02:02:39.612433+00:00",
+        "finished_at": "2026-08-01T02:02:40.794247+00:00",
+        "configuration": {
+            "from_date": "2023-01-01T00:00:00",
+            "to_date": "2026-06-19T00:00:00",
+            "code_file_pattern": "\\.py$|\\.js$",
+            "binary_file_pattern": "\\.exe$|\\.tar$",
+            "pony_threshold": 0.5,
+            "elephant_threshold": 0.5,
+            "dev_categories_thresholds": [
+                0.8,
+                0.95
+            ]
+        }
+    }
+}

scanpipe/tests/pipes/test_scan_repo_grimoirelab.py

@@ -0,0 +1,126 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# http://nexb.com and https://github.com/nexB/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.
+#
+# Data Generated with ScanCode.io is provided on an "AS IS" BASIS, WITHOUT WARRANTIES
+# OR CONDITIONS OF ANY KIND, either express or implied. No content created from
+# ScanCode.io should be considered or used as legal advice. Consult an Attorney
+# for any legal advice.
+#
+# ScanCode.io is a free software code scanning tool from nexB Inc. and others.
+# Visit https://github.com/nexB/scancode.io for support and download.
+
+import json
+import shutil
+import subprocess
+import tempfile
+from pathlib import Path
+from unittest.mock import MagicMock
+from unittest.mock import patch
+
+from django.test import TestCase
+
+from scanpipe.pipelines.scan_repo_grimoirelab import ScanRepoGrimoirelab
+from scanpipe.pipelines.scan_repo_grimoirelab import is_valid_vcs_url
+
+
+class ScanRepoGrimoirelabTest(TestCase):
+    data = Path(__file__).parent.parent / "data" / "grimorielab"
+
+    def setUp(self):
+        mock_run = MagicMock()
+        self.pipeline = ScanRepoGrimoirelab(mock_run)
+        self.pipeline.project = MagicMock()
+        self.pipeline.project.input_sources = [
+            {"download_url": "https://github.com/example/repo.git"}
+        ]
+        self.pipeline.project.get_output_file_path.return_value = "metrics.json"
+        self.pipeline.log = MagicMock()
+
+    @patch("scanpipe.pipelines.scan_repo_grimoirelab.run_command_safely")
+    def test_collect_and_store_grimoire_metric_called_process_error(
+        self, mock_run_command
+    ):
+        """Test handling of a non-zero exit code (CalledProcessError)."""
+        mock_run_command.side_effect = subprocess.CalledProcessError(
+            returncode=1, cmd=["grimoirelab-metrics"]
+        )
+
+        expected_msg = "Grimoirelab-metrics pipeline failed"
+        with self.assertRaisesMessage(RuntimeError, expected_msg):
+            self.pipeline.collect_and_store_grimoire_metric()
+
+    def test_invalid_input_sources(self):
+        """Test handling of invalid number of input sources."""
+        self.pipeline.project.input_sources = []
+
+        expected_msg = "Expected exactly one input source"
+        with self.assertRaisesMessage(Exception, expected_msg):
+            self.pipeline.collect_and_store_grimoire_metric()
+
+        self.pipeline.project.input_sources = [
+            {"download_url": "https://github.com/example/repo1.git"},
+            {"download_url": "https://github.com/example/repo2.git"},
+        ]
+
+        expected_msg = "Expected exactly one input source"
+        with self.assertRaisesMessage(Exception, expected_msg):
+            self.pipeline.collect_and_store_grimoire_metric()
+
+    def test_format_metrics_output(self):
+        """Test formatting the GrimoireLab metrics output using fixture files."""
+        temp_dir = tempfile.mkdtemp()
+
+        input_file = self.data / "metrics.json"
+        expected_file = self.data / "expected-metrics.json"
+
+        temp_metrics_path = Path(temp_dir) / "metrics.json"
+        temp_metrics_path.write_text(input_file.read_text())
+
+        self.pipeline.metrics_output_path = temp_metrics_path
+        self.pipeline.format_metrics_output()
+
+        with open(temp_metrics_path) as f:
+            result = json.load(f)
+
+        with open(expected_file) as f:
+            expected_result = json.load(f)
+
+        self.assertEqual(expected_result, result)
+        shutil.rmtree(temp_dir)
+
+    def test_is_valid_vcs_url(self):
+        """Test VCS repository URL validation."""
+        test_cases = [
+            # Valid URLs
+            ("https://github.com/example/repo.git", True),
+            ("git://github.com/example/repo.git", True),
+            # Invalid URLs
+            ("user@localhost:path/to/repo.git", False),
+            ("hg@bitbucket.org:owner/repo", False),
+            ("https://github.com/repo.git; rm -rf /", False),
+            ("https://github.com/repo.git | ls -la", False),
+            ("https://github.com/repo.git & whoami", False),
+            ("git@github.com:repo.git\nrm -rf /", False),
+            ("https://github.com/ repo.git", False),
+            (" https://github.com/repo.git", False),
+            ("https://", False),
+            ("https:///repo.git", False),
+            ("https://?foo=bar", False),
+            ("", False),
+            (None, False),
+        ]
+
+        for url, expected in test_cases:
+            with self.subTest(url=url):
+                self.assertEqual(is_valid_vcs_url(url), expected)