Merge branch 'snapshots' into logging

example/conf.py

@@ -313,7 +313,7 @@
     "username": config("COVERITY_USERNAME"),
     "password": config("COVERITY_PASSWORD"),
     "stream": config("COVERITY_STREAM"),
-    "snapshot": config("COVERITY_SNAPSHOT")
+    "snapshot": config("COVERITY_SNAPSHOT"),
 }
 
 TRACEABILITY_ITEM_ID_REGEX = r"([A-Z_]+-[A-Z0-9_]+)"

mlx/coverity/coverity.py

@@ -67,8 +67,10 @@ def initialize_environment(self, app):
             LOGGER.info("done")
             if self.snapshot:
                 LOGGER.info("Verify the given snapshot ID and obtain all enabled checkers... ")
-                self.coverity_service.validate_snapshot(self.snapshot)
+                self.snapshot = self.coverity_service.validate_snapshot(self.snapshot)
                 LOGGER.info("done")
+            else:
+                self.snapshot = "last()"
             # Get all column keys
             LOGGER.info("obtaining all column keys... ")
             self.coverity_service.retrieve_column_keys()

mlx/coverity/coverity_services.py

@@ -53,7 +53,6 @@ def __init__(self, hostname):
         self._api_endpoint = f"https://{hostname}/api/{self.version}"
         self._checkers = []
         self._columns = {}
-        self.valid_snapshot = False
 
     @property
     def base_url(self):
@@ -137,11 +136,13 @@ def validate_snapshot(self, snapshot):
         url = f"{self.api_endpoint}/snapshots/{snapshot}"
         response = self.session.get(url)
         if response.ok:
-            self.valid_snapshot = True
             LOGGER.info(f"Snapshot ID {snapshot} is valid")
+            valid_snapshot = snapshot
         else:
             LOGGER.warning(f"No snapshot found for ID {snapshot}; Continue with using the latest snapshot.")
-            self.valid_snapshot = False
+            valid_snapshot = "last()"
+
+        return valid_snapshot
 
     def retrieve_issues(self, filters):
         """Retrieve issues from the server (Coverity Connect).
@@ -312,14 +313,12 @@ def get_defects(self, stream, filters, column_names, snapshot):
         if (filter := filters["component"]) and (filter_values := self.handle_component_filter(filter)):
             query_filters.append(self.assemble_query_filter("Component", filter_values, "nameMatcher"))
 
-        scope = snapshot if snapshot and self.valid_snapshot else "last()"
-
         data = {
             "filters": query_filters,
             "columns": list(self.column_keys(column_names)),
             "snapshotScope": {
                 "show": {
-                    "scope": scope,
+                    "scope": snapshot,
                     "includeOutdatedSnapshots": False
                 }
             }

tests/test_coverity.py

@@ -142,7 +142,8 @@ def test_get_defects(self, filters, column_names, request_data):
         """Check get defects with different filters. Check if the response of `get_defects` is the same as expected.
         The data is obtained from the filters.py file.
         Due to the usage of set in `get_defects` (column_keys), the function `ordered` is used to compare the returned
-        data of the request where order does not matter."""
+        data of the request where order does not matter.
+        """
         with open(f"{TEST_FOLDER}/columns_keys.json", "r") as content:
             column_keys = json.loads(content.read())
         self.fake_checkers = {
@@ -167,7 +168,7 @@ def test_get_defects(self, filters, column_names, request_data):
             coverity_service.retrieve_column_keys()
             # Get defects
             with patch.object(CoverityDefectService, "retrieve_issues") as mock_method:
-                coverity_service.get_defects(self.fake_stream, filters, column_names, "")
+                coverity_service.get_defects(self.fake_stream, filters, column_names, "last()")
                 data = mock_method.call_args[0][0]
                 mock_method.assert_called_once()
                 assert ordered(data) == ordered(request_data)