+++ /dev/null
-#!/usr/bin/env python
-"""CBA Verification test case."""
-import difflib
-import json
-import os
-import shutil
-from io import BytesIO
-
-from onapsdk.cds.blueprint import Blueprint
-from onapsdk.cds.cds_element import CdsElement
-from onapsdk.configuration import settings
-from onapsdk.exceptions import ResourceNotFound
-
-from onaptests.steps.base import BaseStep
-from onaptests.utils.exceptions import CbaVerificationException
-from onaptests.utils.gitlab import GitLabClient
-
-
-class TestCbaBaseStep(BaseStep):
- """Test CBA Base Step"""
-
- def __init__(self, cleanup=BaseStep.HAS_NO_CLEANUP, break_on_error: bool = True):
- """TestCbaBaseStep."""
- super().__init__(cleanup=cleanup, break_on_error=break_on_error)
-
- @property
- def component(self) -> str:
- """Component name.
-
- Name of component which step is related with.
- Most is the name of ONAP component.
-
- Returns:
- str: Component name
-
- """
- return "CBA"
-
-
-class DownloadCbaStep(TestCbaBaseStep, CdsElement):
- """Step to download CBA content and convert it to zip."""
-
- def __init__(self, cba_data):
- """Download CBA step."""
-
- super().__init__(cleanup=settings.CLEANUP_FLAG, break_on_error=True)
- self.cba_data = cba_data
- self.cba = None
-
- def get_details_by_name_and_version(self, name: str, version: str):
- """Get CBA details from its name and version."""
- cba_details = self.send_message(
- action="Download CBA Details",
- method="GET",
- url=f"{self._url}/api/v1/blueprint-model/by-name/{name}/version/{version}",
- auth=self.auth
- )
- return cba_details
-
- def get_by_name_and_version(self, name: str, version: str) -> Blueprint:
- """Get CBA blueprint from its name and version."""
- cba_data = self.send_message(
- action="Download CBA Content",
- method="GET",
- url=f"{self._url}/api/v1/blueprint-model/download/by-name/{name}/version/{version}",
- auth=self.auth
- )
- return Blueprint(BytesIO(cba_data.content).read())
-
- @property
- def description(self) -> str:
- """Step description.
-
- Used for reports
-
- Returns:
- str: Step description
-
- """
- name = self.cba_data["name"]
- return f"Downloading CBA {name} from CDS"
-
- def zip_cba_from_cds(self):
- """Check CBA zip downloaded from CDS."""
- name = self.cba_data["name"]
- version = self.cba_data["version"]
- zip_path = f"{settings.LOCAL_PATH}/{name}/{version}"
- os.makedirs(zip_path, exist_ok=True)
- zip_file = f"{zip_path}/cba.zip"
- try:
- self.get_details_by_name_and_version(name, version)
- blueprint = self.cba = self.get_by_name_and_version(name, version)
- blueprint.save(zip_file)
- shutil.unpack_archive(zip_file, zip_path, 'zip')
- except ResourceNotFound as exc:
- self._logger.error(f"CBA {name}-{version} Not Found")
- raise CbaVerificationException("CBA Not Found in CDS") from exc
-
- def delete_cba_directory(self):
- """Delete local CBA content."""
- name = self.cba_data["name"]
- delete_path = f"{settings.LOCAL_PATH}/{name}"
- if os.path.exists(delete_path):
- try:
- shutil.rmtree(delete_path)
- self._logger.info(f"The directory '{delete_path}' has been successfully deleted.")
- except OSError as e:
- self._logger.eror(f"Error: {e}")
- error_message = f"Error while deleting directory: {e}"
- self._logger.error(error_message)
- raise CbaVerificationException(error_message) from e
- else:
- self._logger.info(f"The directory '{delete_path}' does not exist.")
-
- @BaseStep.store_state
- def execute(self) -> None:
- """Download CBA content and convert it to zip."""
- super().execute()
- self.zip_cba_from_cds()
-
- @BaseStep.store_state(cleanup=True)
- def cleanup(self) -> None:
- """Delete downloaded cba content directory."""
- self.delete_cba_directory()
- super().cleanup()
-
-
-class TestCbaStep(TestCbaBaseStep):
- """Step to check if workflow exists and compare CBA from Gitlab with CBA downloaded from CDS."""
-
- def __init__(self, cba_data):
- """Initialize step.
-
- Substeps:
- - DownloadCbaStep
- """
- super().__init__(break_on_error=False)
- self.cba_data = cba_data
- self.download_cba_step = DownloadCbaStep(cba_data)
- self.add_step(self.download_cba_step)
-
- @property
- def description(self) -> str:
- """Step description.
-
- Used for reports
-
- Returns:
- str: Step description
-
- """
- name = self.cba_data["name"]
- return f"Compare CBA {name} Content with Gitlab"
-
- def show_differences(self, git_file_content, local_file_content, entry_path):
- """Show CBA file difference: git and local."""
- diff = difflib.unified_diff(
- git_file_content.splitlines(),
- local_file_content.splitlines(),
- fromfile=entry_path + ' (git)',
- tofile=entry_path + ' (local)'
- )
- diff_text = "\n".join(diff)
- return f"\n\n{diff_text}\n"
-
- def load_json(self, git_file_content, local_file_path):
- """Load CBA json file with formatting."""
- try:
- git_file_content = json.loads(git_file_content)
- if os.path.exists(local_file_path):
- with open(local_file_path, 'r', encoding="utf-8") as local_file:
- local_file_content = local_file.read()
- local_file_content = json.loads(local_file_content)
- git_file_content = dict(sorted(git_file_content.items()))
- local_file_content = dict(sorted(local_file_content.items()))
- return (json.dumps(git_file_content, indent=4, sort_keys=True),
- json.dumps(local_file_content, indent=4, sort_keys=True))
- except json.JSONDecodeError as e:
- self._logger.error(f"Error decoding JSON: {e}")
- return None, None
-
- def check_if_path_in_pattern_list(self, path, patterns):
- """Check if file path is in the pattern."""
- for pattern in patterns:
- if pattern in path:
- return True
- return False
-
- def compare_directories_recursive( # noqa: C901
- self, branch, git_directory_path, local_directory_path, gitlab_id):
- """Compare local and gitlab CBA directories."""
- enrichment = self.cba_data["enrichment"]
- try:
- git_directory_entries = GitLabClient.get_directory_entries(
- branch, git_directory_path, gitlab_id)
- if len(git_directory_entries) == 0:
- self._logger.error(f"Folder '{git_directory_path}' on gitlab "
- f"with projct ID: {gitlab_id} does not exist")
- raise CbaVerificationException("Cannot locate repo folder in gitlab")
- except TypeError as exc:
- self._logger.error(f"Branch '{branch}' on gitlab "
- f"with projct ID: {gitlab_id} does not exist")
- raise CbaVerificationException("Cannot locate branch in gitlab") from exc
- ident_files = []
- diff_files = []
- git_only_files = []
- differences = {}
- for entry in git_directory_entries:
- if self.check_if_path_in_pattern_list(entry.path, settings.IGNORE_FILES):
- # Check if the path is in ignore_files set
- continue
- binary_check = False
- if entry.type == "tree":
- ident, diff, git, show = self.compare_directories_recursive(
- branch, entry.path, local_directory_path + '/' + entry.name, gitlab_id)
- ident_files.extend(ident)
- diff_files.extend(diff)
- git_only_files.extend(git)
- differences.update(show)
- else: # It's a file
- git_file_content = GitLabClient.get_text_file_content(branch, entry.path, gitlab_id)
- git_file_content = git_file_content.replace("\r\n", "\n").strip()
- entry_name = os.path.basename(entry.path)
- local_file_path = os.path.join(local_directory_path, entry_name)
- if os.path.exists(local_file_path):
- try:
- with open(local_file_path, 'r', encoding='utf-8') as local_file:
- local_file_content = local_file.read().replace("\r\n", "\n").strip()
- except UnicodeDecodeError:
- binary_check = True
- if binary_check is False:
- if 'Definitions/' in entry.path:
- git_file_content, local_file_content = self.load_json(
- git_file_content, local_file_path)
- if git_file_content == local_file_content:
- ident_files.append(entry.path)
- else:
- if (not enrichment or
- not self.check_if_path_in_pattern_list(entry.path,
- settings.ENRICHMENT_FILES)):
- diff_files.append(entry.path)
- diff_text = self.show_differences(
- git_file_content, local_file_content, entry.path
- )
- differences[entry.path] = diff_text
- else:
- git_only_files.append(entry.path)
- return ident_files, diff_files, git_only_files, differences
-
- def summarize_comparison(self, local_path, gitlab_id, branch):
- """Summarize CBA comparison."""
- gitlab_repo_cba = self.cba_data["gitlab_repo_cba"]
- ident_files, diff_files, git_only_files, differences = self.compare_directories_recursive(
- branch, gitlab_repo_cba,
- local_path, gitlab_id)
- error = False
- if ident_files:
- self._logger.info(f"Identical files: {ident_files}")
- self._logger.info(f"There are {len(ident_files)} identical files")
- if diff_files:
- dif_error_message = f"Different files: {diff_files}"
- self._logger.error(dif_error_message)
- self._logger.error(f"There are {len(diff_files)} different files")
- error = True
- if git_only_files:
- git_error_message = f"Files that exists only on Gitlab: {git_only_files}"
- self._logger.error(git_error_message)
- self._logger.error(f"There are {len(git_only_files)} files that exist only on Gitlab")
- error = True
- if differences:
- for file_path, diff_text in differences.items():
- self._logger.info(f"Differences in file: {file_path}")
- self._logger.info(diff_text)
- if error:
- raise CbaVerificationException(
- "CBA content differencies between Gitlab and CDS")
-
- @BaseStep.store_state
- def execute(self) -> None:
- """Check if workflow exists and compare CBA from Gitlab with CBA downloaded from CDS."""
- super().execute()
- gitlab_id = str(self.cba_data["gitlab_project_id"])
- branch = self.cba_data["gitlab_branch"]
- name = self.cba_data["name"]
- version = self.cba_data["version"]
- local_path = f"{settings.LOCAL_PATH}/{name}/{version}"
- self.summarize_comparison(local_path, gitlab_id, branch)
Code Review / testsuite / pythonsdk-tests.git / commitdiff