Add support for content_status.json (#4656)

* Add support for content_status.json

.changelog/4656.yml

@@ -0,0 +1,4 @@
+changes:
+- description: Adds support for content_status.json to be used in the content pipeline
+  type: feature
+pr_number: 4656

demisto_sdk/commands/test_content/TestContentClasses.py

@@ -1757,6 +1757,13 @@ def print_test_summary(
                 )
             )
 
+        content_status_update = ContentStatusUpdater(
+            artifacts_folder=self.artifacts_path
+        )
+        content_status_update.update_content_status(
+            successful_tests=succeed_playbooks, failed_tests=list(failed_playbooks)
+        )
+
     @staticmethod
     def print_table(table_name: str, table_data: dict, logging_method: Callable):
         table = prettytable.PrettyTable()
@@ -3244,3 +3251,95 @@ def __str__(self):
 
     def __repr__(self):
         return str(self)
+
+
+class ContentStatusUpdater:
+    def __init__(self, artifacts_folder: Path) -> None:
+        """
+        Initializes the ContentStatusUpdater with the folder and content status filename.
+
+        Args:
+            artifacts_folder (str): The folder where the content status file is located.
+        """
+        self.artifacts_folder = artifacts_folder
+        self.content_status_filename = "content_status.json"
+        self.content_status_path = Path(artifacts_folder, self.content_status_filename)
+        self.content_status: Dict[Any, Any] = {}
+
+    def update_content_status(
+        self, successful_tests: List[str], failed_tests: List[str]
+    ) -> None:
+        """
+        Updates the content status with the provided test results, adding the failed and successful playbooks.
+        content_status.json is a file that keeps track of the failed and successful playbooks in the content repository
+        and is generated by the nightly build.
+
+        Args:
+            successful_tests (List[str]): List of successful playbooks to be added.
+            failed_tests (List[str]): List of failed playbooks to be added.
+        """
+        logging.info(
+            f"Starting update_content_status with {len(failed_tests)} failed tests and "
+            f"{len(successful_tests)} successful tests."
+        )
+
+        self._load_content_status()
+        self._initialize_content_status_keys()
+
+        self._update_playbooks("failed_playbooks", failed_tests)
+        self._update_playbooks("successful_playbooks", successful_tests)
+
+        self._save_content_status()
+
+    def _load_content_status(self) -> None:
+        """
+        Attempts to load the content status from the file. If the file doesn't exist or is invalid, initializes an empty status.
+        """
+        if self.content_status_path.exists():
+            logging.info(f"Content status file exists at {self.content_status_path}")
+            with open(self.content_status_path, "r") as content_file:
+                self.content_status = json.load(content_file)
+                logging.info(f"Loaded content status: {self.content_status}")
+        else:
+            logging.info(
+                f"Initializing empty content status at {self.content_status_path}"
+            )
+            self.content_status = {}
+
+    def _initialize_content_status_keys(self) -> None:
+        """
+        Ensures that the 'failed_playbooks' and 'successful_playbooks' keys are initialized in the content status.
+        """
+        for key in ["failed_playbooks", "successful_playbooks"]:
+            if key not in self.content_status:
+                logging.info(
+                    f"'{key}' key not in content_status. Initializing to empty list."
+                )
+                self.content_status[key] = []
+
+    def _update_playbooks(self, key: str, tests: List[str]) -> None:
+        """
+        Updates the given key in the content status dictionary with the provided list of tests, ensuring no duplicates.
+
+        Args:
+            key (str): The key ('failed_playbooks' or 'successful_playbooks') to update.
+            tests (List[str]): The list of tests (either successful or failed) to add to the content status.
+        """
+        current_playbooks = self.content_status.get(key, [])
+        new_playbooks = sorted(set(tests).difference(current_playbooks))
+        if new_playbooks:
+            logging.info(f"Adding {len(new_playbooks)} new {key}: {new_playbooks}")
+            current_playbooks.extend(new_playbooks)
+        else:
+            logging.info(f"No new {key} to add.")
+
+    def _save_content_status(self) -> None:
+        """
+        Saves the updated content status back to the specified file path.
+        """
+        Path(self.content_status_path.name).mkdir(exist_ok=True)
+        with open(self.content_status_path, "w") as content_file:
+            json.dump(self.content_status, content_file, indent=4)
+            logging.info(
+                f"Saved updated content_status.json to {self.content_status_path}"
+            )

demisto_sdk/commands/test_content/tests/ContentStatusUploader_test.py

@@ -0,0 +1,125 @@
+from pathlib import Path
+from typing import List
+
+from demisto_sdk.commands.common.handlers import DEFAULT_JSON_HANDLER as json
+from demisto_sdk.commands.test_content.TestContentClasses import ContentStatusUpdater
+
+
+def test_update_content_status_initialization(tmp_path: Path) -> None:
+    """
+    Given: content_status.json does not exist in the artifacts folder.
+    When: ContentStatusUpdater is initialized and update_content_status is called with empty lists.
+    Then: It should initialize content_status as empty and create the content_status.json file with correct structure.
+    """
+    # Given
+    artifacts_folder = tmp_path
+    updater = ContentStatusUpdater(artifacts_folder)
+    successful_tests: List[str] = []
+    failed_tests: List[str] = []
+
+    # When
+    updater.update_content_status(successful_tests, failed_tests)
+
+    # Then
+    expected_content_status = {"failed_playbooks": [], "successful_playbooks": []}
+    content_status_file = artifacts_folder / "content_status.json"
+    assert content_status_file.exists()
+    with content_status_file.open("r") as f:
+        content = json.load(f)
+    assert content == expected_content_status
+
+
+def test_update_content_status_with_existing_file(tmp_path: Path) -> None:
+    """
+    Given: content_status.json exists with some playbooks.
+    When: update_content_status is called with new playbooks.
+    Then: It should update the content_status with new playbooks without duplicates.
+    """
+    # Given
+    artifacts_folder = tmp_path
+    content_status_file = artifacts_folder / "content_status.json"
+    initial_content_status = {
+        "failed_playbooks": ["test_fail_1"],
+        "successful_playbooks": ["test_success_1"],
+    }
+    with content_status_file.open("w") as f:
+        json.dump(initial_content_status, f)
+
+    updater = ContentStatusUpdater(artifacts_folder)
+    successful_tests = [
+        "test_success_2",
+        "test_success_1",
+    ]  # "test_success_1" is a duplicate
+    failed_tests = ["test_fail_2"]
+
+    # When
+    updater.update_content_status(successful_tests, failed_tests)
+
+    # Then
+    expected_content_status = {
+        "failed_playbooks": ["test_fail_1", "test_fail_2"],
+        "successful_playbooks": ["test_success_1", "test_success_2"],
+    }
+    with content_status_file.open("r") as f:
+        content = json.load(f)
+    assert content == expected_content_status
+
+
+def test_update_playbooks_no_duplicates(tmp_path: Path) -> None:
+    """
+    Given: content_status.json exists with some playbooks.
+    When: update_content_status is called with playbooks that are already in content_status.
+    Then: It should not add duplicates to the playbooks lists.
+    """
+    # Given
+    artifacts_folder = tmp_path
+    content_status_file = artifacts_folder / "content_status.json"
+    initial_content_status = {
+        "failed_playbooks": ["test_fail"],
+        "successful_playbooks": ["test_success"],
+    }
+    with content_status_file.open("w") as f:
+        json.dump(initial_content_status, f)
+
+    updater = ContentStatusUpdater(artifacts_folder)
+    successful_tests = ["test_success"]  # Duplicate
+    failed_tests = ["test_fail"]  # Duplicate
+
+    # When
+    updater.update_content_status(successful_tests, failed_tests)
+
+    # Then
+    with content_status_file.open("r") as f:
+        content = json.load(f)
+    assert content == initial_content_status  # No changes expected
+
+
+def test_initialize_content_status_keys(tmp_path: Path) -> None:
+    """
+    Given: content_status.json exists without required keys.
+    When: update_content_status is called.
+    Then: It should initialize missing keys in content_status.
+    """
+    # Given
+    artifacts_folder = tmp_path
+    content_status_file = artifacts_folder / "content_status.json"
+    initial_content_status = {"some_other_key": []}
+    with content_status_file.open("w") as f:
+        json.dump(initial_content_status, f)
+
+    updater = ContentStatusUpdater(artifacts_folder)
+    successful_tests: List[str] = []
+    failed_tests: List[str] = []
+
+    # When
+    updater.update_content_status(successful_tests, failed_tests)
+
+    # Then
+    expected_content_status = {
+        "some_other_key": [],
+        "failed_playbooks": [],
+        "successful_playbooks": [],
+    }
+    with content_status_file.open("r") as f:
+        content = json.load(f)
+    assert content == expected_content_status