diff options
Diffstat (limited to 'mock-aai')
| -rw-r--r-- | mock-aai/Dockerfile | 11 | ||||
| -rw-r--r-- | mock-aai/README.md | 2 | ||||
| -rw-r--r-- | mock-aai/app.py | 154 | ||||
| -rw-r--r-- | mock-aai/requirements.txt | 1 | ||||
| -rw-r--r-- | mock-aai/resources/__init__.py | 16 | ||||
| -rw-r--r-- | mock-aai/resources/cloud_region.py | 193 | ||||
| -rw-r--r-- | mock-aai/resources/complex.py | 65 | ||||
| -rw-r--r-- | mock-aai/resources/customer.py | 325 | ||||
| -rw-r--r-- | mock-aai/resources/network.py | 61 | ||||
| -rw-r--r-- | mock-aai/resources/vnf.py | 108 |
10 files changed, 936 insertions, 0 deletions
diff --git a/mock-aai/Dockerfile b/mock-aai/Dockerfile new file mode 100644 index 0000000..d558a0d --- /dev/null +++ b/mock-aai/Dockerfile @@ -0,0 +1,11 @@ +FROM python:3.8-alpine + +COPY . /app + +WORKDIR /app + +RUN pip install -r requirements.txt + +ENTRYPOINT ["python"] + +CMD ["app.py"] diff --git a/mock-aai/README.md b/mock-aai/README.md new file mode 100644 index 0000000..5217b76 --- /dev/null +++ b/mock-aai/README.md @@ -0,0 +1,2 @@ +# mock-aai + diff --git a/mock-aai/app.py b/mock-aai/app.py new file mode 100644 index 0000000..408d3e4 --- /dev/null +++ b/mock-aai/app.py @@ -0,0 +1,154 @@ +"""A&AI mock application.""" +""" + 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 flask import Flask +from flask_restful import Resource, Api + +from resources.cloud_region import ( + CloudRegion, + CloudRegionList, + CloudRegionRelationshipList, + CloudRegionRelationship, + Tenant, + TenantList, +) +from resources.complex import Complex, ComplexList +from resources.customer import ( + Customer, + CustomerList, + ServiceSubscription, + ServiceSubscriptionInstance, + ServiceSubscriptionList, + ServiceSubscriptionRelationship, + ServiceSubscriptionRelationshipList, + ServiceSubscriptionInstanceList, + ServiceSubscriptionInstanceRelationshipList, +) +from resources.network import Network +from resources.vnf import Vnf, VfModule, VfModuleList + +app = Flask(__name__) +api = Api(app) + +API_VERSIONS_SUPPORTED = ["v16", "v20", "v23", "v26"] + +def generate_urls_for_all_versions(endpoint_pattern: str) -> list: + """Helper function. + + Generates list of endpoints by replacing <version> pattern for each supported API version. + + Args: + endpoint_pattern (str): url pattern with <version> tag to be replaced + + Returns: + list: List of string values reporesenting endpoints that are currently supported + + """ + return [endpoint_pattern.replace("<version>", api_version) for api_version in API_VERSIONS_SUPPORTED] + +@app.route("/reset") +def reset() -> str: + """Reset endpoint. + + Reset all resources. + + Returns: + str: Empty string, it has to returns anything + + """ + CloudRegion.reset() + Complex.reset() + Customer.reset() + Network.reset() + Vnf.reset() + return "" + + +# Cloud region resource +api.add_resource(CloudRegionList, *generate_urls_for_all_versions("/aai/<version>/cloud-infrastructure/cloud-regions")) +api.add_resource( + CloudRegion, + *generate_urls_for_all_versions("/aai/<version>/cloud-infrastructure/cloud-regions/cloud-region/<cloud_owner>/<cloud_region_id>"), +) +api.add_resource( + CloudRegionRelationshipList, + *generate_urls_for_all_versions("/aai/<version>/cloud-infrastructure/cloud-regions/cloud-region/<cloud_owner>/<cloud_region_id>/relationship-list"), +) +api.add_resource( + CloudRegionRelationship, + *generate_urls_for_all_versions("/aai/<version>/cloud-infrastructure/cloud-regions/cloud-region/<cloud_owner>/<cloud_region_id>/relationship-list/relationship"), +) +api.add_resource( + TenantList, + *generate_urls_for_all_versions("/aai/<version>/cloud-infrastructure/cloud-regions/cloud-region/<cloud_owner>/<cloud_region_id>/tenants"), +) +api.add_resource( + Tenant, + *generate_urls_for_all_versions("/aai/<version>/cloud-infrastructure/cloud-regions/cloud-region/<cloud_owner>/<cloud_region_id>/tenants/tenant/<tenant_id>"), +) +# Complex resource +api.add_resource(ComplexList, *generate_urls_for_all_versions("/aai/<version>/cloud-infrastructure/complexes")) +api.add_resource( + Complex, *generate_urls_for_all_versions("/aai/<version>/cloud-infrastructure/complexes/complex/<physical_location_id>") +) +# Customer resource +api.add_resource(CustomerList, *generate_urls_for_all_versions("/aai/<version>/business/customers")) +api.add_resource(Customer, *generate_urls_for_all_versions("/aai/<version>/business/customers/customer/<global_customer_id>")) +api.add_resource( + ServiceSubscriptionList, + *generate_urls_for_all_versions("/aai/<version>/business/customers/customer/<global_customer_id>/service-subscriptions"), +) +api.add_resource( + ServiceSubscription, + *generate_urls_for_all_versions("/aai/<version>/business/customers/customer/<global_customer_id>/service-subscriptions/service-subscription/<service_type>"), +) +api.add_resource( + ServiceSubscriptionRelationshipList, + *generate_urls_for_all_versions("/aai/<version>/business/customers/customer/<global_customer_id>/service-subscriptions/service-subscription/<service_type>/relationship-list"), +) +api.add_resource( + ServiceSubscriptionRelationship, + *generate_urls_for_all_versions("/aai/<version>/business/customers/customer/<global_customer_id>/service-subscriptions/service-subscription/<service_type>/relationship-list/relationship"), +) +api.add_resource( + ServiceSubscriptionInstance, + *generate_urls_for_all_versions("/aai/<version>/business/customers/customer/<global_customer_id>/service-subscriptions/service-subscription/<service_type>/service-instances/service-instance/<service_instance_id>"), +) +api.add_resource( + ServiceSubscriptionInstanceList, + *generate_urls_for_all_versions("/aai/<version>/business/customers/customer/<global_customer_id>/service-subscriptions/service-subscription/<service_type>/service-instances"), +) +api.add_resource( + ServiceSubscriptionInstanceRelationshipList, + *generate_urls_for_all_versions("/aai/<version>/business/customers/customer/<global_customer_id>/service-subscriptions/service-subscription/<service_type>/service-instances/service-instance/<service_instance_id>/relationship-list"), +) +# VNF resource +api.add_resource(Vnf, *generate_urls_for_all_versions("/aai/<version>/network/generic-vnfs/generic-vnf/<vnf_instance_id>")) +api.add_resource( + VfModule, + *generate_urls_for_all_versions("/aai/<version>/network/generic-vnfs/generic-vnf/<vnf_instance_id>/vf-modules/<vf_module_instance_id>"), +) +api.add_resource( + VfModuleList, + *generate_urls_for_all_versions("/aai/<version>/network/generic-vnfs/generic-vnf/<vnf_instance_id>/vf-modules"), +) +api.add_resource( + Network, *generate_urls_for_all_versions("/aai/<version>/network/l3-networks/l3-network/<network_instance_id>") +) + + +if __name__ == "__main__": + app.run(host="0.0.0.0", debug=True) diff --git a/mock-aai/requirements.txt b/mock-aai/requirements.txt new file mode 100644 index 0000000..04f7256 --- /dev/null +++ b/mock-aai/requirements.txt @@ -0,0 +1 @@ +Flask-RESTful==0.3.9
\ No newline at end of file diff --git a/mock-aai/resources/__init__.py b/mock-aai/resources/__init__.py new file mode 100644 index 0000000..bbc4e6a --- /dev/null +++ b/mock-aai/resources/__init__.py @@ -0,0 +1,16 @@ +"""A&AI mock resources package.""" +""" + 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. +""" diff --git a/mock-aai/resources/cloud_region.py b/mock-aai/resources/cloud_region.py new file mode 100644 index 0000000..aa5098d --- /dev/null +++ b/mock-aai/resources/cloud_region.py @@ -0,0 +1,193 @@ +"""A&AI CloudRegion mock module.""" +""" + 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 Dict, List, Tuple + +from flask_restful import Resource, request + +CLOUD_REGIONS = {} + + +class CloudRegion(Resource): + """Cloud region resource.""" + + def get(self, cloud_owner: str, cloud_region_id: str) -> Dict[str, str]: + """Get cloud region. + + Get cloud region from CLOUD_REGIONS dictionary. + + Args: + cloud_owner (str): cloud owner key value + cloud_region_id (str): cloud region id key value + + Returns: + Dict[str, str]: Cloud region dictionary + + """ + return CLOUD_REGIONS[cloud_owner][cloud_region_id] + + def put(self, cloud_owner: str, cloud_region_id: str) -> Tuple[str, int]: + """Cloud region resource put method. + + Add cloud region data into CLOUD_REGIONS dictionary. + + Args: + cloud_owner (str): Cloud owner key value + cloud_region_id (str): Cloud region id key value + + Returns: + Tuple[str, int]: Response tuple. First element is a response body, + the second one is HTTP response code. + """ + CLOUD_REGIONS.update({cloud_owner: {cloud_region_id: request.get_json()}}) + return "", 201 + + @staticmethod + def reset() -> None: + """Reset Cloud region resource. + + Clean CLOUD_REGIONS dictionary + + """ + global CLOUD_REGIONS + CLOUD_REGIONS = {} + + +class CloudRegionList(Resource): + """List of cloud regions resource.""" + + def get(self): + """Get the list of cloud regions. + + Return data from CLOUD_REGIONS dictionary. + + Returns: + Dict[str, List]: Cloud regions dictionary + + """ + return { + "cloud-region": [ + data + for cloud_owner, cloud_owner_dict in CLOUD_REGIONS.items() + for cloud_owner, data in cloud_owner_dict.items() + ] + } + + +class CloudRegionRelationship(Resource): + """Cloud region relationship resource.""" + + def put(self, cloud_owner: str, cloud_region_id: str): + """Cloud region relationship resource put method. + + Add cloud region relationship data into CLOUD_REGIONS dictionary. + + Args: + cloud_owner (str): Cloud owner key value + cloud_region_id (str): Cloud region id key value + + """ + try: + CLOUD_REGIONS[cloud_owner][cloud_region_id]["relationships"].apped(request.get_json()) + except KeyError: + CLOUD_REGIONS[cloud_owner][cloud_region_id]["relationships"] = [request.get_json()] + + +class CloudRegionRelationshipList(Resource): + """List of cloud region relationships resource.""" + + def get(self, cloud_owner: str, cloud_region_id: str) -> Dict[str, List]: + """Get the list of cloud region relationships. + + Return data from CLOUD_REGIONS dictionary. + + Args: + cloud_owner (str): Cloud owner key value + cloud_region_id (str): Cloud region id key value + + Returns: + Dict[str, List]: Cloud region relationships dictionary + + """ + try: + return {"relationship": CLOUD_REGIONS[cloud_owner][cloud_region_id]["relationships"]} + except KeyError: + return {"relationship": []} + + +class Tenant(Resource): + """Cloud region tenant resource.""" + + def put(self, cloud_owner: str, cloud_region_id: str, tenant_id: str) -> None: + """Cloud region tenant resource put method. + + Add cloud region tenant data into CLOUD_REGIONS dictionary. + + Args: + cloud_owner (str): Cloud owner key value + cloud_region_id (str): Cloud region id key value + + """ + try: + CLOUD_REGIONS[cloud_owner][cloud_region_id]["tenants"].update( + {tenant_id: request.get_json()} + ) + except KeyError: + CLOUD_REGIONS[cloud_owner][cloud_region_id]["tenants"] = {tenant_id: request.get_json()} + + def get(self, cloud_owner: str, cloud_region_id: str, tenant_id: str) -> Dict[str, str]: + """Get cloud region tenant. + + Get cloud region tenant from CLOUD_REGIONS dictionary. + + Args: + cloud_owner (str): cloud owner key value + cloud_region_id (str): cloud region id key value + + Returns: + Dict[str, str]: Cloud region tenant dictionary + + """ + try: + return CLOUD_REGIONS[cloud_owner][cloud_region_id]["tenants"][tenant_id] + except KeyError: + return "", 404 + + +class TenantList(Resource): + """List of tenants resource.""" + + def get(self, cloud_owner: str, cloud_region_id: str) -> Dict[str, List]: + """Get the list of cloud region tenants. + + Return data from CLOUD_REGIONS dictionary. + + Args: + cloud_owner (str): Cloud owner key value + cloud_region_id (str): Cloud region id key value + + Returns: + Dict[str, List]: Cloud region tenants dictionary + + """ + return { + "tenant": [ + data + for tenant_id, data in CLOUD_REGIONS[cloud_owner][cloud_region_id] + .get("tenants", {}) + .items() + ] + } diff --git a/mock-aai/resources/complex.py b/mock-aai/resources/complex.py new file mode 100644 index 0000000..9185221 --- /dev/null +++ b/mock-aai/resources/complex.py @@ -0,0 +1,65 @@ +"""A&AI Complex mock module.""" +""" + 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 Dict, List + +from flask_restful import reqparse, Resource, request + +COMPLEXES = {} + +parser = reqparse.RequestParser() + + +class Complex(Resource): + """Complex resource class.""" + + def put(self, physical_location_id: str): + """Complex resource put method. + + Add complex data sent in JSON to COMPLEXES dictionary. + + Args: + physical_location_id (str): Complex physical location id + + """ + COMPLEXES.update({physical_location_id: request.get_json()}) + + @staticmethod + def reset(): + """Reset Complex resource. + + Clean COMPLEXES dictionary + + """ + global COMPLEXES + COMPLEXES = {} + + +class ComplexList(Resource): + """List of complexes resource.""" + + def get(self) -> Dict[str, List]: + """Get the list of complexes. + + Return data from COMPLEXES dictionary. + + Returns: + Dict[str, List]: Complexes dictionary + + """ + return { + "complex": [complex_data for physical_location_id, complex_data in COMPLEXES.items()] + } diff --git a/mock-aai/resources/customer.py b/mock-aai/resources/customer.py new file mode 100644 index 0000000..d53eb0d --- /dev/null +++ b/mock-aai/resources/customer.py @@ -0,0 +1,325 @@ +"""A&AI Customer mock module.""" +""" + 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 Dict, List +from uuid import uuid4 + +from flask_restful import reqparse, Resource, request + + +CUSTOMERS = {} + + +def random_resource_version() -> str: + """Generate random resource version string value. + + Returns: + str: UUID value + + """ + return str(uuid4()) + + +class Customer(Resource): + """Customer resource.""" + + @staticmethod + def reset(): + """Reset Cloud region resource. + + Clean CUSTOMERS dictionary + + """ + global CUSTOMERS + CUSTOMERS = {} + + def get(self, global_customer_id: str): + """Get customer. + + Get customer from CUSTOMERS dictionary. + + Args: + global_customer_id (str): global customer id key value + + Returns: + Dict[str, str]: Customer dictionary + + """ + return CUSTOMERS[global_customer_id] + + def put(self, global_customer_id: str): + """Resource put method. + + Add customer data into CUSTOMERS dictionary. + + Args: + global_customer_id (str): global customer id key value + + Returns: + Tuple[str, int]: Response tuple. First element is a response body, + the second one is HTTP response code. + + """ + CUSTOMERS[global_customer_id] = request.get_json() + CUSTOMERS[global_customer_id]["resource-version"] = random_resource_version() + return "", 201 + + +class CustomerList(Resource): + """List of customers resource.""" + + def get(self) -> Dict[str, List]: + """Get the list of customers. + + Return data from CUSTOMERS dictionary. + + Returns: + Dict[str, List]: Customer dictionary + + """ + return {"customer": [data for global_customer_id, data in CUSTOMERS.items()]} + + +class ServiceSubscription(Resource): + """Service subscription resource.""" + + def put(self, global_customer_id: str, service_type: str) -> None: + """Service subscription resource put method. + + Add service subscription data into CUSTOMERS dictionary. + + Args: + global_customer_id (str): Global customer id key value + service_type (str): Service type key value + + """ + try: + CUSTOMERS[global_customer_id]["service_subscriptions"][ + service_type + ] = {"service-type": service_type} + except KeyError: + CUSTOMERS[global_customer_id]["service_subscriptions"] = { + service_type: {"service-type": service_type} + } + CUSTOMERS[global_customer_id]["service_subscriptions"][service_type].update( + {"service-type": service_type} + ) + + +class ServiceSubscriptionList(Resource): + """List of service subscriptions resource.""" + + def get(self, global_customer_id: str) -> Dict[str, List]: + """Get the list of service subscriptions. + + Return data from CUSTOMERS dictionary. + + Args: + global_customer_id (str): Global customer id key value + + Returns: + Dict[str, List]: Service subscriptions dictionary + + """ + service_type: str = request.args.get("service-type") + if not service_type: + return { + "service-subscription": [ + data + for service_type, data in CUSTOMERS[global_customer_id][ + "service_subscriptions" + ].items() + ] + } + try: + return { + "service-subscription": [ + CUSTOMERS[global_customer_id]["service_subscriptions"][service_type] + ] + } + except KeyError: + return "", 404 + + +class ServiceSubscriptionRelationship(Resource): + """Service subscription relationship resource.""" + + def put(self, global_customer_id: str, service_type: str) -> None: + """Service subscription relationship resource put method. + + Add service subscription relationship data into CUSTOMERS dictionary. + + Args: + global_customer_id (str): Global customer id key value + service_type (str): Service type key value + + """ + try: + CUSTOMERS[global_customer_id]["service_subscriptions"][service_type][ + "relationships" + ].append(request.get_json()) + except KeyError: + CUSTOMERS[global_customer_id]["service_subscriptions"][service_type][ + "relationships" + ] = [request.get_json()] + + +class ServiceSubscriptionRelationshipList(Resource): + """Service subscription relationships list resource.""" + + def get(self, global_customer_id: str, service_type: str) -> Dict[str, List]: + """Get the list of service subscription relationships. + + Return data from CUSTOMERS dictionary. + + Args: + global_customer_id (str): Global customer id key value + service_type (str): Service type key value + + Returns: + Dict[str, List]: Service subscription relationships dictionary + + """ + return { + "relationship": CUSTOMERS[global_customer_id]["service_subscriptions"][ + service_type + ].get("relationships", []) + } + + +class ServiceSubscriptionInstance(Resource): + """Service subscription instance resource.""" + + def delete(self, global_customer_id: str, service_type: str, service_instance_id: str) -> None: + """Delete service subscription instance. + + Removes data from CUSTOMERS dictionary. + + Args: + global_customer_id (str): Global customer id key value + service_type (str): Service type key value + service_instance_id (str): Service instance id key value + + """ + del CUSTOMERS[global_customer_id]["service_subscriptions"][service_type][ + "service_instances" + ][service_instance_id] + + +class ServiceSubscriptionInstanceList(Resource): + """Service subscription instances list resource.""" + + def get(self, global_customer_id: str, service_type: str) -> Dict[str, List]: + """Get service subscription's service instances. + + Returns data from CUSTOMERS dictionary + + Args: + global_customer_id (str): Global customer id key value + service_type (str): Service type key value + + Returns: + Dict[str, List]: Service instances dictionary + """ + return { + "service-instance": [ + data + for instance_id, data in CUSTOMERS[global_customer_id]["service_subscriptions"][ + service_type + ] + .get("service_instances", dict()) + .items() + ] + } + + def post(self, global_customer_id: str, service_type: str) -> None: + """Add service instance to service subscription. + + Add service instance data dictionary to service subscription's + service instances dictionary. + + Args: + global_customer_id (str): Global customer id key value + service_type (str): Service type key value + + """ + request_data = request.get_json() + instance_id = request_data["service-instance-id"] + try: + CUSTOMERS[global_customer_id]["service_subscriptions"][service_type][ + "service_instances" + ][instance_id] = request_data + except KeyError: + CUSTOMERS[global_customer_id]["service_subscriptions"][service_type][ + "service_instances" + ] = {instance_id: request_data} + + +class ServiceSubscriptionInstanceRelationshipList(Resource): + """Service subscription instance relationships list resource.""" + + def post(self, global_customer_id: str, service_type: str, service_instance_id) -> None: + """Add relationship into service instance relationships list. + + Args: + global_customer_id (str): Global customer id key value + service_type (str): Service type key value + service_instance_id (str): Service instance id key value + + """ + try: + CUSTOMERS[global_customer_id]["service_subscriptions"][service_type][ + "service_instances" + ][service_instance_id]["relationships"].append(request.get_json()) + except KeyError: + CUSTOMERS[global_customer_id]["service_subscriptions"][service_type][ + "service_instances" + ][service_instance_id]["relationships"] = [request.get_json()] + + def get( + self, global_customer_id: str, service_type: str, service_instance_id: str + ) -> Dict[str, List]: + """Get the service instance relationships list. + + Args: + global_customer_id (str): Global customer id key value + service_type (str): Service type key value + service_instance_id (str): Service instance id key value + + Returns: + Dict[str, List]: Service instance relationships dictionary + + """ + return { + "relationship": CUSTOMERS[global_customer_id]["service_subscriptions"][service_type][ + "service_instances" + ][service_instance_id].get("relationships", []) + } + + def delete(self, global_customer_id: str, service_type: str, service_instance_id: str) -> None: + """Delete service subscription instance relationships. + + Make relationships list clea. + + Args: + global_customer_id (str): Global customer id key value + service_type (str): Service type key value + service_instance_id (str): Service instance id key value + + """ + CUSTOMERS[global_customer_id]["service_subscriptions"][service_type]["service_instances"][ + service_instance_id + ]["relationships"] = [] diff --git a/mock-aai/resources/network.py b/mock-aai/resources/network.py new file mode 100644 index 0000000..3fddd41 --- /dev/null +++ b/mock-aai/resources/network.py @@ -0,0 +1,61 @@ +"""Vnf resources module.""" +""" + 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 Dict, List +from uuid import uuid4 + +from flask_restful import Resource, request + + +NETWORKS = {} + + +class Network(Resource): + """Network resource.""" + + @staticmethod + def reset(): + """Reset resource for tests. + + Create new, empty NETWORKS dictionary + + """ + global NETWORKS + NETWORKS = {} + + def get(self, network_instance_id: str) -> Dict[str, List]: + """Get network instance data. + + Get data from NETWORKS dictionary + + Args: + network_instance_id (str): Network instance id key value + + Returns: + Dict[str, List]: Network instance data dictionary + + """ + try: + return NETWORKS[network_instance_id] + except KeyError: + NETWORKS[network_instance_id] = { + "network-id": network_instance_id, + "is-bound-to-vpn": False, + "is-provider-network": False, + "is-shared-network": False, + "is-external-network": False, + } + return NETWORKS[network_instance_id] diff --git a/mock-aai/resources/vnf.py b/mock-aai/resources/vnf.py new file mode 100644 index 0000000..3a49f01 --- /dev/null +++ b/mock-aai/resources/vnf.py @@ -0,0 +1,108 @@ +"""Vnf resources module.""" +""" + 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 Dict, List +from uuid import uuid4 + +from flask_restful import Resource, request + + +VNFS = {} + + +class Vnf(Resource): + """Vnf resource.""" + + @staticmethod + def reset(): + """Reset resource for tests. + + Create new, empty VNFS dictionary + + """ + global VNFS + VNFS = {} + + def get(self, vnf_instance_id: str) -> Dict[str, List]: + """Get vnf instance data. + + Get data from VNFS dictionary + + Args: + vnf_instance_id (str): Vnf instance id key value + + Returns: + Dict[str, List]: Vnf instance data dictionary + + """ + try: + return VNFS[vnf_instance_id] + except KeyError: + VNFS[vnf_instance_id] = {"vnf-id": vnf_instance_id, "model-version-id": str(uuid4())} + return VNFS[vnf_instance_id] + + +class VfModule(Resource): + """Vf module resource.""" + + def delete(self, vnf_instance_id: str, vf_module_instance_id: str) -> None: + """Delete vf module. + + Removes vf module data from VNFS dictionary. + + Args: + vnf_instance_id (str): Vnf instance id key value + vf_module_instance_id (str): Vf module instance id key value + + """ + del VNFS[vnf_instance_id]["vf_modules"][vf_module_instance_id] + + +class VfModuleList(Resource): + """Vf module list resource.""" + + def post(self, vnf_instance_id: str) -> None: + """Create vf module. + + Add vf module data into VNFS dictionary. + + Args: + vnf_instance_id (str): Vnf instance id key value + + """ + vf_module_data = request.get_json() + vf_module_dict = {vf_module_data["vf-module-id"]: vf_module_data} + try: + VNFS[vnf_instance_id]["vf_modules"].update(vf_module_dict) + except KeyError: + VNFS[vnf_instance_id]["vf_modules"] = vf_module_dict + + def get(self, vnf_instance_id: str) -> Dict[str, List]: + """Get Vnf instance Vf modules list. + + Get data from VNFS dictionary + + Args: + vnf_instance_id (str): Vnf instance id key value + + Returns: + Dict[str, List]: Vnf instance vf modules dictionary + """ + return { + "vf-module": [ + data for vf_module_id, data in VNFS[vnf_instance_id].get("vf_modules", {}).items() + ] + } |