Add a test for sync_vulnerablecode pipeline

Signed-off-by: ziad hany <ziadhany2016@gmail.com>

Author: ziadhany

fedcode/pipes/utils.py

@@ -105,8 +105,8 @@ def get_vulnerability_path(repo_path: str, vulnerability_id: str) -> str:
     """
     vul_filepath = os.path.join(
         repo_path,
-        f"aboutcode-vulnerabilities-{vulnerability_id[5:7]}",
-        vulnerability_id[10:12],
+        "aboutcode-vulnerabilities",
+        vulnerability_id[5:7],
         vulnerability_id,
         f"{vulnerability_id}.yml",
     )

tests/pipelines/test_sync_vulnerablecode.py

@@ -6,16 +6,26 @@
 # See https://github.com/nexB/federatedcode for support or download.
 # See https://aboutcode.org for more information about AboutCode.org OSS projects.
 #
+from unittest.mock import PropertyMock
+from unittest.mock import call
+from unittest.mock import patch
+
 import pytest
 from django.contrib.auth.models import User
 from fedcode_test_utils import mute_post_save_signal  # NOQA
 
+from aboutcode.hashid import get_core_purl
 from fedcode.models import Note
 from fedcode.models import Package
 from fedcode.models import Repository
 from fedcode.models import Service
 from fedcode.models import Vulnerability
 from fedcode.pipelines.sync_vulnerablecode import SyncVulnerableCode
+from fedcode.pipelines.sync_vulnerablecode import note_handler
+from fedcode.pipelines.sync_vulnerablecode import pkg_handler
+from fedcode.pipelines.sync_vulnerablecode import vul_handler
+
+TEST_REPO_1_PATH = "/home/ziad-hany/PycharmProjects/vul-sample"
 
 
 @pytest.fixture
@@ -30,80 +40,246 @@ def service(db):
     )
 
 
+@pytest.fixture
+def mock_latest_commit_hexsha(monkeypatch):
+    """
+    Fixture to override only repo.head.commit.hexsha while keeping
+    the rest of the git repo behavior intact.
+    """
+
+    def _mock(repo_instance, hexsha: str):
+        """
+        repo_instance: a Repository model instance
+        hexsha: the fake hexsha to return for latest commit
+        """
+        real_repo = repo_instance.git_repo_obj  # use real GitPython repo
+
+        # Patch only the hexsha property
+        type(real_repo.head.commit).hexsha = PropertyMock(return_value=hexsha)
+
+        return real_repo
+
+    return _mock
+
+
 @pytest.fixture
 def repo(db, service, mute_post_save_signal):
     """Simple Git Repository"""
     return Repository.objects.create(
-        url="https://github.com/nexB/fake-repo",
-        path="./review/tests/test_data/test_git_repo_v1",
+        url="https://github.com/ziadhany/vul-sample",
+        path=TEST_REPO_1_PATH,
         admin=service,
     )
 
 
-@pytest.mark.skip(reason="Need a real git repo to test the importer")
-@pytest.mark.django_db
-def test_simple_importer(service, repo, mute_post_save_signal):
-    # just add all packages and vulnerabilities
-    repo.path = "/home/ziad/vul-sample/repo1"
-    importer = SyncVulnerableCode()
-    importer.execute()
+@pytest.fixture
+def vul_changes():
+    return {"create": [], "update": [], "delete": set()}
 
-    assert Note.objects.count() > 1
-    assert Vulnerability.objects.count() > 1
-    assert Package.objects.count() > 1
-    assert repo.last_imported_commit
 
-    note_n = Note.objects.count()
-    vul_n = Vulnerability.objects.count()
-    purl_n = Package.objects.count()
-    last_imported_commit = repo.last_imported_commit
+@pytest.fixture
+def pkg_changes():
+    return {"create": [], "update": [], "delete": set()}
 
-    # Run importer again without add any new data
-    importer = SyncVulnerableCode()
-    importer.execute()
 
-    assert note_n == Note.objects.count()
-    assert vul_n == Vulnerability.objects.count()
-    assert purl_n == Package.objects.count()
-    assert last_imported_commit == repo.last_imported_commit
+@pytest.fixture
+def example_notes():
+    return [
+        {
+            "purl": "pkg:alpine/ansible@2.10.1-r0?arch=aarch64&distroversion=edge&reponame=community",
+            "affected_by_vulnerabilities": [],
+            "fixing_vulnerabilities": ["VCID-r7zs-rzfz-aaap"],
+        },
+        {
+            "purl": "pkg:alpine/ansible@2.10.1-r0?arch=armhf&distroversion=edge&reponame=community",
+            "affected_by_vulnerabilities": ["VCID-r7zs-rzfz-aaap"],
+            "fixing_vulnerabilities": [],
+        },
+    ]
 
-    # Edit last_imported_commit
-    repo.last_imported_commit = "c8de84af0a7c11bf151e96142ce711824648ec41"
-    repo.save()
-    importer = SyncVulnerableCode()
-    importer.execute()
+def test_added_vulnerability(repo, vul_changes):
+    vul_handler("A", repo, None, {"vulnerability_id": "VCID-1234"}, None, vul_changes)
+    assert len(vul_changes["create"]) == 1
+    vuln = vul_changes["create"][0]
+    assert isinstance(vuln, Vulnerability)
+    assert vuln.id == "VCID-1234"
+    assert vuln.repo == repo
 
 
-@pytest.mark.skip(reason="Need a real git repo to test the importer")
-@pytest.mark.django_db
-def test_complex_importer(service, repo, mute_post_save_signal):
-    # repo with 1 commit
-    repo.path = "/home/ziad/vul-sample/repo1"
-    importer = SyncVulnerableCode()
-    importer.execute()
+def test_modified_vulnerability_changed_id(repo, vul_changes):
+    vul_handler(
+        "M",
+        repo,
+        {"vulnerability_id": "VCID-1111"},
+        {"vulnerability_id": "VCID-2222"},
+        None,
+        vul_changes,
+    )
 
-    assert Note.objects.count() > 1
-    assert Vulnerability.objects.count() > 1
-    assert Package.objects.count() > 1
-    assert repo.last_imported_commit
+    assert ("VCID-1111", repo) in vul_changes["delete"]
+    assert ("VCID-2222", repo) in vul_changes["update"]
 
-    note_n = Note.objects.count()
-    vul_n = Vulnerability.objects.count()
-    purl_n = Package.objects.count()
-    last_imported_commit = repo.last_imported_commit
 
-    # Run importer again without add any new data
-    # the same repo with 2 commit ( after pull )
-    repo.path = "/home/ziad/vul-sample/repo2"
-    importer = SyncVulnerableCode()
-    importer.execute()
+def test_modified_vulnerability_same_id(repo, vul_changes):
+    vul_handler(
+        "M",
+        repo,
+        {"vulnerability_id": "VCID-1111"},
+        {"vulnerability_id": "VCID-1111"},
+        None,
+        vul_changes,
+    )
+
+    assert not vul_changes["delete"]
+    assert ("VCID-1111", repo) in vul_changes["update"]
+
+
+def test_deleted_vulnerability(repo, vul_changes):
+    vul_handler("D", repo, {"vulnerability_id": "VCID-3333"}, None, None, vul_changes)
+    assert ("VCID-3333", repo) in vul_changes["delete"]
+
 
-    assert note_n > Note.objects.count()
-    assert vul_n > Vulnerability.objects.count()
-    assert purl_n > Package.objects.count()
+def test_added_packages(service, pkg_changes):
+    yaml_data_b_blob = ["pkg:pypi/django@3.2.5", "pkg:pypi/requests@2.28.1"]
 
-    # Edit last_imported_commit
-    repo.last_imported_commit = "9c3ccee39baef6017d9152367402de9909eadd72"
-    repo.save()
+    pkg_handler("A", service, None, yaml_data_b_blob, None, pkg_changes)
+
+    assert len(pkg_changes["create"]) == 2
+    created = [pkg.purl for pkg in pkg_changes["create"]]
+    assert get_core_purl("pkg:pypi/django@3.2.5") in created
+    assert get_core_purl("pkg:pypi/requests@2.28.1") in created
+
+
+def test_modified_packages(service, pkg_changes):
+    yaml_data_a_blob = ["pkg:pypi/django@3.2.5", "pkg:pypi/requests@2.27.0"]
+    yaml_data_b_blob = ["pkg:pypi/django@4.0.0", "pkg:pypi/requests@2.28.1"]
+
+    pkg_handler("M", service, yaml_data_a_blob, yaml_data_b_blob, None, pkg_changes)
+
+    assert len(pkg_changes["update"]) == 2
+    updates = [(a, b) for (a, b, _) in pkg_changes["update"]]
+    assert (
+        get_core_purl("pkg:pypi/django@3.2.5"),
+        get_core_purl("pkg:pypi/django@4.0.0"),
+    ) in updates
+    assert (
+        get_core_purl("pkg:pypi/requests@2.27.0"),
+        get_core_purl("pkg:pypi/requests@2.28.1"),
+    ) in updates
+
+
+def test_deleted_packages(service, pkg_changes):
+    yaml_data_a_blob = ["pkg:pypi/flask@2.0.0", "pkg:pypi/urllib3@1.26.0"]
+
+    pkg_handler("D", service, yaml_data_a_blob, None, None, pkg_changes)
+
+    assert len(pkg_changes["delete"]) == 2
+    deletes = [(p, svc) for (p, svc) in pkg_changes["delete"]]
+    assert (get_core_purl("pkg:pypi/flask@2.0.0"), service) in deletes
+    assert (get_core_purl("pkg:pypi/urllib3@1.26.0"), service) in deletes
+
+
+def test_note_handler_add(service, example_notes):
+    with patch("fedcode.pipelines.sync_vulnerablecode.bulk_create_notes") as mock_create:
+        note_handler("A", service, None, example_notes, None)
+        assert mock_create.called
+        args, _ = mock_create.call_args
+        notes_to_create = args[0]
+        assert len(notes_to_create) == len(example_notes)
+        pkg = Package.objects.get(purl="pkg:alpine/ansible")
+        expected = [
+            (
+                pkg,
+                {
+                    "affected_by_vulnerabilities": [],
+                    "fixing_vulnerabilities": ["VCID-r7zs-rzfz-aaap"],
+                    "purl": "pkg:alpine/ansible@2.10.1-r0?arch=aarch64&distroversion=edge&reponame=community",
+                },
+            ),
+            (
+                pkg,
+                {
+                    "affected_by_vulnerabilities": ["VCID-r7zs-rzfz-aaap"],
+                    "fixing_vulnerabilities": [],
+                    "purl": "pkg:alpine/ansible@2.10.1-r0?arch=armhf&distroversion=edge&reponame=community",
+                },
+            ),
+        ]
+        assert notes_to_create == expected
+
+
+def test_note_handler_modify(service, example_notes):
+    old_notes = example_notes[:1]  # first note only
+    new_notes = example_notes[1:]  # second note only
+
+    with patch("fedcode.pipelines.sync_vulnerablecode.bulk_update_notes") as mock_update, patch(
+        "fedcode.pipelines.sync_vulnerablecode.bulk_create_notes"
+    ) as mock_create, patch(
+        "fedcode.pipelines.sync_vulnerablecode.bulk_delete_notes"
+    ) as mock_delete:
+
+        note_handler("M", service, old_notes, new_notes, None)
+        pkg = Package.objects.get(purl="pkg:alpine/ansible")
+        assert mock_update.call_args_list == [
+            call(
+                [
+                    (
+                        pkg,
+                        {
+                            "purl": "pkg:alpine/ansible@2.10.1-r0?arch=aarch64&distroversion=edge&reponame=community",
+                            "affected_by_vulnerabilities": [],
+                            "fixing_vulnerabilities": ["VCID-r7zs-rzfz-aaap"],
+                        },
+                        {
+                            "purl": "pkg:alpine/ansible@2.10.1-r0?arch=armhf&distroversion=edge&reponame=community",
+                            "affected_by_vulnerabilities": ["VCID-r7zs-rzfz-aaap"],
+                            "fixing_vulnerabilities": [],
+                        },
+                    )
+                ]
+            )
+        ]
+
+        assert mock_delete.called == False
+        assert mock_create.called == False
+        assert mock_update.called
+
+
+def test_note_handler_delete(service, example_notes):
+    with patch("fedcode.pipelines.sync_vulnerablecode.bulk_delete_notes") as mock_delete:
+        note_handler("D", service, example_notes, None, None)
+        assert mock_delete.called
+        args, _ = mock_delete.call_args
+        notes_to_delete = args[0]
+        assert len(notes_to_delete) == len(example_notes)
+
+
+@pytest.mark.skip(reason="A real Git repository is needed to test the pipelines.")
+@pytest.mark.django_db
+def test_simple_importer(service, repo, mock_latest_commit_hexsha, mute_post_save_signal):
+    repo.path = TEST_REPO_1_PATH
     importer = SyncVulnerableCode()
-    importer.execute()
+    commits = [
+        # (commit, last_imported_commit, note_count, vuln_count, pkg_count)
+        (
+            "f7cd453ff1ef29a539723c44f82bcc582dac13b1",
+            None,
+            28,
+            6,
+            3,
+        ),  # vuln_count is 7, but one of them is duplicated.
+        ("d2115ebdc64341f5b9169e42c9edde9002898b3b", "f7cd453ff1ef29a539723c44f82bcc582dac13b1", 45, 6, 3),
+        ("d2115ebdc64341f5b9169e42c9edde9002898b3b", None, 0, 0, 0),
+        ("275987c1d758155e782b7fe0539d7089d4e618ea", None, 0, 0, 0),
+    ]
+
+    for commit, last_imported_commit, note_count, vuln_count, pkg_count in commits:
+        repo.last_imported_commit = last_imported_commit
+        repo.save()
+
+        mock_latest_commit_hexsha(repo, hexsha=commit)
+        importer.execute()
+
+        assert Note.objects.count() == note_count
+        assert Vulnerability.objects.count() == vuln_count
+        assert Package.objects.count() == pkg_count

tests/test_utils.py

@@ -13,6 +13,7 @@
 from fedcode.activitypub import AP_CONTEXT
 from fedcode.activitypub import Activity
 from fedcode.activitypub import create_activity_obj
+from fedcode.pipes.utils import get_vulnerability_path
 from fedcode.utils import check_purl_actor
 from fedcode.utils import full_resolve
 from fedcode.utils import full_reverse
@@ -83,3 +84,25 @@ def test_full_resolve():
 
 def test_check_purl_actor():
     assert check_purl_actor("pkg:maven/org.apache.logging")
+
+
+def test_get_vulnerability_path(tmp_path):
+    repo_path = tmp_path
+    vulnerability_id = "VCID-1n1d-h5qn-nyau"
+
+    vuln_dir = repo_path / "aboutcode-vulnerabilities" / vulnerability_id[5:7] / vulnerability_id
+    vuln_dir.mkdir(parents=True, exist_ok=True)
+
+    vuln_file = vuln_dir / f"{vulnerability_id}.yml"
+    vuln_file.write_text("id: VCID-1n1d-h5qn-nyau\n")
+
+    result = get_vulnerability_path(str(repo_path), vulnerability_id)
+    assert result == str(vuln_file)
+
+
+def test_get_vulnerability_path_not_found(tmp_path):
+    repo_path = tmp_path
+    vulnerability_id = "VCID-1n1d-h5qn-nyau"
+
+    with pytest.raises(FileNotFoundError):
+        get_vulnerability_path(str(repo_path), vulnerability_id)