Model changes

Signed-off-by: Tushar Goel <tushar.goel.dav@gmail.com>

Author: TG1999

vulnerabilities/models.py

@@ -1587,12 +1587,16 @@ class CodeChange(models.Model):
     """
     Abstract base model representing a change in code, either introducing or fixing a vulnerability.
     This includes details about commits, patches, and related metadata.
+
+    We are tracking commits, pulls and downloads as references to the code change. The goal is to
+    keep track and store the actual code patch in the ``patch`` field. When not available the patch
+    will be inferred from these references using improvers.
     """
 
     commits = models.JSONField(
         blank=True,
         default=list,
-        help_text="List of commit identifiers associated with the code change.",
+        help_text="List of commit identifiers using VCS URLs associated with the code change.",
     )
     pulls = models.JSONField(
         blank=True,
@@ -1603,45 +1607,55 @@ class CodeChange(models.Model):
         blank=True, default=list, help_text="List of download URLs for the patched code."
     )
     patch = models.TextField(
-        blank=True, null=True, help_text="The code change in patch format (e.g., git diff)."
-    )
-    notes = models.TextField(
-        blank=True, null=True, help_text="Additional notes or instructions about the code change."
-    )
-    references = models.JSONField(
-        blank=True, default=list, help_text="External references related to this code change."
-    )
-    status_reviewed = models.BooleanField(
-        default=False, help_text="Indicates if the code change has been reviewed."
+        blank=True, null=True, help_text="The code change as a patch in unified diff format."
     )
-    base_version = models.ForeignKey(
+    base_package_version = models.ForeignKey(
         "Package",
         null=True,
         blank=True,
         on_delete=models.SET_NULL,
-        related_name="base_version_codechanges",
-        help_text="The base version of the package to which this code change applies.",
+        related_name="codechanges",
+        help_text="The base package version to which this code change applies.",
     )
-    base_commit = models.CharField(
-        max_length=255,
-        blank=True,
-        null=True,
-        help_text="The commit ID representing the state of the code before applying the fix or change.",
+    notes = models.TextField(
+        blank=True, null=True, help_text="Notes or instructions about this code change."
+    )
+    references = models.JSONField(
+        blank=True, default=list, help_text="URL references related to this code change."
+    )
+    is_reviewed = models.BooleanField(
+        default=False, help_text="Indicates if this code change has been reviewed."
     )
     created_at = models.DateTimeField(
-        auto_now_add=True, help_text="Timestamp indicating when the code change was created."
+        auto_now_add=True, help_text="Timestamp indicating when this code change was created."
     )
     updated_at = models.DateTimeField(
-        auto_now=True, help_text="Timestamp indicating when the code change was last updated."
+        auto_now=True, help_text="Timestamp indicating when this code change was last updated."
     )
 
     class Meta:
         abstract = True
 
 
 class CodeFix(CodeChange):
-    package_vulnerabilities = models.ManyToManyField(
+    """
+    A code fix is a code change that addresses a vulnerability and is associated:
+    - with a specific affected package version
+    - optionally with a specific fixing package version when it is known
+    """
+
+    affected_package_vulnerability = models.ForeignKey(
         "AffectedByPackageRelatedVulnerability",
-        related_name="code_fixes",
-        help_text="The vulnerabilities fixed by this code change.",
+        on_delete=models.CASCADE,
+        related_name="code_fix",
+        help_text="The affected package version to which this code fix applies.",
+    )
+
+    fixed_package_vulnerability = models.ForeignKey(
+        "FixingPackageRelatedVulnerability",
+        null=True,
+        blank=True,
+        on_delete=models.SET_NULL,
+        related_name="code_fix",
+        help_text="The fixing package version with this code fix",
     )

vulnerabilities/pipelines/collect_commits.py

@@ -44,15 +44,25 @@ def collect_and_store_fix_commits(self):
 
         created_fix_count = 0
         progress = LoopProgress(total_iterations=references.count(), logger=self.log)
+
+        Reference
+        AffectedByPackageRelatedVulnerability
+        # FixingPackageRelatedVulnerability
+
+
+        for apv in AffectedByPackageRelatedVulnerability.objects.all():
+            vuln = apv.vulnerability
+            for ref in vuln.references:
+
         for reference in progress.iter(references.paginated(per_page=500)):
             for vulnerability in reference.vulnerabilities.all():
-                vcs_url = normalize_vcs_url(reference.url)
+                vcs_url = normalize_vcs_url(repo_url=reference.url)
 
                 if not vcs_url:
                     continue
 
                 # Skip if already processed
-                if is_reference_already_processed(reference.url, vcs_url):
+                if is_reference_already_processed(reference_url=reference.url, commit_id=vcs_url):
                     self.log(
                         f"Skipping already processed reference: {reference.url} with VCS URL {vcs_url}"
                     )
@@ -97,7 +107,8 @@ def create_codefix_entry(self, vulnerability, package, vcs_url, reference):
                 },
             )
             if created:
-                codefix.vulnerabilities.add(vulnerability)
+                AffectedByPackageRelatedVulnerability.objects.get
+                codefix.package_vulnerabilities.add(vulnerability)
                 codefix.save()
             return codefix
         except Exception as e:
@@ -124,10 +135,13 @@ def create_codefix_entry(self, vulnerability, package, vcs_url, reference):
 )
 
 
+# TODO: This function was borrowed from scancode-toolkit. We need to create a shared library for that.
 def normalize_vcs_url(repo_url, vcs_tool=None):
     """
     Return a normalized vcs_url version control URL given some `repo_url` and an
-    optional `vcs_tool` hint (such as 'git', 'hg', etc.
+    optional `vcs_tool` hint (such as 'git', 'hg', etc.)
+
+    Return None if repo_url is not recognized as a VCS URL.
 
     Handles shortcuts for GitHub, GitHub gist, Bitbucket, or GitLab repositories
     and more using the same approach as npm install:

vulnerabilities/tests/test_collect_commits.py

@@ -0,0 +1,129 @@
+from unittest.mock import patch
+
+from vulnerabilities.models import CodeFix
+from vulnerabilities.pipelines.collect_commits import CollectFixCommitsPipeline
+from vulnerabilities.pipelines.collect_commits import is_reference_already_processed
+from vulnerabilities.pipelines.collect_commits import normalize_vcs_url
+
+
+# --- Mocked Dependencies ---
+class MockVulnerability:
+    def __init__(self, id):
+        self.id = id
+
+
+class MockReference:
+    def __init__(self, url, vulnerabilities):
+        self.url = url
+        self.vulnerabilities = vulnerabilities
+
+
+class MockPackage:
+    def __init__(self, purl):
+        self.purl = purl
+
+
+# --- Tests for Utility Functions ---
+@patch("vulnerabilities.models.CodeFix.objects.filter")
+def test_reference_already_processed_true(mock_filter):
+    mock_filter.return_value.exists.return_value = True
+    result = is_reference_already_processed("http://example.com", "commit123")
+    assert result is True
+    mock_filter.assert_called_once_with(
+        references__contains=["http://example.com"], commits__contains=["commit123"]
+    )
+
+
+@patch("vulnerabilities.models.CodeFix.objects.filter")
+def test_reference_already_processed_false(mock_filter):
+    mock_filter.return_value.exists.return_value = False
+    result = is_reference_already_processed("http://example.com", "commit123")
+    assert result is False
+
+
+# --- Tests for normalize_vcs_url ---
+def test_normalize_plain_url():
+    url = normalize_vcs_url("https://github.com/user/repo.git")
+    assert url == "https://github.com/user/repo.git"
+
+
+def test_normalize_git_ssh_url():
+    url = normalize_vcs_url("git@github.com:user/repo.git")
+    assert url == "https://github.com/user/repo.git"
+
+
+def test_normalize_implicit_github():
+    url = normalize_vcs_url("user/repo")
+    assert url == "https://github.com/user/repo"
+
+
+# --- Tests for CollectFixCommitsPipeline ---
+@patch("vulnerabilities.models.VulnerabilityReference.objects.prefetch_related")
+@patch("vulnerabilities.pipelines.collect_commits.CollectFixCommitsPipeline.get_or_create_package")
+@patch("vulnerabilities.pipelines.collect_commits.is_reference_already_processed")
+@patch("vulnerabilities.pipelines.collect_commits.url2purl")
+def test_collect_and_store_fix_commits(
+    mock_url2purl, mock_is_processed, mock_get_package, mock_prefetch
+):
+    mock_vuln = MockVulnerability(id=1)
+    mock_reference = MockReference(url="http://example.com", vulnerabilities=[mock_vuln])
+    mock_prefetch.return_value.distinct.return_value.paginated.return_value = [mock_reference]
+    mock_url2purl.return_value = "pkg:example/package@1.0.0"
+    mock_is_processed.return_value = False
+    mock_get_package.return_value = MockPackage(purl="pkg:example/package@1.0.0")
+
+    pipeline = CollectFixCommitsPipeline()
+    pipeline.log = lambda msg: None
+    pipeline.collect_and_store_fix_commits()
+
+    mock_is_processed.assert_called_once_with("http://example.com", "pkg:example/package@1.0.0")
+    mock_get_package.assert_called_once_with("pkg:example/package@1.0.0")
+
+
+@patch("vulnerabilities.pipelines.collect_commits.CollectFixCommitsPipeline.get_or_create_package")
+def test_get_or_create_package_success(mock_get_or_create):
+    mock_get_or_create.return_value = (MockPackage(purl="pkg:example/package@1.0.0"), True)
+    pipeline = CollectFixCommitsPipeline()
+    package = pipeline.get_or_create_package("pkg:example/package@1.0.0")
+    assert package.purl == "pkg:example/package@1.0.0"
+
+
+@patch("vulnerabilities.pipelines.collect_commits.CollectFixCommitsPipeline.get_or_create_package")
+def test_get_or_create_package_failure(mock_get_or_create):
+    mock_get_or_create.side_effect = Exception("Error")
+    pipeline = CollectFixCommitsPipeline()
+    logs = []
+    pipeline.log = lambda msg: logs.append(msg)
+    result = pipeline.get_or_create_package("pkg:example/package@1.0.0")
+    assert result is None
+    assert len(logs) == 1
+
+
+@patch("vulnerabilities.models.CodeFix.objects.get_or_create")
+def test_create_codefix_entry_success(mock_get_or_create):
+    mock_get_or_create.return_value = (CodeFix(), True)
+    pipeline = CollectFixCommitsPipeline()
+    result = pipeline.create_codefix_entry(
+        MockVulnerability(1),
+        MockPackage("pkg:example/package@1.0.0"),
+        "http://example.com",
+        "http://reference",
+    )
+    assert result is not None
+    mock_get_or_create.assert_called_once()
+
+
+@patch("vulnerabilities.models.CodeFix.objects.get_or_create")
+def test_create_codefix_entry_failure(mock_get_or_create):
+    mock_get_or_create.side_effect = Exception("Error")
+    pipeline = CollectFixCommitsPipeline()
+    logs = []
+    pipeline.log = lambda msg: logs.append(msg)
+    result = pipeline.create_codefix_entry(
+        MockVulnerability(1),
+        MockPackage("pkg:example/package@1.0.0"),
+        "http://example.com",
+        "http://reference",
+    )
+    assert result is None
+    assert len(logs) == 1