diff options
author | 2021-07-20 10:15:32 +0000 | |
---|---|---|
committer | 2021-07-30 11:49:19 +0000 | |
commit | c9b3ea89ea31773cdb6ee343dc6ea803d0371739 (patch) | |
tree | d6b73bdffaa65452b67a06a64e2e3ebaf136a877 /src/onaptests/steps/onboard/cds.py | |
parent | 7fbd9f67fa43997bc8b030ab24dcc4315493c1b1 (diff) |
CDS resource-resolution test
Issue-ID: TEST-291
Signed-off-by: Michal Jagiello <michal.jagiello@t-mobile.pl>
Change-Id: I441f2486b38481624eeefa18f47840d4c7a3db9d
Diffstat (limited to 'src/onaptests/steps/onboard/cds.py')
-rw-r--r-- | src/onaptests/steps/onboard/cds.py | 38 |
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") |