class SOHandler(BaseHTTPRequestHandler):
- def __init__(self, expected_requests, expected_responses, *args, **kwargs):
+ def __init__(self, expected_requests, expected_responses, requests, responses, *args, **kwargs):
self._expected_requests = expected_requests
self._expected_responses = expected_responses
+ self._requests = requests["requests"]
+ self._responses = responses["responses"]
+ self._known_endpoints = self._get_known_endpoints()
super().__init__(*args, **kwargs)
def do_POST(self):
return
def do_GET(self):
- logging.info(
- 'GET called. Expected GET REQUEST: ' + json.dumps(
- self._expected_requests["get"]) + '\nExpected GET response: ' +
- json.dumps(self._expected_responses["get"]))
- self.send_response(200)
- self._set_headers()
-
- self.wfile.write(json.dumps(self._expected_responses["get"]).encode("utf-8"))
- return self._expected_responses["get"]
+ print(self._known_endpoints)
+ if self._does_path_exist_as_endpoint():
+ expected_response = self._get_response_by_path()
+ if expected_response:
+ self.send_response(expected_response["responseCode"])
+ self._set_headers()
+ self.wfile.write(json.dumps(expected_response["body"]).encode("utf-8"))
+ else:
+ response_body = "{\"message:\" \"no response for endpoint\"}"
+ self.send_response(400)
+ self._set_headers()
+ self.wfile.write(json.dumps(response_body).encode("utf-8"))
+ else:
+ logging.info('GET called. Expected GET REQUEST: '
+ + json.dumps(self._expected_requests["get"])
+ + '\nExpected GET response: '
+ + json.dumps(self._expected_responses["get"]))
+ self.send_response(200)
+ self._set_headers()
+ self.wfile.write(json.dumps(self._expected_responses["get"]).encode("utf-8"))
def do_PUT(self):
request_body_json = self._get_request_body()
def _apply_expected_data(self, request_body_json):
if self.path == '/setResponse':
logging.info("IN PUT /setResponse: " + str(request_body_json))
- print("TYPE: %s and text: %s", type(request_body_json), str(request_body_json))
self._expected_responses.update(request_body_json)
- print("TYPE: %s", type(request_body_json))
elif self.path == '/setRequest':
logging.info("IN PUT /setRequest: " + str(request_body_json))
self._expected_requests.update(request_body_json)
self.send_header('Content-Type', 'application/json')
self.end_headers()
+ def _get_response_by_path(self):
+ for response in self._responses:
+ if response["path"] == self.path:
+ return response
+
+ def _does_path_exist_as_endpoint(self):
+ does_exist = False
+ for ep in self._known_endpoints:
+ if ep == self.path:
+ does_exist = True
+ break
+ return does_exist
+
+ def _create_path_from_request(self, request):
+ base_uri = request["path"]
+ query_params = request["queryParams"]
+ return base_uri + self._get_params_uri(query_params)
+
+ def _get_known_endpoints(self):
+ endpoints = []
+ for request in self._requests:
+ endpoints.append(self._create_path_from_request(request))
+ return endpoints
+
+ @staticmethod
+ def _get_params_uri(query_params):
+ params_uri = "?"
+ for parameter in query_params:
+ params_uri += parameter + '=' + query_params.get(parameter, '') + '&'
+ return params_uri[:-1]
+
class JsonFileToDictReader(object):
def init_so_simulator():
+ requests = JsonFileToDictReader.read_expected_test_data("test_data_assets/requests.json")
+ responses = JsonFileToDictReader.read_expected_test_data("test_data_assets/responses.json")
+
expected_so_requests = JsonFileToDictReader.read_expected_test_data(argv[1])
expected_so_responses = JsonFileToDictReader.read_expected_test_data(argv[2])
logging.basicConfig(level=logging.INFO)
- handler = partial(SOHandler, expected_so_requests, expected_so_responses)
+ handler = partial(SOHandler, expected_so_requests, expected_so_responses, requests, responses)
handler.protocol_version = "HTTP/1.0"
httpd = HTTPServer(('', DEFAULT_PORT), handler)
logging.info("serving on: " + str(httpd.socket.getsockname()))
--- /dev/null
+{
+ "responses": [
+ {
+ "responseCode": 200,
+ "path": "/aai/v13/service-design-and-creation/models?depth=2&model-invariant-id=88a71d72-ec80-4357-808e-f288823cb353",
+ "body": {
+ "model": [
+ {
+ "model-invariant-id": "88a71d72-ec80-4357-808e-f288823cb353",
+ "model-type": "resource",
+ "resource-version": "1549550682576",
+ "model-vers": {
+ "model-ver": [
+ {
+ "model-version-id": "d2dcf256-2687-4631-9e36-1b7fc352b7bf",
+ "model-name": "HealthVF",
+ "model-version": "1.0",
+ "model-description": "HealthVSP",
+ "resource-version": "1549550682585",
+ "model-elements": {
+ "model-element": [
+ {
+ "model-element-uuid": "ad8c74b6-d2ff-48a7-8fcb-4cf653efc1f9",
+ "new-data-del-flag": "T",
+ "cardinality": "unbounded",
+ "resource-version": "1549550682592",
+ "relationship-list": {
+ "relationship": [
+ {
+ "related-to": "model-ver",
+ "relationship-label": "org.onap.relationships.inventory.IsA",
+ "related-link": "/aai/v13/service-design-and-creation/models/model/acc6edd8-a8d4-4b93-afaa-0994068be14c/model-vers/model-ver/93a6166f-b3d5-4f06-b4ba-aed48d009ad9",
+ "relationship-data": [
+ {
+ "relationship-key": "model.model-invariant-id",
+ "relationship-value": "acc6edd8-a8d4-4b93-afaa-0994068be14c"
+ }
+ ],
+ "related-to-property": [
+ {
+ "property-key": "model-ver.model-name",
+ "property-value": "generic-vnf"
+ }
+ ]
+ }
+ ]
+ }
+ }
+ ]
+ },
+ "relationship-list": {
+ "relationship": [
+ {
+ "related-to": "model-element",
+ "relationship-label": "org.onap.relationships.inventory.IsA",
+ "related-link": "/aai/v13/service-design-and-creation/models/model/4b5158b9-a0d5-4aeb-90a8-474bc2ccd8af/model-vers/model-ver/ba463093-6f2c-4d22-aa8d-fb4615660db2/model-elements/model-element/57d23b71-d5a0-484d-ac7c-1b94e9a1563e/model-elements/model-element/5d290a6a-3b35-46bb-a2d7-1b9c9f520c39",
+ "relationship-data": [
+ {
+ "relationship-key": "model.model-invariant-id",
+ "relationship-value": "4b5158b9-a0d5-4aeb-90a8-474bc2ccd8af"
+ }
+ ]
+ }
+ ]
+ }
+ }
+ ]
+ }
+ }
+ ]
+ }
+ },
+ {
+ "responseCode": 200,
+ "path": "/aai/v../business/customers/customer/MSO_1610_ST?depth=2",
+ "body": {
+ "global-customer-id": "MSO_1610_ST",
+ "subscriber-name": "MSO_1610_ST",
+ "subscriber-type": "INFRA",
+ "resource-version": "1549008901724",
+ "service-subscriptions": {
+ "service-subscription": [
+ {
+ "service-type": "gNB",
+ "resource-version": "1549008901954",
+ "service-instances": {
+ "service-instance": [
+ {
+ "service-instance-id": "40003c27-2876-4f3a-8a23-ef1ad7ef665a",
+ "service-instance-name": "AwServicePreload3",
+ "service-type": "vAwServiceType",
+ "environment-context": "General_Revenue-Bearing",
+ "workload-context": "Production",
+ "model-invariant-id": "33ebe1a9-1df3-4c35-b8a0-cd0504159511",
+ "model-version-id": "41658d87-b37c-4237-9a56-d27d1e430da6",
+ "resource-version": "1550761112266",
+ "orchestration-status": "Active",
+ "relationship-list": {
+ "relationship": [
+ {
+ "related-to": "project",
+ "relationship-label": "org.onap.relationships.inventory.Uses",
+ "related-link": "/aai/v13/business/projects/project/AwTestProject",
+ "relationship-data": [
+ {
+ "relationship-key": "project.project-name",
+ "relationship-value": "AwTestProject"
+ }
+ ]
+ }
+ ]
+ }
+ }
+ ]
+ },
+ "relationship-list": {
+ "relationship": [
+ {
+ "related-to": "tenant",
+ "relationship-label": "org.onap.relationships.inventory.Uses",
+ "related-link": "/aai/v13/cloud-infrastructure/cloud-regions/cloud-region/CloudOwner/RegionOne/tenants/tenant/982c540f6e69488eb6be5664255e00c0",
+ "relationship-data": [
+ {
+ "relationship-key": "cloud-region.cloud-owner",
+ "relationship-value": "CloudOwner"
+ }
+ ],
+ "related-to-property": [
+ {
+ "property-key": "tenant.tenant-name",
+ "property-value": "onap-wro"
+ }
+ ]
+ }
+ ]
+ }
+ }
+ ]
+ }
+ }
+ },
+ {
+ "responseCode": 200,
+ "path": "/aai/v13/query?format=simple",
+ "body": {
+ "results": [
+ {
+ "id": "385160",
+ "node-type": "service-instance",
+ "url": "/aai/v13/business/customers/customer/Demonstration/service-subscriptions/service-subscription/vLB/service-instances/service-instance/12965035-a690-400c-bf18-211a5b4710c8",
+ "properties": {
+ "service-instance-id": "12965035-a690-400c-bf18-211a5b4710c8",
+ "service-instance-name": "ws-service-02",
+ "service-type": "vAwServiceType",
+ "environment-context": "General_Revenue-Bearing",
+ "workload-context": "Production",
+ "model-invariant-id": "33ebe1a9-1df3-4c35-b8a0-cd0504159511",
+ "model-version-id": "cb8501da-6e03-40e0-900c-ef0df30d1183",
+ "resource-version": "1551434658559",
+ "selflink": "restconf/config/GENERIC-RESOURCE-API:services/service/12965035-a690-400c-bf18-211a5b4710c8/service-data/service-topology/",
+ "orchestration-status": "Active"
+ },
+ "related-to": [
+ {
+ "id": "200768",
+ "relationship-label": "org.onap.relationships.inventory.Uses",
+ "node-type": "project",
+ "url": "/aai/v13/business/projects/project/Project-Demonstration"
+ }
+ ]
+ }
+ ]
+ }
+ },
+ {
+ "responseCode": 200,
+ "path": "/workflowSpecifications/v1/workflows?vnfModelVersionId=103b4a1b-4a15-4559-a019-1ff132180c7c",
+ "body": {
+ "workflowSpecificationList": [
+ {
+ "workflowSpecification": {
+ "artifactInfo": {
+ "artifactType": "workflow",
+ "artifactUuid": "ab6478e4-ea33-3346-ac12-ab121484a333",
+ "artifactName": "inPlaceSoftwareUpdate-1_0.bpmn",
+ "artifactVersion": "1.0",
+ "artifactDescription": "xyz xyz",
+ "workflowName": "inPlaceSoftwareUpdate",
+ "operationName": "inPlaceSoftwareUpdate",
+ "workflowSource": "sdc",
+ "workflowResourceTarget": "vnf"
+ },
+ "activitySequence": [
+ {
+ "name": "VNFQuiesceTrafficActivity",
+ "description": "Activity to QuiesceTraffic on VNF"
+ }
+ ],
+ "workflowInputParameters": [
+ {
+ "label": "Cloud Owner",
+ "inputType": "text",
+ "required": true,
+ "validation": [
+ {
+ "maxLength": "7",
+ "allowableChars": "[a-zA-Z0-9]*"
+ }
+ ],
+ "soFieldName": "cloudOwner",
+ "soPayloadLocation": "cloudConfiguration"
+ }
+ ]
+ }
+ }
+ ]
+ }
+ }
+ ]
+}
\ No newline at end of file
Code Review / integration / csit.git / commitdiff