aboutsummaryrefslogtreecommitdiffstats
path: root/src/onaptests/steps/onboard
diff options
context:
space:
mode:
Diffstat (limited to 'src/onaptests/steps/onboard')
-rw-r--r--src/onaptests/steps/onboard/cps.py144
1 files changed, 144 insertions, 0 deletions
diff --git a/src/onaptests/steps/onboard/cps.py b/src/onaptests/steps/onboard/cps.py
new file mode 100644
index 0000000..1c59086
--- /dev/null
+++ b/src/onaptests/steps/onboard/cps.py
@@ -0,0 +1,144 @@
+# 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()