Replace availability API with getting list of revisions from folder contents

Author: bartaz

tests/publisher/cve/test_has_cve.py

@@ -1,28 +1,49 @@
 import unittest
-from unittest.mock import patch, MagicMock
+from unittest.mock import patch
 
 from webapp.publisher.cve.cve_helper import CveHelper
+from werkzeug.exceptions import NotFound
 
 
-class HasCvesTest(unittest.TestCase):
+class HasRevisionsWithCvesTest(unittest.TestCase):
 
-    def setUp(self):
-        self.file_metadata = {"download_url": "https://example.com/file.json"}
+    @patch("webapp.publisher.cve.cve_helper.CveHelper._get_cve_file_metadata")
+    def test_returns_revision_numbers(self, mock_get_metadata):
+        mock_get_metadata.return_value = [
+            {"name": "123.yaml"},
+            {"name": "456.yaml"},
+            {"name": "789.yaml"},
+        ]
 
-    @patch("requests.get")
-    def test_has_cve_data(self, mock_get):
-        mock_get.side_effect = [
-            MagicMock(status_code=200, json=lambda: self.file_metadata),
+        result = CveHelper.has_revisions_with_cves("my-snap")
+        self.assertEqual(result, [123, 456, 789])
+
+    @patch("webapp.publisher.cve.cve_helper.CveHelper._get_cve_file_metadata")
+    def test_ignores_non_yaml_files(self, mock_get_metadata):
+        mock_get_metadata.return_value = [
+            {"name": "README.md"},
+            {"name": "123.yaml"},
+            {"name": "abc.yaml"},
+            {"name": "456.yaml"},
+            {"name": "data.txt"},
         ]
 
-        result = CveHelper.has_cve_data("my-snap")
-        self.assertTrue(result)
+        result = CveHelper.has_revisions_with_cves("my-snap")
+        self.assertEqual(result, [123, 456])
 
-    @patch("requests.get")
-    def test_has_cve_data_not_found(self, mock_get):
-        mock_get.side_effect = [
-            MagicMock(status_code=404, json=lambda: {}),
+    @patch("webapp.publisher.cve.cve_helper.CveHelper._get_cve_file_metadata")
+    def test_returns_empty_list_if_no_revision_files(self, mock_get_metadata):
+        mock_get_metadata.return_value = [
+            {"name": "README.md"},
+            {"name": "notes.txt"},
         ]
 
-        result = CveHelper.has_cve_data("my-snap")
-        self.assertFalse(result)
+        result = CveHelper.has_revisions_with_cves("my-snap")
+        self.assertEqual(result, [])
+
+    @patch("webapp.publisher.cve.cve_helper.CveHelper._get_cve_file_metadata")
+    def test_returns_empty_list_on_not_found(self, mock_get_metadata):
+        mock_get_metadata.side_effect = NotFound()
+
+        result = CveHelper.has_revisions_with_cves("my-snap")
+        self.assertEqual(result, [])

tests/publisher/cve/test_has_cve_api.py

@@ -31,8 +31,8 @@ def _set_user_is_canonical(self, value):
 
 class TestModelServiceEndpoints(TestEndpoints):
     @patch(
-        "webapp.publisher.cve.cve_helper.CveHelper.has_cve_data",
-        return_value=True,
+        "webapp.publisher.cve.cve_helper.CveHelper.has_revisions_with_cves",
+        return_value=[123, 321],
     )
     @patch(
         "canonicalwebteam.store_api.dashboard.Dashboard.get_snap_info",
@@ -41,15 +41,15 @@ class TestModelServiceEndpoints(TestEndpoints):
     def test_has_cves_for_canonical_user(self, mock_get_snap_info, mock_get):
         self._set_user_is_canonical(True)
 
-        response = self.client.get("api/test/cves/available")
+        response = self.client.get("api/test/cves")
         data = response.json
 
         self.assertEqual(response.status_code, 200)
         self.assertEqual(data["success"], True)
 
     @patch(
-        "webapp.publisher.cve.cve_helper.CveHelper.has_cve_data",
-        return_value=False,
+        "webapp.publisher.cve.cve_helper.CveHelper.has_revisions_with_cves",
+        return_value=[],
     )
     @patch(
         "canonicalwebteam.store_api.dashboard.Dashboard.get_snap_info",
@@ -58,14 +58,14 @@ def test_has_cves_for_canonical_user(self, mock_get_snap_info, mock_get):
     def test_has_cves_no_data(self, mock_get_snap_info, mock_get):
         self._set_user_is_canonical(True)
 
-        response = self.client.get("api/test/cves/available")
+        response = self.client.get("api/test/cves")
         data = response.json
 
         self.assertEqual(response.status_code, 404)
         self.assertEqual(data["success"], False)
 
     def test_has_cves_for_non_canonical_user(self):
-        response = self.client.get("api/test/cves/available")
+        response = self.client.get("api/test/cves")
         data = response.json
 
         self.assertEqual(response.status_code, 403)

webapp/publisher/cve/cve_helper.py

@@ -1,6 +1,7 @@
 import json
 from os import getenv
 import requests
+import re
 
 from werkzeug.exceptions import NotFound
 
@@ -110,14 +111,24 @@ def _fetch_file_content(snap_name, revision, file_metadata):
             raise NotFound
 
     @staticmethod
-    def has_cve_data(snap_name):
+    def has_revisions_with_cves(snap_name):
         try:
-            CveHelper._get_cve_file_metadata(
-                "snap-cves/{}.json".format(snap_name)
+            contents = CveHelper._get_cve_file_metadata(
+                f"snap-cves/{snap_name}"
             )
-            return True
+
+            # find all revision YAML files in the folder
+            # e.g., 123.yaml, 456.yaml, 789.yaml
+            # and extract the revision numbers
+            revision_files = [
+                int(match.group(1))
+                for item in contents
+                if (match := re.match(r"(\d+)\.yaml$", item["name"]))
+            ]
+
+            return revision_files
         except NotFound:
-            return False
+            return []
 
     @staticmethod
     def get_cve_with_revision(snap_name, revision):

webapp/publisher/cve/cve_views.py

@@ -40,7 +40,7 @@ def can_user_access_cve_data(snap_name):
 
 
 @login_required
-def has_cves(snap_name):
+def get_revisions_with_cves(snap_name):
 
     # Check if the user has access to CVE data for the given snap
     has_access, error_message, status_code = can_user_access_cve_data(
@@ -52,9 +52,11 @@ def has_cves(snap_name):
             status_code,
         )
 
-    snap_has_cves = CveHelper.has_cve_data(snap_name)
-    if snap_has_cves:
-        return flask.jsonify({"success": True})
+    revisions_with_cves = CveHelper.has_revisions_with_cves(snap_name)
+    if len(revisions_with_cves) > 0:
+        return flask.jsonify(
+            {"success": True, "revisions": revisions_with_cves}
+        )
     else:
         return (
             flask.jsonify(

webapp/publisher/snaps/views.py

@@ -286,8 +286,8 @@
 )
 
 publisher_snaps.add_url_rule(
-    "/api/<snap_name>/cves/available",
-    view_func=cve_views.has_cves,
+    "/api/<snap_name>/cves",
+    view_func=cve_views.get_revisions_with_cves,
 )