diff --git a/.changelog/4656.yml b/.changelog/4656.yml new file mode 100644 index 00000000000..ea0ef1a26f4 --- /dev/null +++ b/.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 diff --git a/demisto_sdk/commands/test_content/TestContentClasses.py b/demisto_sdk/commands/test_content/TestContentClasses.py index cbf10faaf92..f7175123415 100644 --- a/demisto_sdk/commands/test_content/TestContentClasses.py +++ b/demisto_sdk/commands/test_content/TestContentClasses.py @@ -3276,7 +3276,7 @@ def update_content_status( successful_tests (List[str]): List of successful playbooks to be added. failed_tests (List[str]): List of failed playbooks to be added. """ - logging.warning( + logging.info( f"Starting update_content_status with {len(failed_tests)} failed tests and " f"{len(successful_tests)} successful tests." ) @@ -3284,11 +3284,9 @@ def update_content_status( self._load_content_status() self._initialize_content_status_keys() - # Update playbooks with new failed and successful tests self._update_playbooks("failed_playbooks", failed_tests) self._update_playbooks("successful_playbooks", successful_tests) - # Save the updated content status self._save_content_status() def _load_content_status(self) -> None: @@ -3300,14 +3298,14 @@ def _load_content_status(self) -> None: try: with open(self.content_status_path, "r") as content_file: self.content_status = json.load(content_file) - logging.warning(f"Loaded content status: {self.content_status}") + logging.info(f"Loaded content status: {self.content_status}") except json.JSONDecodeError as e: logging.error( f"JSON decode error while loading content_status.json: {e}" ) self.content_status = {} else: - logging.warning( + logging.info( f"Initializing empty content status at {self.content_status_path}" ) self.content_status = {} @@ -3318,7 +3316,7 @@ def _initialize_content_status_keys(self) -> None: """ for key in ["failed_playbooks", "successful_playbooks"]: if key not in self.content_status: - logging.warning( + logging.info( f"'{key}' key not in content_status. Initializing to empty list." ) self.content_status[key] = [] @@ -3334,10 +3332,10 @@ def _update_playbooks(self, key: str, tests: List[str]) -> None: current_playbooks = self.content_status.get(key, []) new_playbooks = [test for test in tests if test not in current_playbooks] if new_playbooks: - logging.warning(f"Adding {len(new_playbooks)} new {key}: {new_playbooks}") + logging.info(f"Adding {len(new_playbooks)} new {key}: {new_playbooks}") current_playbooks.extend(new_playbooks) else: - logging.warning(f"No new {key} to add.") + logging.info(f"No new {key} to add.") def _save_content_status(self) -> None: """ @@ -3346,6 +3344,6 @@ def _save_content_status(self) -> None: os.makedirs(os.path.dirname(self.content_status_path), exist_ok=True) with open(self.content_status_path, "w") as content_file: json.dump(self.content_status, content_file, indent=4) - logging.warning( + logging.info( f"Saved updated content_status.json to {self.content_status_path}" ) diff --git a/demisto_sdk/commands/test_content/tests/ContentStatusUploader_test.py b/demisto_sdk/commands/test_content/tests/ContentStatusUploader_test.py new file mode 100644 index 00000000000..5b476162e72 --- /dev/null +++ b/demisto_sdk/commands/test_content/tests/ContentStatusUploader_test.py @@ -0,0 +1,153 @@ +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_content_status_with_invalid_json(tmp_path: Path) -> None: + """ + Given: content_status.json exists but contains invalid JSON. + When: update_content_status is called. + Then: It should initialize content_status as empty and overwrite the invalid file. + """ + # Given + artifacts_folder = tmp_path + content_status_file = artifacts_folder / "content_status.json" + content_status_file.write_text('{"invalid_json": }') # Invalid JSON + + updater = ContentStatusUpdater(artifacts_folder) + successful_tests = ["test_success"] + failed_tests = ["test_fail"] + + # When + updater.update_content_status(successful_tests, failed_tests) + + # Then + expected_content_status = { + "failed_playbooks": ["test_fail"], + "successful_playbooks": ["test_success"], + } + 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