diff options
author | Michal Jagiello <michal.jagiello@t-mobile.pl> | 2023-03-13 11:00:21 +0000 |
---|---|---|
committer | Michal Jagiello <michal.jagiello@t-mobile.pl> | 2023-03-14 14:25:16 +0000 |
commit | 870ff702088b89549bc21631eb48443fff0bcd71 (patch) | |
tree | 5d52aca25035644bdc2ac687275ed858ba524e7e /mock-so/resources/service_instance.py | |
parent | 98e0e65ba145ab4c85c301118d4a0d02940221ec (diff) |
Migrate mock applications from Orange GitLab
Move from Orange repositories into ONAP one.
Issue-ID: INT-2208
Signed-off-by: Michal Jagiello <michal.jagiello@t-mobile.pl>
Change-Id: I6e165da5144c28a6ff151e02e32f5ae89ce124e3
Diffstat (limited to 'mock-so/resources/service_instance.py')
-rw-r--r-- | mock-so/resources/service_instance.py | 344 |
1 files changed, 344 insertions, 0 deletions
diff --git a/mock-so/resources/service_instance.py b/mock-so/resources/service_instance.py new file mode 100644 index 0000000..60880c2 --- /dev/null +++ b/mock-so/resources/service_instance.py @@ -0,0 +1,344 @@ +"""SO mock instance resources.""" +""" + Copyright 2023 Deutsche Telekom AG, Orange + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +""" +from typing import Callable, Dict, List +from uuid import uuid4 + +import requests +from flask_restful import Resource, request + + +SERVICE_INSTANCES = {} + + +class SoResource(Resource): + """Base SO resource class.""" + + AAI_MOCK_URL = "" + + @classmethod + def set_aai_mock(cls, aai_mock_url: str) -> None: + """Set A&AI mock address. + + Instance resources needs to communicate with A&AI mock + to update it's state. + + Args: + aai_mock_url (str): A&AI mock url + + """ + cls.AAI_MOCK_URL = aai_mock_url + + @classmethod + def reset(cls) -> None: + """Reset SO resource. + + Clean SERVICE_INSTANCES dictionary + and A&AI mock url address. + + """ + global SERVICE_INSTANCES + SERVICE_INSTANCES = {} + cls.AAI_MOCK_URL = "" + + def check_aai_mock_address(method: Callable) -> Callable: + """Decorate method to check if A&AI address is set. + + If A&AI mock address is not set it returns 500 HTTP + response for resource's method + + Args: + method (Callable): method to decorate + + """ + + def decorator(self, *args, **kwargs): + if not self.AAI_MOCK_URL: + return "A&AI mock address not set", 500 + return method(self, *args, **kwargs) + + return decorator + + +class ServiceInstance(SoResource): + """Service instance resource class.""" + + def delete(self, service_instance_id: str) -> Dict[str, Dict[str, str]]: + """Delete service instance. + + Remove service instance data from SERVICE_INSTANCES + dictionary. + + Args: + service_instance_id (str): Service instance id key value + + Returns: + Dict[str, Dict[str, str]]: Deletion request dictionary + + """ + service_instance = SERVICE_INSTANCES[service_instance_id] + requests.delete( + ( + f"{self.AAI_MOCK_URL}/aai/v16/business/customers/customer/" + f"{service_instance['customerId']}/service-subscriptions/service-subscription/" + f"{service_instance['serviceSubscriptionId']}/service-instances/service-instance/" + f"{service_instance_id}" + ) + ) + del SERVICE_INSTANCES[service_instance_id] + return {"requestReferences": {"requestId": str(uuid4())}} + + +class ServiceInstanceList(SoResource): + """Service instances list resource.""" + + @SoResource.check_aai_mock_address + def post(self) -> Dict[str, Dict[str, str]]: + """Create service instance. + + Create service instance data dictionary. + Call request to A&AI mock to create service instance there. + + Returns: + Dict[str, Dict[str, str]]: Creation request dictionary + + """ + instance_id = str(uuid4()) + request_data = request.get_json() + customer_id = request_data["requestDetails"]["subscriberInfo"][ + "globalSubscriberId" + ] + service_subscription_id = request_data["requestDetails"]["requestParameters"][ + "subscriptionServiceType" + ] + instance_name = request_data["requestDetails"]["requestInfo"]["instanceName"] + service_instance = { + "requestId": str(uuid4()), + "instanceId": instance_id, + "customerId": customer_id, + "serviceSubscriptionId": service_subscription_id, + "instanceName": instance_name, + } + requests.post( + ( + f"{self.AAI_MOCK_URL}/aai/v16/business/customers/customer/{customer_id}/" + f"service-subscriptions/service-subscription/{service_subscription_id}/" + "service-instances" + ), + json={ + "service-instance-name": instance_name, + "service-instance-id": instance_id, + }, + ) + SERVICE_INSTANCES[service_instance["instanceId"]] = service_instance + return {"requestReferences": service_instance} + + +class VnfInstance(SoResource): + """Vnf instance resource.""" + + @SoResource.check_aai_mock_address + def delete( + self, service_instance_id: str, vnf_instance_id: str + ) -> Dict[str, Dict[str, str]]: + """Delete vnf instance. + + Remove vnf instanca data from SERVICE_INSTANCES dictionary. + Call DELETE request to A&AI mock. + + Args: + service_instance_id (str): Service instance id key value + vnf_instance_id (str): Vnf instance id key value + + Returns: + Dict[str, Dict[str, str]]: Deletion request dictionary. + + """ + related_service = SERVICE_INSTANCES[service_instance_id] + requests.delete( + ( + f"{self.AAI_MOCK_URL}/aai/v16/business/customers/customer/" + f"{related_service['customerId']}/service-subscriptions/service-subscription/" + f"{related_service['serviceSubscriptionId']}/service-instances/service-instance/" + f"{service_instance_id}/relationship-list" + ) + ) + return {"requestReferences": {"requestId": str(uuid4())}} + + +class VnfInstanceList(SoResource): + """Vnf instances list resource.""" + + @SoResource.check_aai_mock_address + def post(self, service_instance_id: str) -> Dict[str, Dict[str, str]]: + """Create vnf instance. + + Create vnf instance data dictionary. + Call request to A&AI mock to create vnf instance there. + + Returns: + Dict[str, Dict[str, str]]: Creation request dictionary + + """ + instance_id = str(uuid4()) + request_data = request.get_json() + related_instance_id = request_data["requestDetails"]["relatedInstanceList"][0][ + "relatedInstance" + ]["instanceId"] + related_service = SERVICE_INSTANCES[related_instance_id] + requests.post( + ( + f"{self.AAI_MOCK_URL}/aai/v16/business/customers/customer/{related_service['customerId']}/" + f"service-subscriptions/service-subscription/{related_service['serviceSubscriptionId']}/" + f"service-instances/service-instance/{related_instance_id}/relationship-list" + ), + json={ + "related-to": "generic-vnf", + "related-link": f"/aai/v16/network/generic-vnfs/generic-vnf/{instance_id}", + }, + ) + return { + "requestReferences": {"requestId": str(uuid4()), "instanceId": instance_id} + } + + +class VfModuleInstance(SoResource): + """Vf module instance resource class.""" + + @SoResource.check_aai_mock_address + def delete( + self, service_instance_id: str, vnf_instance_id: str, vf_module_instance_id: str + ) -> Dict[str, Dict[str, str]]: + """Delete vf module instance. + + Call DELETE request to A&AI mock to delete vf module instance. + + Args: + service_instance_id (str): Service instance id key value. + vnf_instance_id (str): Vnf instance id key value. + vf_module_instance_id (str): Vf module instance id key value. + + Returns: + Dict[str, Dict[str, str]]: Deletion request dictionary + + """ + requests.delete( + ( + f"{self.AAI_MOCK_URL}/aai/v16/network/generic-vnfs/generic-vnf/" + f"{vnf_instance_id}/vf-modules/{vf_module_instance_id}" + ) + ) + return {"requestReferences": {"requestId": str(uuid4())}} + + +class VfModuleInstanceList(SoResource): + """Vf module instances list resource.""" + + @SoResource.check_aai_mock_address + def post( + self, service_instance_id: str, vnf_instance_id: str + ) -> Dict[str, Dict[str, str]]: + """Create vf module instance. + + Call POST request to A&AI mock to create vf module instance. + + Args: + service_instance_id (str): Service instance id key value + vnf_instance_id (str): Vnf instance id key value + + Returns: + Dict[str, Dict[str, str]]: Creation request dictionary + + """ + instance_id = str(uuid4()) + requests.post( + ( + f"{self.AAI_MOCK_URL}/aai/v16/network/generic-vnfs/generic-vnf/" + f"{vnf_instance_id}/vf-modules" + ), + json={"vf-module-id": instance_id}, + ) + return { + "requestReferences": {"requestId": str(uuid4()), "instanceId": instance_id} + } + + +class NetworkInstance(SoResource): + """Network instance resource.""" + + @SoResource.check_aai_mock_address + def delete( + self, service_instance_id: str, network_instance_id: str + ) -> Dict[str, Dict[str, str]]: + """Delete network instance. + + Remove network instanca data from SERVICE_INSTANCES dictionary. + Call DELETE request to A&AI mock. + + Args: + service_instance_id (str): Service instance id key value + network_instance_id (str): Network instance id key value + + Returns: + Dict[str, Dict[str, str]]: Deletion request dictionary. + + """ + related_service = SERVICE_INSTANCES[service_instance_id] + requests.delete( + ( + f"{self.AAI_MOCK_URL}/aai/v16/business/customers/customer/" + f"{related_service['customerId']}/service-subscriptions/service-subscription/" + f"{related_service['serviceSubscriptionId']}/service-instances/service-instance/" + f"{service_instance_id}/relationship-list" + ) + ) + return {"requestReferences": {"requestId": str(uuid4())}} + + +class NetworkInstanceList(SoResource): + """Network instances list resource.""" + + @SoResource.check_aai_mock_address + def post(self, service_instance_id: str) -> Dict[str, Dict[str, str]]: + """Create network instance. + + Create network instance data dictionary. + Call request to A&AI mock to create network instance there. + + Returns: + Dict[str, Dict[str, str]]: Creation request dictionary + + """ + instance_id = str(uuid4()) + request_data = request.get_json() + related_instance_id = request_data["requestDetails"]["relatedInstanceList"][0][ + "relatedInstance" + ]["instanceId"] + related_service = SERVICE_INSTANCES[related_instance_id] + requests.post( + ( + f"{self.AAI_MOCK_URL}/aai/v16/business/customers/customer/{related_service['customerId']}/" + f"service-subscriptions/service-subscription/{related_service['serviceSubscriptionId']}/" + f"service-instances/service-instance/{related_instance_id}/relationship-list" + ), + json={ + "related-to": "l3-network", + "related-link": f"/aai/v16/network/l3-networks/l3-network/{instance_id}", + }, + ) + return { + "requestReferences": {"requestId": str(uuid4()), "instanceId": instance_id} + } |