src/onaptests/steps/onboard/cps.py

   1 # http://www.apache.org/licenses/LICENSE-2.0
   2 """CPS onboard module."""
   3 import base64
   4 from abc import ABC
   5
   6 import pg8000
   7 from kubernetes import client, config
   8 from onapsdk.configuration import settings
   9 from onapsdk.cps import Anchor, Dataspace, SchemaSet
  10
  11 from onaptests.utils.exceptions import EnvironmentPreparationException
  12
  13 from ..base import BaseStep
  14
  15
  16 class CpsBaseStep(BaseStep, ABC):
  17     """Abstract CPS base step."""
  18
  19     @property
  20     def component(self) -> str:
  21         """Component name."""
  22         return "CPS"
  23
  24
  25 class CreateCpsDataspaceStep(CpsBaseStep):
  26     """Step to create a dataspace."""
  27
  28     def __init__(self) -> None:
  29         """Initialize step."""
  30         super().__init__(cleanup=settings.CLEANUP_FLAG)
  31
  32     @property
  33     def description(self) -> str:
  34         """Step description."""
  35         return "Create CPS dataspace."
  36
  37     @BaseStep.store_state
  38     def execute(self) -> None:
  39         """Create a dataspace.
  40
  41         Use settings values:
  42          - DATASPACE_NAME.
  43
  44         """
  45         super().execute()
  46         Dataspace.create(settings.DATASPACE_NAME)
  47
  48     @BaseStep.store_state(cleanup=True)
  49     def cleanup(self) -> None:
  50         """Delete created dataspace."""
  51         dataspace: Dataspace = Dataspace(settings.DATASPACE_NAME)
  52         dataspace.delete()
  53         super().cleanup()
  54
  55
  56 class CreateCpsSchemaSetStep(CpsBaseStep):
  57     """Step to check schema-set creation."""
  58
  59     def __init__(self) -> None:
  60         """Initialize step.
  61
  62         Substeps:
  63             - CreateCpsDataspaceStep.
  64         """
  65         super().__init__(cleanup=settings.CLEANUP_FLAG)
  66         self.add_step(CreateCpsDataspaceStep())
  67
  68     @property
  69     def description(self) -> str:
  70         """Step description."""
  71         return "Create CPS bookstore schema set"
  72
  73     @BaseStep.store_state
  74     def execute(self) -> None:
  75         """Get dataspace created on substep and create schema-set.
  76
  77         Use settings values:
  78          - DATASPACE_NAME,
  79          - SCHEMA_SET_NAME.
  80
  81         """
  82         super().execute()
  83         dataspace: Dataspace = Dataspace(settings.DATASPACE_NAME)
  84         with settings.SCHEMA_SET_FILE.open("rb") as schema_set_file:
  85             dataspace.create_schema_set(settings.SCHEMA_SET_NAME, schema_set_file)
  86
  87     @BaseStep.store_state(cleanup=True)
  88     def cleanup(self) -> None:
  89         """Delete created schema-set."""
  90         dataspace: Dataspace = Dataspace(settings.DATASPACE_NAME)
  91         schema_set = dataspace.get_schema_set(settings.SCHEMA_SET_NAME)
  92         schema_set.delete()
  93         super().cleanup()
  94
  95
  96 class CreateCpsAnchorStep(CpsBaseStep):
  97     """Step to create an anchor."""
  98
  99     def __init__(self) -> None:
 100         """Initialize step.
 101
 102         Substeps:
 103             - CreateCpsSchemaSetStep.
 104         """
 105         super().__init__(cleanup=settings.CLEANUP_FLAG)
 106         self.add_step(CreateCpsSchemaSetStep())
 107
 108     @property
 109     def description(self) -> str:
 110         """Step description."""
 111         return "Create CPS anchor"
 112
 113     @BaseStep.store_state
 114     def execute(self) -> None:
 115         """Create anchor.
 116
 117         Get dataspace and schema-set created substeps and create anchor.
 118
 119         Use settings values:
 120          - DATASPACE_NAME,
 121          - SCHEMA_SET_NAME,
 122          - ANCHOR_DATA.
 123
 124         """
 125         super().execute()
 126         dataspace: Dataspace = Dataspace(settings.DATASPACE_NAME)
 127         schema_set: SchemaSet = dataspace.get_schema_set(settings.SCHEMA_SET_NAME)
 128         dataspace.create_anchor(schema_set, settings.ANCHOR_NAME)
 129
 130     @BaseStep.store_state(cleanup=True)
 131     def cleanup(self) -> None:
 132         """Delete an anchor."""
 133         dataspace: Dataspace = Dataspace(settings.DATASPACE_NAME)
 134         anchor: Anchor = dataspace.get_anchor(settings.ANCHOR_NAME)
 135         anchor.delete()
 136         super().cleanup()
 137
 138
 139 class CreateCpsAnchorNodeStep(CpsBaseStep):
 140     """Step to create anchor node."""
 141
 142     def __init__(self) -> None:
 143         """Initialize step.
 144
 145         Substeps:
 146             - CreateCpsAnchorStep.
 147         """
 148         super().__init__(cleanup=settings.CLEANUP_FLAG)
 149         self.add_step(CreateCpsAnchorStep())
 150
 151     @property
 152     def description(self) -> str:
 153         """Step description."""
 154         return "Create CPS anchor node"
 155
 156     @BaseStep.store_state
 157     def execute(self) -> None:
 158         """Create a node on an anchor created on substep.
 159
 160         Use settings values:
 161          - DATASPACE_NAME,
 162          - ANCHOR_NAME,
 163          - ANCHOR_DATA.
 164
 165          """
 166         super().execute()
 167         dataspace: Dataspace = Dataspace(settings.DATASPACE_NAME)
 168         anchor: Anchor = dataspace.get_anchor(settings.ANCHOR_NAME)
 169         anchor.create_node(settings.ANCHOR_DATA)
 170
 171     @BaseStep.store_state(cleanup=True)
 172     def cleanup(self) -> None:
 173         """Delete nodes."""
 174         dataspace: Dataspace = Dataspace(settings.DATASPACE_NAME)
 175         anchor: Anchor = dataspace.get_anchor(settings.ANCHOR_NAME)
 176         anchor.delete_nodes("/")
 177         super().cleanup()
 178
 179
 180 class UpdateCpsAnchorNodeStep(CpsBaseStep):
 181     """Step to update node on anchor creation."""
 182
 183     def __init__(self) -> None:
 184         """Initialize step.
 185
 186         Substeps:
 187             - CreateCpsAnchorNodeStep.
 188         """
 189         super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
 190         self.add_step(CreateCpsAnchorNodeStep())
 191
 192     @property
 193     def description(self) -> str:
 194         """Step description."""
 195         return "Update CPS anchor node"
 196
 197     @BaseStep.store_state
 198     def execute(self) -> None:
 199         """Update a node on an anchor created on substep.
 200
 201         Use settings values:
 202          - DATASPACE_NAME,
 203          - ANCHOR_NAME,
 204          - ANCHOR_DATA_2.
 205
 206          """
 207         super().execute()
 208         dataspace: Dataspace = Dataspace(settings.DATASPACE_NAME)
 209         anchor: Anchor = dataspace.get_anchor(settings.ANCHOR_NAME)
 210         anchor.update_node("/", settings.ANCHOR_DATA_2)
 211
 212
 213 class QueryCpsAnchorNodeStep(CpsBaseStep):
 214     """Step to check query on node."""
 215
 216     def __init__(self) -> None:
 217         """Initialize step.
 218
 219         Substeps:
 220             - UpdateCpsAnchorNodeStep.
 221
 222         """
 223         super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
 224         self.add_step(UpdateCpsAnchorNodeStep())
 225
 226     @property
 227     def description(self) -> str:
 228         """Step description."""
 229         return "Query node"
 230
 231     @BaseStep.store_state
 232     def execute(self) -> None:
 233         """Query on node on an anchor created on substep.
 234
 235         Use settings values:
 236          - DATASPACE_NAME,
 237          - ANCHOR_NAME,
 238          - QUERY_1,
 239          - QUERY_2,
 240          - QUERY_3.
 241
 242          """
 243         super().execute()
 244         dataspace: Dataspace = Dataspace(settings.DATASPACE_NAME)
 245         anchor: Anchor = dataspace.get_anchor(settings.ANCHOR_NAME)
 246         anchor.query_node(settings.QUERY_1)
 247         anchor.query_node(settings.QUERY_2)
 248         anchor.query_node(settings.QUERY_3)
 249
 250
 251 class CheckPostgressDataBaseConnectionStep(CpsBaseStep):
 252     """Step to test connection with postgress."""
 253
 254     def __init__(self) -> None:
 255         """Initialize step."""
 256         super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
 257
 258     @property
 259     def description(self) -> str:
 260         """Step description."""
 261         return "Establish connection with Postgress and execute the query"
 262
 263     def get_database_credentials(self):
 264         config.load_kube_config()
 265         api_instance = client.CoreV1Api()
 266         try:
 267             secret = api_instance.read_namespaced_secret(
 268                 settings.SECRET_NAME, settings.K8S_ONAP_NAMESPACE)
 269             if secret.data:
 270                 if settings.DB_LOGIN in secret.data and settings.DB_PASSWORD in secret.data:
 271                     login_base64 = secret.data[settings.DB_LOGIN]
 272                     self.login = base64.b64decode(login_base64).decode("utf-8")
 273                     password_base64 = secret.data[settings.DB_PASSWORD]
 274                     self.password = base64.b64decode(password_base64).decode("utf-8")
 275                 else:
 276                     raise EnvironmentPreparationException(
 277                         "Login key or password key not found in secret")
 278             else:
 279                 raise EnvironmentPreparationException("Secret data not found in secret")
 280         except client.rest.ApiException as e:
 281             self.login = None
 282             self.password = None
 283             raise EnvironmentPreparationException("Error accessing secret") from e
 284
 285     def connect_to_postgress(self):
 286         self.get_database_credentials()
 287         if self.login and self.password:
 288             db_params = {
 289                 "user": self.login,
 290                 "password": self.password,
 291                 "host": settings.DB_PRIMARY_HOST,
 292                 "database": settings.DATABASE,
 293                 "port": settings.DB_PORT
 294             }
 295             try:
 296                 connection = pg8000.connect(**db_params)
 297                 cursor = connection.cursor()
 298                 select_query = "SELECT * FROM yang_resource LIMIT 1;"
 299                 cursor.execute(select_query)
 300                 if cursor:
 301                     cursor.close()
 302                 if connection:
 303                     connection.close()
 304             except pg8000.Error as e:
 305                 self._logger.exception(f"Error while connecting to PostgreSQL: {str(e)}")
 306                 raise
 307
 308     @BaseStep.store_state
 309     def execute(self) -> None:
 310         """Establish connection with Postgress and execute the query.
 311
 312         Use settings values:
 313          - DB_PRIMARY_HOST,
 314          - DATABASE,
 315          - DB_PORT,
 316          - K8S_ONAP_NAMESPACE,
 317          - SECRET_NAME,
 318          - DB_LOGIN,
 319          - DB_PASSWORD.
 320
 321          """
 322         super().execute()
 323         self.connect_to_postgress()