src/onaptests/steps/cloud/k8s_connectivity_info_create.py

   1 from onapsdk.configuration import settings
   2 from onapsdk.msb.k8s import ConnectivityInfo
   3
   4 from ..base import BaseStep
   5
   6 class K8SConnectivityInfoStep(BaseStep):
   7     """CreateConnnectivityInfoStep."""
   8
   9     @BaseStep.store_state
  10     def execute(self):
  11         """Creation k8s connectivity information
  12
  13         Use settings values:
  14          - CLOUD_REGION_ID,
  15          - CLOUD_REGION_CLOUD_OWNER,
  16          - K8S_KUBECONFIG_FILE.
  17         """
  18         super().execute()
  19         ######## Create Connectivity Info #########################################
  20         try:
  21             self._logger.info("Check if k8s connectivity information exists")
  22             ConnectivityInfo.get_connectivity_info_by_region_id(settings.CLOUD_REGION_ID)
  23         except ValueError:
  24             self._logger.info("Create the k8s connectivity information")
  25             ConnectivityInfo.create(settings.CLOUD_REGION_ID,
  26                                     settings.CLOUD_REGION_CLOUD_OWNER,
  27                                     open(settings.K8S_KUBECONFIG_FILE, 'rb').read())
  28
  29     def cleanup(self) -> None:
  30         """Cleanup K8S Connectivity information.
  31         """
  32         self._logger.info("Clean the k8s connectivity information")
  33         super().cleanup()
  34         connectinfo = ConnectivityInfo.get_connectivity_info_by_region_id(settings.CLOUD_REGION_ID)
  35         connectinfo.delete()