feat: VCIO v3 advisories migration (#534)

Signed-off-by: tdruez <tdruez@aboutcode.org>

Author: tdruez

Makefile

@@ -27,9 +27,10 @@ shell:
 	# Open a bash session in a standalone container (no stack required)
 	docker run -it $(IMAGE_NAME) bash
 
+# make test              - full suite
+# make test k=<pattern>  - filter by name, e.g. make test k=test_name
 test:
-	@echo "-> Run the test suite"
-	${MANAGE} test --noinput --parallel auto
+	${MANAGE} test --noinput --parallel auto $(if $(k),-k $(k),)
 
 migrations:
 	@echo "-> Creates new database migrations"

component_catalog/api.py

@@ -436,8 +436,8 @@ class ComponentFilterSet(DataspacedAPIFilterSet):
         field_name="affected_by_vulnerabilities",
     )
     affected_by = django_filters.CharFilter(
-        field_name="affected_by_vulnerabilities__vulnerability_id",
-        label="Affected by (vulnerability_id)",
+        field_name="affected_by_vulnerabilities__advisory_id",
+        label="Affected by (advisory_id)",
     )
 
     class Meta:
@@ -630,7 +630,8 @@ class PackageSerializer(
         read_only=True,
         many=True,
         fields=[
-            "vulnerability_id",
+            "advisory_uid",
+            "advisory_id",
             "api_url",
             "uuid",
         ],
@@ -809,8 +810,8 @@ class PackageAPIFilterSet(DataspacedAPIFilterSet):
         field_name="affected_by_vulnerabilities",
     )
     affected_by = django_filters.CharFilter(
-        field_name="affected_by_vulnerabilities__vulnerability_id",
-        label="Affected by (vulnerability_id)",
+        field_name="affected_by_vulnerabilities__advisory_id",
+        label="Affected by (advisory_id)",
     )
     risk_score = ScoreRangeFilter(score_ranges=RISK_SCORE_RANGES)
 

component_catalog/filters.py

@@ -119,7 +119,7 @@ class ComponentFilterSet(DataspacedFilterSet):
         widget=DropDownRightWidget(link_content='<i class="fas fa-bug"></i>'),
     )
     affected_by = django_filters.CharFilter(
-        field_name="affected_by_vulnerabilities__vulnerability_id",
+        field_name="affected_by_vulnerabilities__advisory_id",
         label=_("Affected by"),
     )
 
@@ -267,7 +267,7 @@ class PackageFilterSet(DataspacedFilterSet):
         widget=DropDownRightWidget(link_content='<i class="fas fa-bug"></i>'),
     )
     affected_by = django_filters.CharFilter(
-        field_name="affected_by_vulnerabilities__vulnerability_id",
+        field_name="affected_by_vulnerabilities__advisory_id",
         label=_("Affected by"),
     )
     affected_by_last_modified_date = django_filters.DateRangeFilter(

component_catalog/license_expression_dje.py

@@ -59,7 +59,7 @@ def get_dataspace_licensing(dataspace, license_keys=None):
         # Bypass cache if license_keys is provided
         return fetch_licensing_for_dataspace(dataspace, license_keys)
 
-    cache_key = str({dataspace.name})
+    cache_key = str(dataspace.uuid)
     # First look in the cache for an existing Licensing for this Dataspace
     licensing = licensing_cache.get(cache_key)
 

component_catalog/templates/component_catalog/tabs/tab_vulnerabilities.html

@@ -51,11 +51,11 @@
           <strong>
             {% if vulnerability.resource_url %}
               <a href="{{ vulnerability.resource_url }}" target="_blank">
-                {{ vulnerability.vulnerability_id }}
+                {{ vulnerability.advisory_id }}
                 <i class="fa-solid fa-up-right-from-square mini"></i>
               </a>
             {% else %}
-              {{ vulnerability.vulnerability_id }}
+              {{ vulnerability.advisory_id }}
             {% endif %}
           </strong>
           <div class="mt-2">

component_catalog/tests/test_api.py

@@ -1350,17 +1350,17 @@ def test_api_package_endpoint_vulnerabilities_features(self):
         results = response.data["results"]
         self.assertEqual("9.0", results[0]["risk_score"])
         self.assertEqual(
-            vulnerability1.vulnerability_id,
-            results[0]["affected_by_vulnerabilities"][0]["vulnerability_id"],
+            vulnerability1.advisory_id,
+            results[0]["affected_by_vulnerabilities"][0]["advisory_id"],
         )
 
-        data = {"affected_by": vulnerability1.vulnerability_id}
+        data = {"affected_by": vulnerability1.advisory_id}
         response = self.client.get(self.package_list_url, data)
         self.assertEqual(1, response.data["count"])
         self.assertContains(response, self.package1_detail_url)
         self.assertNotContains(response, self.package2_detail_url)
 
-        data = {"affected_by": vulnerability2.vulnerability_id}
+        data = {"affected_by": vulnerability2.advisory_id}
         response = self.client.get(self.package_list_url, data)
         self.assertEqual(0, response.data["count"])
         self.assertNotContains(response, self.package1_detail_url)

component_catalog/tests/test_filters.py

@@ -406,6 +406,6 @@ def test_package_filterset_affected_by_filter(self):
         self.assertIn(package1, filterset.qs)
         self.assertIn(package2, filterset.qs)
 
-        data = {"affected_by": vulnerability1.vulnerability_id}
+        data = {"affected_by": vulnerability1.advisory_id}
         filterset = PackageFilterSet(dataspace=self.dataspace, data=data)
         self.assertQuerySetEqual(filterset.qs, [package1])

component_catalog/tests/test_license_expression_dje.py

@@ -299,7 +299,7 @@ def test_fetch_licensing_for_dataspace(self):
 
     def test_get_dataspace_licensing(self):
         licensing_cache = caches["licensing"]
-        cache_key = str({self.dataspace.name})
+        cache_key = str(self.dataspace.uuid)
         self.assertFalse(licensing_cache.has_key(cache_key))
 
         licensing = get_dataspace_licensing(self.dataspace)

component_catalog/tests/test_views.py

@@ -1055,7 +1055,7 @@ def test_component_details_view_tab_vulnerabilities(self):
         )
         self.assertContains(response, expected)
         self.assertContains(response, 'id="tab_vulnerabilities"')
-        self.assertContains(response, vulnerability1.vcid)
+        self.assertContains(response, vulnerability1.advisory_id)
 
     def test_component_catalog_component_create_ajax_view(self):
         component_create_ajax_url = reverse("component_catalog:component_add_ajax")
@@ -3021,7 +3021,7 @@ def test_package_details_view_tab_vulnerabilities(self):
         )
         self.assertContains(response, expected)
         self.assertContains(response, 'id="tab_vulnerabilities"')
-        self.assertContains(response, self.vulnerability1.vcid)
+        self.assertContains(response, self.vulnerability1.advisory_id)
 
     def test_vulnerablecode_get_plain_purls(self):
         purls = get_plain_purls(packages=[])
@@ -3057,65 +3057,34 @@ def test_vulnerablecode_get_vulnerable_purls(self):
         with mock.patch(
             "dejacode_toolkit.vulnerablecode.VulnerableCode.bulk_search_by_purl"
         ) as bulk_search:
-            bulk_search.return_value = []
+            bulk_search.return_value = {"count": 0, "results": []}
             vulnerable_purls = vulnerablecode.get_vulnerable_purls(packages=[self.package1])
             self.assertEqual([], vulnerable_purls)
 
-            bulk_search.return_value = ["pkg:pypi/django@2.1"]
+            bulk_search.return_value = {"count": 1, "results": ["pkg:pypi/django@2.1"]}
             vulnerable_purls = vulnerablecode.get_vulnerable_purls(packages=[self.package1])
             self.assertEqual(["pkg:pypi/django@2.1"], vulnerable_purls)
 
-    def test_vulnerablecode_get_vulnerable_cpes(self):
-        vulnerablecode = VulnerableCode(self.dataspace)
-        vulnerable_cpes = vulnerablecode.get_vulnerable_cpes(components=[])
-        self.assertEqual([], vulnerable_cpes)
-
-        components = [self.component1, self.component2]
-        vulnerable_cpes = vulnerablecode.get_vulnerable_cpes(components=components)
-        self.assertEqual([], vulnerable_cpes)
-
-        self.component1.cpe = "cpe:2.3:a:djangoproject:django:0.95:*:*:*:*:*:*:*"
-        self.component1.save()
-
-        with mock.patch(
-            "dejacode_toolkit.vulnerablecode.VulnerableCode.bulk_search_by_cpes"
-        ) as bulk_search:
-            bulk_search.return_value = [
-                {
-                    "vulnerability_id": "VCID-188m-1bke-aaae",
-                    "summary": "The administrative interface in django.contrib.admin ",
-                    "references": [
-                        {"reference_id": ""},
-                    ],
-                }
-            ]
-            vulnerable_cpes = vulnerablecode.get_vulnerable_cpes(components=components)
-            self.assertEqual([], vulnerable_cpes)
-
-            bulk_search.return_value[0]["references"] = [{"reference_id": self.component1.cpe}]
-            vulnerable_cpes = vulnerablecode.get_vulnerable_cpes(components=components)
-            self.assertEqual([self.component1.cpe], vulnerable_cpes)
-
-    @mock.patch("dejacode_toolkit.vulnerablecode.VulnerableCode.request_get")
-    def test_vulnerablecode_get_vulnerabilities_cache(self, mock_request_get):
+    @mock.patch("dejacode_toolkit.vulnerablecode.VulnerableCode.bulk_search_by_purl")
+    def test_vulnerablecode_get_vulnerabilities_cache(self, mock_bulk_search):
         vulnerablecode = VulnerableCode(self.dataspace)
 
         self.package1.set_package_url("pkg:pypi/django@2.1")
         self.package1.save()
 
-        mock_request_get.return_value = {
+        mock_bulk_search.return_value = {
             "count": 1,
             "results": True,
         }
 
         results = vulnerablecode.get_vulnerabilities_by_purl(self.package1.package_url)
-        self.assertEqual(1, mock_request_get.call_count)
+        self.assertEqual(1, mock_bulk_search.call_count)
         self.assertTrue(results)
 
         results = vulnerablecode.get_vulnerabilities_by_purl(self.package1.package_url)
-        # request.get was only called once since the results are returned from the cached
+        # bulk_search_by_purl was only called once since the results are returned from the cache
         # on the second call of `get_vulnerabilities_by_purl`.
-        self.assertEqual(1, mock_request_get.call_count)
+        self.assertEqual(1, mock_bulk_search.call_count)
         self.assertTrue(results)
 
     def test_send_scan_notification(self):

dejacode/settings.py

@@ -725,6 +725,7 @@ def get_fake_redis_connection(config, use_strict_redis):
 VULNERABLECODE_USER_AGENT = env.str("VULNERABLECODE_USER_AGENT", default="VCIO_API_AGENT")
 CREATE_DEPENDENCIES_DEFAULT = env.bool("CREATE_DEPENDENCIES_DEFAULT", default=True)
 
+
 if IS_TESTS:
     # Silent the django-axes logging during tests
     LOGGING["loggers"].update({"axes": {"handlers": ["null"]}})