Use snapshot if defined

example/.env.example

@@ -1,4 +1,5 @@
 COVERITY_USERNAME = 'yourusername'
 COVERITY_PASSWORD = 'yourpassword'
 COVERITY_STREAM = 'yourstream'
+COVERITY_SNAPSHOT = ''
 

example/conf.py

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

mlx/coverity.py

@@ -46,7 +46,7 @@ def initialize_environment(self, app):
     \\makeatother"""
 
         self.stream = app.config.coverity_credentials["stream"]
-
+        self.snaphsot = app.config.coverity_credentials["snapshot"]
         # Login to Coverity and obtain stream information
         try:
             self.input_credentials(app.config.coverity_credentials)
@@ -61,6 +61,10 @@ def initialize_environment(self, app):
             report_info("Verify the given stream name... ", True)
             self.coverity_service.validate_stream(self.stream)
             report_info("done")
+            if self.snaphsot:
+                report_info("Verify the given snapshot ID and obtain all enabled checkers... ", True)
+                self.coverity_service.validate_snapshot(self.snaphsot)
+                report_info("done")
             # Get all column keys
             report_info("obtaining all column keys... ", True)
             self.coverity_service.retrieve_column_keys()
@@ -100,7 +104,11 @@ def process_coverity_nodes(self, app, doctree, fromdocname):
             # Get items from server
             try:
                 defects = self.get_filtered_defects(node)
-                node.perform_replacement(defects, self, app, fromdocname)
+                if defects["totalRows"] == -1:
+                    error_message = "There are no defects with the specified filters"
+                    report_warning(error_message, fromdocname, lineno=node["line"])
+                else:
+                    node.perform_replacement(defects, self, app, fromdocname)
             except (URLError, AttributeError, Exception) as err:  # pylint: disable=broad-except
                 error_message = f"failed to process coverity-list with {err!r}"
                 report_warning(error_message, fromdocname, lineno=node["line"])
@@ -142,7 +150,7 @@ def get_filtered_defects(self, node):
         column_names = set(node["col"])
         if "chart_attribute" in node and node["chart_attribute"].upper() in node.column_map:
             column_names.add(node["chart_attribute"])
-        defects = self.coverity_service.get_defects(self.stream, node["filters"], column_names)
+        defects = self.coverity_service.get_defects(self.stream, self.snaphsot, node["filters"], column_names)
         report_info("%d received" % (defects["totalRows"]))
         report_info("building defects table and/or chart... ", True)
         return defects

mlx/coverity_services.py

@@ -125,6 +125,16 @@ def validate_stream(self, stream):
         url = f"{self.api_endpoint}/streams/{stream}"
         self._request(url)
 
+    def validate_snapshot(self, snapshot):
+        """Validate snapshot by retrieving the specified snapshot.
+        When the request fails, the snapshot does not exist or the user does not have acces to it.
+
+        Args:
+            snapshot (str): The snapshot ID
+        """
+        url = f"{self.api_endpoint}/snapshots/{snapshot}"
+        self._request(url)
+
     def retrieve_issues(self, filters):
         """Retrieve issues from the server (Coverity Connect).
 
@@ -236,14 +246,16 @@ def assemble_query_filter(self, column_name, filter_values, matcher_type):
             "matchers": matchers
         }
 
-    def get_defects(self, stream, filters, column_names):
-        """Gets a list of defects for given stream, filters and column names.
+    def get_defects(self, stream, snapshot, filters, column_names):
+        """Gets a list of defects for given stream, snapshot ID, filters and column names.
+        If the snapshot is empty, the last snapshot is taken.
         If a column name does not match the name of the `columns` property, the column can not be obtained because
         it need the correct corresponding column key.
         Column key `cid` is always obtained to use later in other functions.
 
         Args:
             stream (str): Name of the stream to query
+            snapshot (str): The snapshot ID; If empty the last snapshot is taken.
             filters (dict): Dictionary with attribute names as keys and CSV lists of attribute values to query as values
             column_names (list[str]): The column names
 
@@ -291,16 +303,15 @@ def get_defects(self, stream, filters, column_names):
         if (filter := filters["component"]) and (filter_values := self.handle_component_filter(filter)):
             query_filters.append(self.assemble_query_filter("Component", filter_values, "nameMatcher"))
 
+        scope = "last()"
+        if snapshot:
+            scope = snapshot
         data = {
             "filters": query_filters,
             "columns": list(self.column_keys(column_names)),
             "snapshotScope": {
                 "show": {
-                    "scope": "last()",
-                    "includeOutdatedSnapshots": False
-                },
-                "compareTo": {
-                    "scope": "last()",
+                    "scope": scope,
                     "includeOutdatedSnapshots": False
                 }
             }