--- /dev/null
+import logging.config
+import onaptests.utils.exceptions as onap_test_exceptions
+from onapsdk.configuration import settings
+from onaptests.steps.onboard.cps import CreateCpsAnchorStep, CreateCpsSchemaSetStep, CreateCpsDataspaceStep, CreateCpsAnchorNodeStep
+
+
+
+if __name__ == "__main__":
+ # logging configuration for onapsdk, it is not requested for onaptests
+ # Correction requested in onapsdk to avoid having this duplicate code
+ logging.config.dictConfig(settings.LOG_CONFIG)
+ logger = logging.getLogger("Basic CPS")
+
+ basic_cps = CreateCpsAnchorNodeStep(
+ cleanup=settings.CLEANUP_FLAG)
+ try:
+ basic_cps.execute()
+ basic_cps.cleanup()
+ except onap_test_exceptions.TestConfigurationException:
+ logger.error("Basic CPS configuration error")
+ basic_cps.reports_collection.generate_report()
--- /dev/null
+#!/usr/bin/env python
+"""Basic CPS test case."""
+import logging
+import time
+from xtesting.core import testcase
+from onapsdk.configuration import settings
+from onapsdk.exceptions import SDKException
+from onaptests.steps.onboard.service import YamlTemplateServiceOnboardStep
+from onaptests.utils.exceptions import OnapTestException
+
+class BasicCps(testcase.TestCase):
+ """Create CPS resources:
+ - dataspace
+ - schemaset
+ - anchor
+ And create a node. 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.
+
+ """
+
+ __logger = logging.getLogger(__name__)
+
+ def __init__(self, **kwargs):
+ """Init Basic CPS."""
+ if "case_name" not in kwargs:
+ kwargs["case_name"] = 'basic_cps'
+ super().__init__(**kwargs)
+ self.__logger.debug("BasicCps init started")
+ # self.test = YamlTemplateServiceOnboardStep( # To be created
+ # cleanup=settings.CLEANUP_FLAG)
+ self.start_time = None
+ self.stop_time = None
+ self.result = 0
+
+ def run(self):
+ """Run basic_cps and create some CPS resources"""
+ self.start_time = time.time()
+ self.__logger.debug("start time")
+ try:
+ self.test.execute()
+ self.test.cleanup()
+ self.__logger.info("Basic CPS test passed")
+ self.result = 100
+ except OnapTestException as exc:
+ self.result = 0
+ self.__logger.exception(exc.error_message)
+ except SDKException:
+ self.result = 0
+ self.__logger.exception("SDK Exception")
+ finally:
+ self.stop_time = time.time()
+
+ def clean(self):
+ """Clean Additional resources if needed."""
+ self.__logger.info("Generate Test report")
+ self.test.reports_collection.generate_report()
--- /dev/null
+# http://www.apache.org/licenses/LICENSE-2.0
+"""CPS onboard module."""
+from abc import ABC
+
+from onapsdk.configuration import settings
+from onapsdk.cps import Anchor, Dataspace, SchemaSet
+
+from ..base import BaseStep
+
+
+class CpsBaseStep(BaseStep, ABC):
+ """Abstract CPS base step."""
+
+ @property
+ def component(self) -> str:
+ """Component name."""
+ return "CPS"
+
+
+class CreateCpsDataspaceStep(CpsBaseStep):
+ """Step to create a dataspace."""
+
+ @property
+ def description(self) -> str:
+ """Step description."""
+ return "Create CPS dataspace."
+
+ @BaseStep.store_state
+ def execute(self) -> None:
+ """Create a dataspace."""
+ super().execute()
+ Dataspace.create(settings.DATASPACE_NAME)
+
+ @BaseStep.store_state(cleanup=True)
+ def cleanup(self) -> None:
+ """Delete created dataspace."""
+ dataspace: Dataspace = Dataspace(settings.DATASPACE_NAME)
+ dataspace.delete()
+ super().cleanup()
+
+
+class CreateCpsSchemaSetStep(CpsBaseStep):
+ """Step to check schema-set creation."""
+
+ def __init__(self, cleanup: bool = False) -> None:
+ """Initialize step.
+
+ Substeps:
+ - CreateCpsDataspaceStep.
+ """
+ super().__init__(cleanup)
+ self.add_step(CreateCpsDataspaceStep(cleanup))
+
+ @property
+ def description(self) -> str:
+ """Step description."""
+ return "Create CPS bookstore schema set"
+
+ @BaseStep.store_state
+ def execute(self) -> None:
+ """Get dataspace created on substep and create schema-set."""
+ super().execute()
+ dataspace: Dataspace = Dataspace(settings.DATASPACE_NAME)
+ with settings.SCHEMA_SET_FILE.open("rb") as schema_set_file:
+ dataspace.create_schema_set(settings.SCHEMA_SET_NAME, schema_set_file)
+
+ @BaseStep.store_state(cleanup=True)
+ def cleanup(self) -> None:
+ """Delete created schema-set."""
+ dataspace: Dataspace = Dataspace(settings.DATASPACE_NAME)
+ schema_set = dataspace.get_schema_set(settings.SCHEMA_SET_NAME)
+ schema_set.delete()
+ super().cleanup()
+
+
+class CreateCpsAnchorStep(CpsBaseStep):
+ """Step to create an anchor."""
+
+ def __init__(self, cleanup: bool = False) -> None:
+ """Initialize step.
+
+ Substeps:
+ - CreateCpsSchemaSetStep.
+ """
+ super().__init__(cleanup)
+ self.add_step(CreateCpsSchemaSetStep(cleanup))
+
+ @property
+ def description(self) -> str:
+ """Step description."""
+ return "Create CPS anchor"
+
+ @BaseStep.store_state
+ def execute(self) -> None:
+ """Create anchor.
+
+ Get dataspace and schema-set created substeps and create anchor.
+ """
+ super().execute()
+ dataspace: Dataspace = Dataspace(settings.DATASPACE_NAME)
+ schema_set: SchemaSet = dataspace.get_schema_set(settings.SCHEMA_SET_NAME)
+ dataspace.create_anchor(schema_set, settings.ANCHOR_NAME)
+
+ @BaseStep.store_state(cleanup=True)
+ def cleanup(self) -> None:
+ """Delete an anchor."""
+ dataspace: Dataspace = Dataspace(settings.DATASPACE_NAME)
+ anchor: Anchor = dataspace.get_anchor(settings.ANCHOR_NAME)
+ anchor.delete()
+ super().cleanup()
+
+
+class CreateCpsAnchorNodeStep(CpsBaseStep):
+ """Step to check node on anchor creation."""
+
+ def __init__(self, cleanup: bool = False) -> None:
+ """Initialize step.
+
+ Substeps:
+ - CreateCpsAnchorStep.
+ """
+ super().__init__(cleanup)
+ self.add_step(CreateCpsAnchorStep(cleanup))
+
+ @property
+ def description(self) -> str:
+ """Step description."""
+ return "Create CPS anchor node"
+
+ @BaseStep.store_state
+ def execute(self) -> None:
+ """Create a node on an anchor created on substep."""
+ super().execute()
+ dataspace: Dataspace = Dataspace(settings.DATASPACE_NAME)
+ anchor: Anchor = dataspace.get_anchor(settings.ANCHOR_NAME)
+ anchor.create_node(settings.ANCHOR_DATA)
+
+ @BaseStep.store_state(cleanup=True)
+ def cleanup(self) -> None:
+ """Delete nodes."""
+ dataspace: Dataspace = Dataspace(settings.DATASPACE_NAME)
+ anchor: Anchor = dataspace.get_anchor(settings.ANCHOR_NAME)
+ anchor.delete_nodes("/")
+ super().cleanup()