aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--requirements.txt4
-rw-r--r--run_basic_sdnc.py18
-rw-r--r--setup.cfg1
-rw-r--r--src/onaptests/configuration/basic_sdnc_settings.py69
-rw-r--r--src/onaptests/configuration/settings.py4
-rw-r--r--src/onaptests/scenario/basic_sdnc.py45
-rw-r--r--src/onaptests/steps/instantiate/sdnc_service.py104
-rw-r--r--src/onaptests/steps/reports_collection.py20
-rw-r--r--tests/test_generate_json.py64
9 files changed, 325 insertions, 4 deletions
diff --git a/requirements.txt b/requirements.txt
index 7866370..c9db44a 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -2,8 +2,8 @@ cryptography==38.0.4
xtesting==0.91.0
avionix>=0.4.5
openstacksdk>=0.61.0
-onapsdk==10.4.3
+onapsdk==10.5.0
jinja2>3
kubernetes>=22.6.0
setuptools==65.3.0
-natural==0.2.0 \ No newline at end of file
+natural==0.2.0
diff --git a/run_basic_sdnc.py b/run_basic_sdnc.py
new file mode 100644
index 0000000..18e6347
--- /dev/null
+++ b/run_basic_sdnc.py
@@ -0,0 +1,18 @@
+import logging.config
+from onapsdk.configuration import settings
+
+from onaptests.scenario.basic_sdnc import BasicSdnc
+import onaptests.utils.exceptions as onap_test_exceptions
+
+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 SDNC")
+
+ basic_sdnc = BasicSdnc(cleanup=settings.CLEANUP_FLAG)
+ try:
+ basic_sdnc.run()
+ basic_sdnc.clean()
+ except onap_test_exceptions.TestConfigurationException:
+ logger.error("Basic SDNC configuration error")
diff --git a/setup.cfg b/setup.cfg
index adc5034..d1483a4 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -59,3 +59,4 @@ xtesting.testcase =
basic_cnf_macro = onaptests.scenario.basic_cnf_macro:BasicCnfMacro
basic_cps = onaptests.scenario.basic_cps:BasicCps
status = onaptests.scenario.status:Status
+ basic_sdnc = onaptests.scenario.basic_sdnc:BasicSdnc
diff --git a/src/onaptests/configuration/basic_sdnc_settings.py b/src/onaptests/configuration/basic_sdnc_settings.py
new file mode 100644
index 0000000..e6bbdcc
--- /dev/null
+++ b/src/onaptests/configuration/basic_sdnc_settings.py
@@ -0,0 +1,69 @@
+from .settings import *
+
+CLEANUP_FLAG = True
+
+SERVICE_NAME = "Basic SDNC test"
+
+SERVICE_COMPONENTS = "SDNC"
+
+SERVICE_ID = "pythonsdk-tests-service-01"
+
+SERVICE_STATUS = {
+ "rpc-action": "activate",
+ "response-message": "test-message-1",
+ "response-timestamp": "2023-05-09T13:14:30.540Z",
+ "request-status": "synccomplete",
+ "final-indicator": "Y",
+ "action": "CreateVnfInstance",
+ "rpc-name": "vnf-topology-operation",
+ "response-code": "200"
+}
+
+SERVICE_CHANGED_STATUS = {
+ "rpc-action": "activate",
+ "response-message": "changed-test-message-1",
+ "response-timestamp": "2023-05-09T13:14:30.540Z",
+ "request-status": "synccomplete",
+ "final-indicator": "Y",
+ "action": "CreateVnfInstance",
+ "rpc-name": "vnf-topology-operation",
+ "response-code": "200"
+}
+
+SERVICE_DATA = {
+ "service-level-oper-status": {
+ "last-rpc-action": "assign",
+ "last-action": "CreateServiceInstance",
+ "order-status": "Created"
+ },
+ "service-request-input": {
+ "service-input-parameters": {
+ "param": [
+ {
+ "name": "orchestrator",
+ "value": "multicloud"
+ }
+ ]
+ },
+ "service-instance-name": "gnb-93100001"
+ }
+}
+
+SERVICE_CHANGED_DATA = {
+ "service-level-oper-status": {
+ "last-rpc-action": "assign",
+ "last-action": "CreateServiceInstance",
+ "order-status": "Created"
+ },
+ "service-request-input": {
+ "service-input-parameters": {
+ "param": [
+ {
+ "name": "orchestrator",
+ "value": "multicloud"
+ }
+ ]
+ },
+ "service-instance-name": "gnb-93100002"
+ }
+}
diff --git a/src/onaptests/configuration/settings.py b/src/onaptests/configuration/settings.py
index 74bc840..436e82c 100644
--- a/src/onaptests/configuration/settings.py
+++ b/src/onaptests/configuration/settings.py
@@ -40,7 +40,9 @@ LOG_CONFIG = {
CLEANUP_FLAG = False
SDC_CLEANUP = False
-REPORTING_FILE_PATH = "/tmp/reporting.html"
+REPORTING_FILE_DIRECTORY = "/tmp/"
+HTML_REPORTING_FILE_NAME = "reporting.html"
+JSON_REPORTING_FILE_NAME = "reporting.json"
K8S_REGION_TYPE = "k8s"
TILLER_HOST = "localhost"
K8S_CONFIG = None # None means it will use default config (~/.kube/config)
diff --git a/src/onaptests/scenario/basic_sdnc.py b/src/onaptests/scenario/basic_sdnc.py
new file mode 100644
index 0000000..d791080
--- /dev/null
+++ b/src/onaptests/scenario/basic_sdnc.py
@@ -0,0 +1,45 @@
+import logging
+import time
+from xtesting.core import testcase
+
+from onapsdk.configuration import settings
+from onapsdk.exceptions import SDKException
+
+from onaptests.steps.instantiate.sdnc_service import UpdateSdncService
+from onaptests.utils.exceptions import OnapTestException
+
+
+class BasicSdnc(testcase.TestCase):
+ """Create SDNC service.
+ Check and delete the service.
+ """
+
+ __logger = logging.getLogger()
+
+ def __init__(self, **kwargs):
+ """Init Basic SDNC use case."""
+ if "case_name" not in kwargs:
+ kwargs["case_name"] = 'basic_SDNC'
+ super().__init__(**kwargs)
+ self.__logger.debug("Basic SDNC init started")
+ self.test = UpdateSdncService(cleanup=settings.CLEANUP_FLAG)
+
+ def run(self):
+ """Run basic SDNC test."""
+ self.start_time = time.time()
+ try:
+ self.test.execute()
+ self.test.cleanup()
+ 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):
+ """Generate report."""
+ self.test.reports_collection.generate_report()
diff --git a/src/onaptests/steps/instantiate/sdnc_service.py b/src/onaptests/steps/instantiate/sdnc_service.py
new file mode 100644
index 0000000..c5308d6
--- /dev/null
+++ b/src/onaptests/steps/instantiate/sdnc_service.py
@@ -0,0 +1,104 @@
+from onapsdk.sdnc.services import Service
+from onapsdk.configuration import settings
+from onapsdk.exceptions import APIError
+
+from ..base import BaseStep
+
+from onaptests.utils.exceptions import OnapTestException
+
+
+class ServiceCreateStep(BaseStep):
+ """Service creation step."""
+
+ def __init__(self, service: Service = None, cleanup: bool = False):
+ """Initialize step."""
+ super().__init__(cleanup=cleanup)
+ self.service = service
+
+ @property
+ def description(self) -> str:
+ """Step description."""
+ return "Create SDNC service."
+
+ @property
+ def component(self) -> str:
+ """Component name."""
+ return "SDNC"
+
+ @BaseStep.store_state
+ def execute(self):
+ """Create service at SDNC."""
+ self._logger.info("Create new service instance in SDNC by GR-API")
+ super().execute()
+ try:
+ self.service = Service(
+ service_instance_id=settings.SERVICE_ID,
+ service_status=settings.SERVICE_STATUS,
+ service_data=settings.SERVICE_DATA
+ )
+ self.service.create()
+ self._logger.info("SDNC service is created.")
+ except APIError:
+ raise OnapTestException("SDNC service creation failed.")
+
+ @BaseStep.store_state()
+ def cleanup(self) -> None:
+ """Cleanup Service."""
+ if self.service is not None:
+ self.service.delete()
+ self._logger.info("SDNC service is deleted.")
+ super().cleanup()
+
+
+class UpdateSdncService(BaseStep):
+ """Service update step.
+
+ The step needs in an existing SDNC service as a prerequisite.
+ """
+
+ def __init__(self, cleanup=False):
+ """Initialize step.
+
+ Sub steps:
+ - ServiceCreateStep.
+ """
+ super().__init__(cleanup=cleanup)
+ self.add_step(ServiceCreateStep(cleanup=cleanup))
+
+ @property
+ def description(self) -> str:
+ """Step description.
+
+ Used for reports
+
+ Returns:
+ str: Step description
+
+ """
+ return "Update SDNC service"
+
+ @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 "SDNC"
+
+ @BaseStep.store_state
+ def execute(self):
+ self._logger.info("Get existing SDNC service instance and update it over GR-API")
+ super().execute()
+ try:
+ service = Service.get(settings.SERVICE_ID)
+ service.service_status = settings.SERVICE_CHANGED_STATUS
+ service.service_data = settings.SERVICE_CHANGED_DATA
+ service.update()
+ self._logger.info("SDNC service update is checked.")
+ except APIError:
+ raise OnapTestException("SDNC service update is failed.")
diff --git a/src/onaptests/steps/reports_collection.py b/src/onaptests/steps/reports_collection.py
index 4b7f79b..0e5076f 100644
--- a/src/onaptests/steps/reports_collection.py
+++ b/src/onaptests/steps/reports_collection.py
@@ -1,5 +1,7 @@
from dataclasses import dataclass
from enum import Enum
+import json
+from pathlib import Path
from typing import List
from jinja2 import Environment, FileSystemLoader, select_autoescape
from onapsdk.configuration import settings
@@ -83,4 +85,20 @@ class ReportsCollection:
details=details,
components=components,
log_path="./pythonsdk.debug.log").dump(
- settings.REPORTING_FILE_PATH)
+ str(Path(settings.REPORTING_FILE_DIRECTORY).joinpath(settings.HTML_REPORTING_FILE_NAME)))
+
+ report_dict = {
+ 'usecase': usecase,
+ 'details': details,
+ 'components': components,
+ 'steps': [
+ {
+ 'description': step_report.step_description,
+ 'status': step_report.step_execution_status.value,
+ 'duration': step_report.step_execution_duration
+ }
+ for step_report in reversed(self.report)
+ ]
+ }
+ with (Path(settings.REPORTING_FILE_DIRECTORY).joinpath(settings.JSON_REPORTING_FILE_NAME)).open('w') as file:
+ json.dump(report_dict, file, indent=4) \ No newline at end of file
diff --git a/tests/test_generate_json.py b/tests/test_generate_json.py
new file mode 100644
index 0000000..b5bc48c
--- /dev/null
+++ b/tests/test_generate_json.py
@@ -0,0 +1,64 @@
+import unittest
+import os
+import json
+from unittest import mock
+from onaptests.steps.reports_collection import ReportsCollection, Report, ReportStepStatus
+
+
+class TestReportsCollection(unittest.TestCase):
+
+ def setUp(self):
+ self.collection = ReportsCollection()
+
+ @mock.patch("onaptests.steps.reports_collection.settings")
+ def test_generate_report_json(self, settings):
+ settings.SERVICE_NAME = "Status Check"
+ settings.SERVICE_DETAILS = "Checks status of all k8s resources in the selected namespace"
+ settings.SERVICE_COMPONENTS = "ALL"
+ settings.REPORTING_FILE_DIRECTORY = "/tmp/"
+ settings.HTML_REPORTING_FILE_NAME = "reporting.html"
+ settings.JSON_REPORTING_FILE_NAME = "reporting.json"
+
+ self.collection.put(Report("Step 1", ReportStepStatus.PASS, 10.0))
+ self.collection.put(Report("Step 2", ReportStepStatus.FAIL, 5.0))
+ self.collection.put(Report("Step 3", ReportStepStatus.NOT_EXECUTED, 0.0))
+ self.collection.put(Report("Step 10", ReportStepStatus.NOT_EXECUTED, 0.0))
+ self.collection.put(Report("Step 12", ReportStepStatus.PASS, 20.0))
+ self.collection.put(Report("Step 21", ReportStepStatus.FAIL, 15.0))
+
+ report_dict = self.collection.generate_report()
+ report_json_path = os.path.join(settings.REPORTING_FILE_DIRECTORY, settings.JSON_REPORTING_FILE_NAME)
+ self.assertTrue(os.path.exists(report_json_path))
+
+ with open(report_json_path, 'r') as file:
+ report_dict = json.load(file)
+
+ self.assertEqual(report_dict['usecase'], 'Status Check')
+ self.assertEqual(report_dict['details'], 'Checks status of all k8s resources in the selected namespace')
+ self.assertEqual(report_dict['components'], 'ALL')
+ self.assertEqual(len(report_dict['steps']), 6)
+ step1 = report_dict['steps'][0]
+ step2 = report_dict['steps'][1]
+ step3 = report_dict['steps'][2]
+ step10 = report_dict['steps'][3]
+ step12 = report_dict['steps'][4]
+ step21 = report_dict['steps'][5]
+ self.assertEqual(step1['description'], 'Step 1')
+ self.assertEqual(step1['duration'], 10.0)
+ self.assertEqual(step2['description'], 'Step 2')
+ self.assertEqual(step2['duration'], 5.0)
+ self.assertEqual(step3['description'], 'Step 3')
+ self.assertEqual(step3['duration'], 0.0)
+ self.assertEqual(step10['description'], 'Step 10')
+ self.assertEqual(step10['duration'], 0.0)
+ self.assertEqual(step12['description'], 'Step 12')
+ self.assertEqual(step12['duration'], 20.0)
+ self.assertEqual(step21['description'], 'Step 21')
+ self.assertEqual(step21['duration'], 15.0)
+ report_json_path = os.path.join(settings.REPORTING_FILE_DIRECTORY, settings.JSON_REPORTING_FILE_NAME)
+ self.assertTrue(os.path.exists(report_json_path))
+ self.assertEqual(report_json_path,'/tmp/reporting.json')
+ os.remove(report_json_path)
+
+if __name__ == '__main__':
+ unittest.main()