Add tests for v3 OpenSSL importer pipeline

Signed-off-by: Keshav Priyadarshi <[email protected]>

Author: keshav-space

vulnerabilities/tests/pipelines/v2_importers/test_openssl_importer.py

@@ -14,26 +14,26 @@
 from django.test import TestCase
 
 from vulnerabilities.models import AdvisoryV2
-from vulnerabilities.models import PackageV2
 from vulnerabilities.pipelines.v2_importers.openssl_importer import OpenSSLImporterPipeline
 from vulnerabilities.tests import util_tests
+from vulnerabilities.tests.pipelines import TestLogger
 
 TEST_DATA = Path(__file__).parent.parent.parent / "test_data" / "openssl" / "release_metadata"
 
 
 class TestOpenSSLImporterPipeline(TestCase):
+    def setUp(self):
+        self.logger = TestLogger()
+
     @patch("vulnerabilities.pipelines.v2_importers.openssl_importer.OpenSSLImporterPipeline.clone")
     def test_redhat_advisories_v2(self, mock_clone):
         mock_clone.__name__ = "clone"
         pipeline = OpenSSLImporterPipeline()
         pipeline.advisory_path = TEST_DATA
         pipeline.vcs_response = None
+        pipeline.log = self.logger.write
         pipeline.execute()
 
-        # self.assertEqual(6, AdvisoryV2.objects.count())
-        # self.assertEqual(93, PackageV2.objects.count())
-
         expected_file = TEST_DATA / "openssl_advisoryv2-expected.json"
-        result = [adv.to_advisory_data() for adv in AdvisoryV2.objects.all()]
-        print(result)
-        # util_tests.check_results_against_json(result, expected_file, regen=True)
+        result = [adv.to_advisory_data().to_dict() for adv in AdvisoryV2.objects.all()]
+        util_tests.check_results_against_json(result, expected_file)

vulnerabilities/tests/pipes/test_openssl.py

@@ -6,3 +6,105 @@
 # See https://github.com/aboutcode-org/vulnerablecode for support or download.
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
+
+from datetime import datetime
+from unittest import TestCase
+
+import pytest
+from django.core.exceptions import ValidationError
+from django.utils import timezone
+from packageurl import PackageURL
+from univers.version_range import VersionRange
+
+from vulnerabilities import models
+from vulnerabilities.importer import AdvisoryData
+from vulnerabilities.importer import AffectedPackage
+from vulnerabilities.importer import PackageCommitPatchData
+from vulnerabilities.importer import Reference
+from vulnerabilities.importer import ReferenceV2
+from vulnerabilities.models import AdvisoryAlias
+from vulnerabilities.models import AdvisoryReference
+from vulnerabilities.models import AdvisorySeverity
+from vulnerabilities.models import AdvisoryWeakness
+from vulnerabilities.models import PackageCommitPatch
+from vulnerabilities.pipes.advisory import get_or_create_advisory_aliases
+from vulnerabilities.pipes.advisory import get_or_create_advisory_package_commit_patches
+from vulnerabilities.pipes.advisory import get_or_create_advisory_references
+from vulnerabilities.pipes.advisory import get_or_create_advisory_severities
+from vulnerabilities.pipes.advisory import get_or_create_advisory_weaknesses
+from vulnerabilities.pipes.advisory import get_or_create_aliases
+from vulnerabilities.pipes.advisory import import_advisory
+from vulnerabilities.pipes.openssl import get_commit_patch
+from vulnerabilities.pipes.openssl import get_reference
+from vulnerabilities.pipes.openssl import parse_affected_fixed
+from vulnerabilities.tests.pipelines import TestLogger
+from vulnerabilities.utils import compute_content_id
+
+
+class TestPipeOpenSSL(TestCase):
+    def setUp(self):
+        self.logger = TestLogger()
+
+    def test_vulnerability_pipes_openssl_get_reference(self):
+        refrence_name = "OpenSSL Advisory"
+        tag = "vendor-advisory"
+        refrence_url = "https://www.openssl.org/news/secadv/20221213.txt"
+        result = get_reference(
+            reference_name=refrence_name,
+            tag=tag,
+            reference_url=refrence_url,
+        )
+        expected = ReferenceV2(
+            reference_id=refrence_name,
+            reference_type=AdvisoryReference.ADVISORY,
+            url=refrence_url,
+        )
+
+        self.assertEqual(result, expected)
+
+    def test_vulnerability_pipes_openssl_get_commit_patch(self):
+        url = "https://git.openssl.org/gitweb/?p=openssl.git;a=commitdiff;h=cca1cd9a3447dd067503e4a85ebd1679ee78a48e"
+        result_patch = get_commit_patch(url=url, logger=self.logger.write)
+        expected_vcs = "https://github.com/openssl/openssl/"
+        expected_hash = "cca1cd9a3447dd067503e4a85ebd1679ee78a48e"
+
+        self.assertEqual(result_patch.vcs_url, expected_vcs)
+        self.assertEqual(result_patch.commit_hash, expected_hash)
+
+    def test_vulnerability_pipes_openssl_get_commit_patch_unsupported(self):
+        url = "https://someunsupported.url/commit/93l232slfsll3l23l2"
+        get_commit_patch(url=url, logger=self.logger.write)
+
+        self.assertIn("Unsupported commit url", self.logger.getvalue())
+
+    def test_vulnerability_pipes_openssl_parse_affected_fixed_lessthan(self):
+        affected = {
+            "lessThan": "0.9.7a",
+            "status": "affected",
+            "version": "0.9.7",
+            "versionType": "custom",
+        }
+
+        result_affected, result_fixed = parse_affected_fixed(affected)
+        result_affected = [str(const) for const in result_affected]
+        expected_affected = [">=0.9.7", "<0.9.7a"]
+        expected_fixed = "0.9.7a"
+
+        self.assertCountEqual(result_affected, expected_affected)
+        self.assertEqual(result_fixed, expected_fixed)
+
+    def test_vulnerability_pipes_openssl_parse_affected_fixed_lessthanorequal(self):
+        affected = {
+            "lessThanOrEqual": "3.0.7",
+            "status": "affected",
+            "version": "3.0.0",
+            "versionType": "semver",
+        }
+
+        result_affected, result_fixed = parse_affected_fixed(affected)
+        result_affected = [str(const) for const in result_affected]
+        expected_affected = [">=3.0.0", "<=3.0.7"]
+        expected_fixed = None
+
+        self.assertCountEqual(result_affected, expected_affected)
+        self.assertEqual(result_fixed, expected_fixed)