aboutsummaryrefslogtreecommitdiffstats
path: root/src/onaptests/steps/onboard/cds.py
diff options
context:
space:
mode:
authorMichal Jagiello <michal.jagiello@t-mobile.pl>2021-10-11 09:12:43 +0000
committerGerrit Code Review <gerrit@onap.org>2021-10-11 09:12:43 +0000
commit3a93d05df6e61d25d2383658adac87c8ef34a168 (patch)
tree8b247e3c7d2bf22a9b6d50b764afba6946b5e746 /src/onaptests/steps/onboard/cds.py
parent1a77bcd6eb5ab695a232004b88e5e59d9d9ff6a3 (diff)
parentc31b7fb464fbb8bb0c7d8d2b3dc7b20f4a04cff5 (diff)
Merge "[TEST] CDS resource-resolution test"
Diffstat (limited to 'src/onaptests/steps/onboard/cds.py')
-rw-r--r--src/onaptests/steps/onboard/cds.py38
1 files changed, 36 insertions, 2 deletions
diff --git a/src/onaptests/steps/onboard/cds.py b/src/onaptests/steps/onboard/cds.py
index cbd69ce..9239c43 100644
--- a/src/onaptests/steps/onboard/cds.py
+++ b/src/onaptests/steps/onboard/cds.py
@@ -8,6 +8,7 @@ from typing import Any, Dict
from kubernetes import client, config
from kubernetes.client.exceptions import ApiException
from onapsdk.cds import Blueprint, DataDictionarySet
+from onapsdk.cds.blueprint import Workflow
from onapsdk.cds.blueprint_processor import Blueprintprocessor
from onapsdk.configuration import settings
import urllib3
@@ -222,12 +223,45 @@ class CbaPublishStep(CDSBaseStep):
@BaseStep.store_state
def execute(self) -> None:
- """Enrich CBA file.
+ """Publish CBA file.
Use settings values:
- - CDS_DD_FILE.
+ - CDS_CBA_ENRICHED.
"""
super().execute()
blueprint: Blueprint = Blueprint.load_from_file(settings.CDS_CBA_ENRICHED)
blueprint.publish()
+
+
+class CbaProcessStep(CDSBaseStep):
+ """Process CBA step."""
+
+ def __init__(self, cleanup=False) -> None:
+ """Initialize CBA process step."""
+ super().__init__(cleanup=cleanup)
+ self.add_step(CbaPublishStep(cleanup=cleanup))
+
+ @property
+ def description(self) -> str:
+ """Step description."""
+ return "Process CBA file."
+
+ @BaseStep.store_state
+ def execute(self) -> None:
+ """Process CBA file.
+
+ Check if output is equal to expected
+
+ Use settings values:
+ - CDS_CBA_ENRICHED,
+ - CDS_WORKFLOW_NAME,
+ - CDS_WORKFLOW_INPUT
+
+ """
+ super().execute()
+ blueprint: Blueprint = Blueprint.load_from_file(settings.CDS_CBA_ENRICHED)
+ workflow: Workflow = blueprint.get_workflow_by_name(settings.CDS_WORKFLOW_NAME)
+ output: Dict[str, Any] = workflow.execute(settings.CDS_WORKFLOW_INPUT)
+ if not output == settings.CDS_WORKFLOW_EXPECTED_OUTPUT:
+ raise OnapTestException("Response is not equal to the expected one")