Improve-CPS-sanity-test

Improve CPS sanity test

Issue-ID: INT-2263
Signed-off-by: pawel.denst <pawel.denst@external.t-mobile.pl>
Change-Id: I64c54920d1c70c5aa8befb058402fe234eaab1e8

diff --git a/setup.cfg b/setup.cfg
index b333c6e..92ff5a0 100644 (file)
--- a/setup.cfg
+++ b/setup.cfg
@@ -20,6 +20,8 @@ package_dir=
 packages=find_namespace:
 setup_requires =
   pytest-runner==5.2
+install_requires =
+  pg8000==1.30.1
 tests_require =
   mock
   pytest

diff --git a/src/onaptests/configuration/basic_cps_settings.py b/src/onaptests/configuration/basic_cps_settings.py
index e359f65..896469e 100644 (file)
--- a/src/onaptests/configuration/basic_cps_settings.py
+++ b/src/onaptests/configuration/basic_cps_settings.py
@@ -31,6 +31,32 @@ ANCHOR_DATA = json.dumps(
         }
     }
 )
+
+ANCHOR_DATA_2 = json.dumps(
+    {
+        "bookstore": {
+            "bookstore-name": "Chapters_2",
+            "categories": [{
+                "code": 1,
+                "name": "SciFi",
+                "books": [{
+                    "title": "2001: A Space Odyssey",
+                    "price": 5
+                }, {
+                    "title": "Dune",
+                    "price": 5
+                }]
+            }, {
+                "code": 2,
+                "name": "Kids",
+                "books": [{
+                    "title": "Matilda"
+                }]
+            }]
+        }
+    }
+)
+
 ANCHOR_NAME = "basic-cps-test-anchor"
 DATASPACE_NAME = "basic-cps-test-dataspace"
 SCHEMA_SET_NAME = "basic-cps-test-schema-set"
@@ -38,3 +64,12 @@ SCHEMA_SET_FILE = Path(get_resource_location("templates/artifacts/cps/bookstore.
 
 SERVICE_NAME = "Basic CPS test"
 SERVICE_COMPONENTS = "CPS"
+QUERY_1 = "/bookstore/categories[@code='1']/books"
+QUERY_2 = "//categories[@code='1']/books[@price=5 and @title='Dune']"
+QUERY_3 = "//bookstore"
+SECRET_NAME = "onap-cps-core-pg-user-creds"
+DATABASE = "cpsdb"
+DB_PRIMARY_HOST = "cps-core-pg-primary"
+DB_PORT = 5432
+DB_LOGIN = "login"
+DB_PASSWORD = "password"

diff --git a/src/onaptests/scenario/basic_cps.py b/src/onaptests/scenario/basic_cps.py
index 13a3d37..59ecf96 100644 (file)
--- a/src/onaptests/scenario/basic_cps.py
+++ b/src/onaptests/scenario/basic_cps.py
@@ -1,7 +1,46 @@
 #!/usr/bin/env python
 """Basic CPS test case."""
-from onaptests.scenario.scenario_base import ScenarioBase
-from onaptests.steps.onboard.cps import CreateCpsAnchorNodeStep
+from onaptests.scenario.scenario_base import BaseStep, ScenarioBase
+from onaptests.steps.onboard.cps import (CheckPostgressDataBaseConnectionStep,
+                                         QueryCpsAnchorNodeStep)
+
+
+class BasicCpsStep(BaseStep):
+    def __init__(self):
+        """Initialize step.
+
+        Substeps:
+            - QueryCpsAnchorNodeStep
+            - CheckPostgressDataBaseConnectionStep.
+        """
+        super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
+        self.add_step(QueryCpsAnchorNodeStep())
+        self.add_step(CheckPostgressDataBaseConnectionStep())
+
+    @property
+    def description(self) -> str:
+        """Step description.
+
+        Used for reports
+
+        Returns:
+            str: Step description
+
+        """
+        return "Basic CPS scenario step"
+
+    @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 "TEST"
 
 
 class BasicCps(ScenarioBase):
@@ -9,7 +48,9 @@ class BasicCps(ScenarioBase):
             - dataspace
             - schemaset
             - anchor
-        And create a node. Use bookstore YANG model (available on CPS repository
+            - node
+        Update and Query on node
+        And check PostgreSQL connection. Use bookstore YANG model (available on CPS repository
         https://github.com/onap/cps/blob/master/cps-service/src/test/resources/bookstore.yang).
         At the end delete what's available to be deleted.
 
@@ -18,4 +59,4 @@ class BasicCps(ScenarioBase):
     def __init__(self, **kwargs):
         """Init Basic CPS."""
         super().__init__('basic_cps', **kwargs)
-        self.test = CreateCpsAnchorNodeStep()
+        self.test = BasicCpsStep()

diff --git a/src/onaptests/steps/onboard/cps.py b/src/onaptests/steps/onboard/cps.py
index 55598bf..60e1990 100644 (file)
--- a/src/onaptests/steps/onboard/cps.py
+++ b/src/onaptests/steps/onboard/cps.py
@@ -1,10 +1,15 @@
 # http://www.apache.org/licenses/LICENSE-2.0
 """CPS onboard module."""
+import base64
 from abc import ABC
 
+import pg8000
+from kubernetes import client, config
 from onapsdk.configuration import settings
 from onapsdk.cps import Anchor, Dataspace, SchemaSet
 
+from onaptests.utils.exceptions import EnvironmentPreparationException
+
 from ..base import BaseStep
 
 
@@ -132,7 +137,7 @@ class CreateCpsAnchorStep(CpsBaseStep):
 
 
 class CreateCpsAnchorNodeStep(CpsBaseStep):
-    """Step to check node on anchor creation."""
+    """Step to create anchor node."""
 
     def __init__(self) -> None:
         """Initialize step.
@@ -170,3 +175,149 @@ class CreateCpsAnchorNodeStep(CpsBaseStep):
         anchor: Anchor = dataspace.get_anchor(settings.ANCHOR_NAME)
         anchor.delete_nodes("/")
         super().cleanup()
+
+
+class UpdateCpsAnchorNodeStep(CpsBaseStep):
+    """Step to update node on anchor creation."""
+
+    def __init__(self) -> None:
+        """Initialize step.
+
+        Substeps:
+            - CreateCpsAnchorNodeStep.
+        """
+        super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
+        self.add_step(CreateCpsAnchorNodeStep())
+
+    @property
+    def description(self) -> str:
+        """Step description."""
+        return "Update CPS anchor node"
+
+    @BaseStep.store_state
+    def execute(self) -> None:
+        """Update a node on an anchor created on substep.
+
+        Use settings values:
+         - DATASPACE_NAME,
+         - ANCHOR_NAME,
+         - ANCHOR_DATA_2.
+
+         """
+        super().execute()
+        dataspace: Dataspace = Dataspace(settings.DATASPACE_NAME)
+        anchor: Anchor = dataspace.get_anchor(settings.ANCHOR_NAME)
+        anchor.update_node("/", settings.ANCHOR_DATA_2)
+
+
+class QueryCpsAnchorNodeStep(CpsBaseStep):
+    """Step to check query on node."""
+
+    def __init__(self) -> None:
+        """Initialize step.
+
+        Substeps:
+            - UpdateCpsAnchorNodeStep.
+
+        """
+        super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
+        self.add_step(UpdateCpsAnchorNodeStep())
+
+    @property
+    def description(self) -> str:
+        """Step description."""
+        return "Query node"
+
+    @BaseStep.store_state
+    def execute(self) -> None:
+        """Query on node on an anchor created on substep.
+
+        Use settings values:
+         - DATASPACE_NAME,
+         - ANCHOR_NAME,
+         - QUERY_1,
+         - QUERY_2,
+         - QUERY_3.
+
+         """
+        super().execute()
+        dataspace: Dataspace = Dataspace(settings.DATASPACE_NAME)
+        anchor: Anchor = dataspace.get_anchor(settings.ANCHOR_NAME)
+        anchor.query_node(settings.QUERY_1)
+        anchor.query_node(settings.QUERY_2)
+        anchor.query_node(settings.QUERY_3)
+
+
+class CheckPostgressDataBaseConnectionStep(CpsBaseStep):
+    """Step to test connection with postgress."""
+
+    def __init__(self) -> None:
+        """Initialize step."""
+        super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
+
+    @property
+    def description(self) -> str:
+        """Step description."""
+        return "Establish connection with Postgress and execute the query"
+
+    def get_database_credentials(self):
+        config.load_kube_config()
+        api_instance = client.CoreV1Api()
+        try:
+            secret = api_instance.read_namespaced_secret(
+                settings.SECRET_NAME, settings.K8S_ONAP_NAMESPACE)
+            if secret.data:
+                if settings.DB_LOGIN in secret.data and settings.DB_PASSWORD in secret.data:
+                    login_base64 = secret.data[settings.DB_LOGIN]
+                    self.login = base64.b64decode(login_base64).decode("utf-8")
+                    password_base64 = secret.data[settings.DB_PASSWORD]
+                    self.password = base64.b64decode(password_base64).decode("utf-8")
+                else:
+                    raise EnvironmentPreparationException(
+                        "Login key or password key not found in secret")
+            else:
+                raise EnvironmentPreparationException("Secret data not found in secret")
+        except client.rest.ApiException as e:
+            self.login = None
+            self.password = None
+            raise EnvironmentPreparationException("Error accessing secret") from e
+
+    def connect_to_postgress(self):
+        self.get_database_credentials()
+        if self.login and self.password:
+            db_params = {
+                "user": self.login,
+                "password": self.password,
+                "host": settings.DB_PRIMARY_HOST,
+                "database": settings.DATABASE,
+                "port": settings.DB_PORT
+            }
+            try:
+                connection = pg8000.connect(**db_params)
+                cursor = connection.cursor()
+                select_query = "SELECT * FROM yang_resource LIMIT 1;"
+                cursor.execute(select_query)
+                if cursor:
+                    cursor.close()
+                if connection:
+                    connection.close()
+            except pg8000.Error as e:
+                self._logger.exception(f"Error while connecting to PostgreSQL: {str(e)}")
+                raise
+
+    @BaseStep.store_state
+    def execute(self) -> None:
+        """Establish connection with Postgress and execute the query.
+
+        Use settings values:
+         - DB_PRIMARY_HOST,
+         - DATABASE,
+         - DB_PORT,
+         - K8S_ONAP_NAMESPACE,
+         - SECRET_NAME,
+         - DB_LOGIN,
+         - DB_PASSWORD.
+
+         """
+        super().execute()
+        self.connect_to_postgress()