From 6bc82510fcc73a6dc2f817caa2ce74b52daa345c Mon Sep 17 00:00:00 2001 From: "Zhang Rong(Jon)" Date: Wed, 24 Nov 2021 23:57:29 +0800 Subject: [PATCH] Rename some variables; Add test case for flask 1. Rename api in the ims to specify version 2. Add unit test and integration test case for flask Signed-off-by: Zhang Rong(Jon) Change-Id: I8b54927eb3064e03ad4902402b2fdebbb808e4e6 --- o2ims/views/__init__.py | 5 +- o2ims/views/ocloud_dto.py | 20 +-- o2ims/views/ocloud_route.py | 117 ++++++------ tests/conftest.py | 12 +- tests/integration/test_ocloud_route.py | 182 +++++++++++++++++++ tests/unit/test_ocloud.py | 319 ++++++++++++++++++--------------- 6 files changed, 439 insertions(+), 216 deletions(-) create mode 100644 tests/integration/test_ocloud_route.py diff --git a/o2ims/views/__init__.py b/o2ims/views/__init__.py index 5d7745f..0647235 100644 --- a/o2ims/views/__init__.py +++ b/o2ims/views/__init__.py @@ -14,5 +14,6 @@ from flask_restx import Namespace -api = Namespace("O2IMS_Inventory", - description='IMS Inventory related operations.') +api_ims_inventory_v1 = Namespace( + "O2IMS_Inventory", + description='IMS Inventory related operations.') diff --git a/o2ims/views/ocloud_dto.py b/o2ims/views/ocloud_dto.py index e480e09..1a489e1 100644 --- a/o2ims/views/ocloud_dto.py +++ b/o2ims/views/ocloud_dto.py @@ -14,12 +14,12 @@ from flask_restx import fields -from o2ims.views import api +from o2ims.views import api_ims_inventory_v1 class OcloudDTO: - ocloud = api.model( + ocloud = api_ims_inventory_v1.model( "OcloudDto", { 'oCloudId': fields.String(required=True), @@ -33,7 +33,7 @@ class OcloudDTO: class ResourceTypeDTO: - resource_type_get = api.model( + resource_type_get = api_ims_inventory_v1.model( "ResourceTypeGetDto", { 'resourceTypeId': fields.String(required=True, @@ -48,7 +48,7 @@ class ResourceTypeDTO: class ResourcePoolDTO: - resource_pool_get = api.model( + resource_pool_get = api_ims_inventory_v1.model( "ResourcePoolGetDto", { 'resourcePoolId': fields.String(required=True, @@ -63,7 +63,7 @@ class ResourcePoolDTO: class ResourceDTO: - resource_list = api.model( + resource_list = api_ims_inventory_v1.model( "ResourceListDto", { 'resourceId': fields.String(required=True, @@ -75,7 +75,7 @@ class ResourceDTO: } ) - resource_get = api.model( + resource_get = api_ims_inventory_v1.model( "ResourceGetDto", { 'resourceId': fields.String(required=True, @@ -90,7 +90,7 @@ class ResourceDTO: class DeploymentManagerDTO: - deployment_manager_get = api.model( + deployment_manager_get = api_ims_inventory_v1.model( "DeploymentManagerGetDto", { 'deploymentManagerId': fields.String( @@ -108,7 +108,7 @@ class DeploymentManagerDTO: class SubscriptionDTO: - subscription_get = api.model( + subscription_get = api_ims_inventory_v1.model( "SubscriptionGetDto", { 'subscriptionId': fields.String(required=True, @@ -119,7 +119,7 @@ class SubscriptionDTO: } ) - subscription = api.model( + subscription = api_ims_inventory_v1.model( "SubscriptionCreateDto", { 'callback': fields.String( @@ -129,7 +129,7 @@ class SubscriptionDTO: } ) - subscription_post_resp = api.model( + subscription_post_resp = api_ims_inventory_v1.model( "SubscriptionCreatedRespDto", { 'subscriptionId': fields.String(required=True, diff --git a/o2ims/views/ocloud_route.py b/o2ims/views/ocloud_route.py index 4519b6d..bb4e63f 100644 --- a/o2ims/views/ocloud_route.py +++ b/o2ims/views/ocloud_route.py @@ -15,7 +15,7 @@ import uuid from flask_restx import Resource -from o2ims.views import ocloud_view, api +from o2ims.views import ocloud_view, api_ims_inventory_v1 from o2common.config import config from o2ims.domain.ocloud import Subscription from o2ims.views.ocloud_dto import OcloudDTO, ResourceTypeDTO,\ @@ -26,8 +26,8 @@ apibase = config.get_o2ims_api_base() # ---------- OClouds ---------- # -@api.route("/") -@api.response(404, 'oCloud not found') +@api_ims_inventory_v1.route("/") +@api_ims_inventory_v1.response(404, 'oCloud not found') class OcloudsListRouter(Resource): """Ocloud get endpoint O2 interface ocloud endpoint @@ -35,147 +35,151 @@ class OcloudsListRouter(Resource): ocloud_get = OcloudDTO.ocloud - @api.marshal_with(ocloud_get) + @api_ims_inventory_v1.marshal_with(ocloud_get) def get(self): res = ocloud_view.oclouds(bus.uow) if len(res) > 0: return res[0] - api.abort( + api_ims_inventory_v1.abort( 404, "oCloud doesn't exist") # ---------- ResourceTypes ---------- # -@api.route("/resourceTypes") +@api_ims_inventory_v1.route("/resourceTypes") class ResourceTypesListRouter(Resource): model = ResourceTypeDTO.resource_type_get - @api.marshal_list_with(model) + @api_ims_inventory_v1.marshal_list_with(model) def get(self): return ocloud_view.resource_types(bus.uow) -@api.route("/resourceTypes/") -@api.param('resourceTypeID', 'ID of the resource type') -@api.response(404, 'Resource type not found') +@api_ims_inventory_v1.route("/resourceTypes/") +@api_ims_inventory_v1.param('resourceTypeID', 'ID of the resource type') +@api_ims_inventory_v1.response(404, 'Resource type not found') class ResourceTypeGetRouter(Resource): model = ResourceTypeDTO.resource_type_get - @api.doc('Get resource type') - @api.marshal_with(model) + @api_ims_inventory_v1.doc('Get resource type') + @api_ims_inventory_v1.marshal_with(model) def get(self, resourceTypeID): result = ocloud_view.resource_type_one(resourceTypeID, bus.uow) if result is not None: return result - api.abort( + api_ims_inventory_v1.abort( 404, "Resource type {} doesn't exist".format(resourceTypeID)) # ---------- ResourcePools ---------- # -@api.route("/resourcePools") +@api_ims_inventory_v1.route("/resourcePools") class ResourcePoolsListRouter(Resource): model = ResourcePoolDTO.resource_pool_get - @api.marshal_list_with(model) + @api_ims_inventory_v1.marshal_list_with(model) def get(self): return ocloud_view.resource_pools(bus.uow) -@api.route("/resourcePools/") -@api.param('resourcePoolID', 'ID of the resource pool') -@api.response(404, 'Resource pool not found') +@api_ims_inventory_v1.route("/resourcePools/") +@api_ims_inventory_v1.param('resourcePoolID', 'ID of the resource pool') +@api_ims_inventory_v1.response(404, 'Resource pool not found') class ResourcePoolGetRouter(Resource): model = ResourcePoolDTO.resource_pool_get - @api.doc('Get resource pool') - @api.marshal_with(model) + @api_ims_inventory_v1.doc('Get resource pool') + @api_ims_inventory_v1.marshal_with(model) def get(self, resourcePoolID): result = ocloud_view.resource_pool_one(resourcePoolID, bus.uow) if result is not None: return result - api.abort( + api_ims_inventory_v1.abort( 404, "Resource pool {} doesn't exist".format(resourcePoolID)) # ---------- Resources ---------- # -@api.route("/resourcePools//resources") -@api.param('resourcePoolID', 'ID of the resource pool') +@api_ims_inventory_v1.route("/resourcePools//resources") +@api_ims_inventory_v1.param('resourcePoolID', 'ID of the resource pool') class ResourcesListRouter(Resource): model = ResourceDTO.resource_list - @api.marshal_list_with(model) + @api_ims_inventory_v1.marshal_list_with(model) def get(self, resourcePoolID): return ocloud_view.resources(resourcePoolID, bus.uow) -@api.route("/resourcePools//resources/") -@api.param('resourcePoolID', 'ID of the resource pool') -@api.param('resourceID', 'ID of the resource') -@api.response(404, 'Resource not found') +@api_ims_inventory_v1.route( + "/resourcePools//resources/") +@api_ims_inventory_v1.param('resourcePoolID', 'ID of the resource pool') +@api_ims_inventory_v1.param('resourceID', 'ID of the resource') +@api_ims_inventory_v1.response(404, 'Resource not found') class ResourceGetRouter(Resource): model = ResourceDTO.resource_get - @api.doc('Get resource') - @api.marshal_with(model) + @api_ims_inventory_v1.doc('Get resource') + @api_ims_inventory_v1.marshal_with(model) def get(self, resourcePoolID, resourceID): result = ocloud_view.resource_one(resourceID, bus.uow) if result is not None: return result - api.abort(404, "Resource {} doesn't exist".format(resourceID)) + api_ims_inventory_v1.abort( + 404, "Resource {} doesn't exist".format(resourceID)) # ---------- DeploymentManagers ---------- # -@api.route("/deploymentManagers") +@api_ims_inventory_v1.route("/deploymentManagers") class DeploymentManagersListRouter(Resource): model = DeploymentManagerDTO.deployment_manager_get - @api.marshal_list_with(model) + @api_ims_inventory_v1.marshal_list_with(model) def get(self): return ocloud_view.deployment_managers(bus.uow) -@api.route("/deploymentManagers/") -@api.param('deploymentManagerID', 'ID of the deployment manager') -@api.response(404, 'Deployment manager not found') +@api_ims_inventory_v1.route("/deploymentManagers/") +@api_ims_inventory_v1.param('deploymentManagerID', + 'ID of the deployment manager') +@api_ims_inventory_v1.response(404, 'Deployment manager not found') class DeploymentManagerGetRouter(Resource): model = DeploymentManagerDTO.deployment_manager_get - @api.doc('Get deployment manager') - @api.marshal_with(model) + @api_ims_inventory_v1.doc('Get deployment manager') + @api_ims_inventory_v1.marshal_with(model) def get(self, deploymentManagerID): result = ocloud_view.deployment_manager_one( deploymentManagerID, bus.uow) if result is not None: return result - api.abort(404, "Deployment manager {} doesn't exist".format( - deploymentManagerID)) + api_ims_inventory_v1.abort( + 404, + "Deployment manager {} doesn't exist".format(deploymentManagerID)) # ---------- Subscriptions ---------- # -@api.route("/subscriptions") +@api_ims_inventory_v1.route("/subscriptions") class SubscriptionsListRouter(Resource): model = SubscriptionDTO.subscription_get expect = SubscriptionDTO.subscription post_resp = SubscriptionDTO.subscription_post_resp - @api.doc('List subscriptions') - @api.marshal_list_with(model) + @api_ims_inventory_v1.doc('List subscriptions') + @api_ims_inventory_v1.marshal_list_with(model) def get(self): return ocloud_view.subscriptions(bus.uow) - @api.doc('Create a subscription') - @api.expect(expect) - @api.marshal_with(post_resp, code=201) + @api_ims_inventory_v1.doc('Create a subscription') + @api_ims_inventory_v1.expect(expect) + @api_ims_inventory_v1.marshal_with(post_resp, code=201) def post(self): - data = api.payload + data = api_ims_inventory_v1.payload sub_uuid = str(uuid.uuid4()) subscription = Subscription( sub_uuid, data['callback'], data['consumerSubscriptionId'], @@ -184,25 +188,25 @@ class SubscriptionsListRouter(Resource): return {"subscriptionId": sub_uuid}, 201 -@api.route("/subscriptions/") -@api.param('subscriptionID', 'ID of the subscription') -@api.response(404, 'Subscription not found') +@api_ims_inventory_v1.route("/subscriptions/") +@api_ims_inventory_v1.param('subscriptionID', 'ID of the subscription') +@api_ims_inventory_v1.response(404, 'Subscription not found') class SubscriptionGetDelRouter(Resource): model = SubscriptionDTO.subscription_get - @api.doc('Get subscription by ID') - @api.marshal_with(model) + @api_ims_inventory_v1.doc('Get subscription by ID') + @api_ims_inventory_v1.marshal_with(model) def get(self, subscriptionID): result = ocloud_view.subscription_one( subscriptionID, bus.uow) if result is not None: return result - api.abort(404, "Subscription {} doesn't exist".format( + api_ims_inventory_v1.abort(404, "Subscription {} doesn't exist".format( subscriptionID)) - @api.doc('Delete subscription by ID') - @api.response(204, 'Subscription deleted') + @api_ims_inventory_v1.doc('Delete subscription by ID') + @api_ims_inventory_v1.response(204, 'Subscription deleted') def delete(self, subscriptionID): with bus.uow: bus.uow.subscriptions.delete(subscriptionID) @@ -216,5 +220,4 @@ def configure_namespace(app, bus_new): global bus bus = bus_new - api_v1 = api - app.add_namespace(api_v1, path=apibase) + app.add_namespace(api_ims_inventory_v1, path=apibase) diff --git a/tests/conftest.py b/tests/conftest.py index 60f3d2e..9d7f136 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -36,12 +36,11 @@ def mock_uow(): def mock_flask_uow(mock_uow): session, uow = mock_uow app = Flask(__name__) - # app.config["TESTING"] = True + app.config["TESTING"] = True api = Api(app) bus = bootstrap(False, uow) configure_namespace(api, bus) - client = app.test_client() - return session, client + return session, app @pytest.fixture @@ -74,11 +73,11 @@ def sqlite_uow(sqlite_session_factory): @pytest.fixture def sqlite_flask_uow(sqlite_uow): app = Flask(__name__) - # app.config["TESTING"] = True + app.config["TESTING"] = True api = Api(app) bus = bootstrap(False, sqlite_uow) configure_namespace(api, bus) - yield app.test_client() + yield sqlite_uow, app @pytest.fixture @@ -134,10 +133,11 @@ def postgres_uow(postgres_session_factory): @pytest.fixture def postgres_flask_uow(postgres_uow): app = Flask(__name__) + app.config["TESTING"] = True api = Api(app) bus = bootstrap(False, postgres_uow) configure_namespace(api, bus) - yield app.test_client() + yield postgres_uow, app @pytest.fixture diff --git a/tests/integration/test_ocloud_route.py b/tests/integration/test_ocloud_route.py new file mode 100644 index 0000000..441e56d --- /dev/null +++ b/tests/integration/test_ocloud_route.py @@ -0,0 +1,182 @@ +# Copyright (C) 2021 Wind River Systems, Inc. +# +# 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. + +import uuid +import pytest + +from o2common.config import config +from o2ims.domain import ocloud +from o2ims.domain import resource_type as rt + + +pytestmark = pytest.mark.usefixtures("mappers") + + +def setup_ocloud(): + ocloudid1 = str(uuid.uuid4()) + ocloud1 = ocloud.Ocloud( + ocloudid1, "ocloud1", config.get_api_url(), + "ocloud 1 for integration test", 1) + return ocloud1 + + +def test_route_olcouds(sqlite_flask_uow): + uow, app = sqlite_flask_uow + + with uow: + ocloud1 = setup_ocloud() + ocloud1_UUID = ocloud1.oCloudId + uow.oclouds.add(ocloud1) + uow.commit() + + with app.test_client() as c: + apibase = config.get_o2ims_api_base() + resp = c.get(apibase+"/") + assert resp.status_code == 200 + assert ocloud1_UUID.encode() in resp.data + + +def test_route_resource_types(sqlite_flask_uow): + uow, app = sqlite_flask_uow + + with uow: + ocloud1_id = str(uuid.uuid4()) + resource_type_id1 = str(uuid.uuid4()) + resource_type1 = ocloud.ResourceType( + resource_type_id1, "resourcetype1", rt.ResourceTypeEnum.PSERVER, + ocloud1_id) + uow.resource_types.add(resource_type1) + uow.commit() + + with app.test_client() as c: + apibase = config.get_o2ims_api_base() + resp = c.get(apibase+"/resourceTypes") + assert resp.status_code == 200 + assert resource_type_id1.encode() in resp.data + + resp = c.get(apibase+'/resourceTypes/'+resource_type_id1) + assert resp.status_code == 200 + json_data = resp.get_json() + assert resource_type_id1 in json_data['resourceTypeId'] + + +def test_route_resource_pools(sqlite_flask_uow): + uow, app = sqlite_flask_uow + + with uow: + ocloud1_id = str(uuid.uuid4()) + resource_pool_id1 = str(uuid.uuid4()) + resource_pool1 = ocloud.ResourcePool( + resource_pool_id1, "resourcepool1", config.get_api_url(), + ocloud1_id) + uow.resource_pools.add(resource_pool1) + uow.commit() + + with app.test_client() as c: + apibase = config.get_o2ims_api_base() + resp = c.get(apibase+"/resourcePools") + assert resp.status_code == 200 + assert resource_pool_id1.encode() in resp.data + + resp = c.get(apibase+'/resourcePools/'+resource_pool_id1) + assert resp.status_code == 200 + json_data = resp.get_json() + assert resource_pool_id1 in json_data['resourcePoolId'] + + +def test_route_resources(sqlite_flask_uow): + uow, app = sqlite_flask_uow + + with uow: + resource_id1 = str(uuid.uuid4()) + resource_type_id1 = str(uuid.uuid4()) + resource_pool_id1 = str(uuid.uuid4()) + resource1 = ocloud.Resource( + resource_id1, resource_type_id1, resource_pool_id1) + uow.resources.add(resource1) + uow.commit() + + with app.test_client() as c: + apibase = config.get_o2ims_api_base() + resp = c.get(apibase+"/resourcePools/"+resource_pool_id1+"/resources") + assert resp.status_code == 200 + assert resource_id1.encode() in resp.data + + resp = c.get(apibase+"/resourcePools/"+resource_pool_id1 + + "/resources/" + resource_id1) + assert resp.status_code == 200 + json_data = resp.get_json() + assert resource_pool_id1 in json_data['resourcePoolId'] + assert resource_type_id1 in json_data['resourceTypeId'] + assert resource_id1 in json_data['resourceId'] + + +def test_route_deployment_managers(sqlite_flask_uow): + uow, app = sqlite_flask_uow + + with uow: + ocloud_id1 = str(uuid.uuid4()) + deployment_manager_id1 = str(uuid.uuid4()) + deployment_manager1 = ocloud.DeploymentManager( + deployment_manager_id1, "k8s1", ocloud_id1, + config.get_api_url()+"/k8s1") + uow.deployment_managers.add(deployment_manager1) + uow.commit() + + with app.test_client() as c: + apibase = config.get_o2ims_api_base() + resp = c.get(apibase+"/deploymentManagers") + assert resp.status_code == 200 + assert deployment_manager_id1.encode() in resp.data + + resp = c.get(apibase+'/deploymentManagers/'+deployment_manager_id1) + assert resp.status_code == 200 + json_data = resp.get_json() + assert deployment_manager_id1 in json_data['deploymentManagerId'] + + +def test_route_subscriptions(sqlite_flask_uow): + _, app = sqlite_flask_uow + + with app.test_client() as c: + apibase = config.get_o2ims_api_base() + + sub_callback = 'http://subscription/callback/url' + resp = c.post(apibase+'/subscriptions', json={ + 'callback': sub_callback, + 'consumerSubscriptionId': 'consumerSubId1', + 'filter': 'empty' + }) + assert resp.status_code == 201 + json_data = resp.get_json() + assert 'subscriptionId' in json_data + subscriptionId1 = json_data['subscriptionId'] + + resp = c.get(apibase+'/subscriptions') + assert resp.status_code == 200 + json_data = resp.get_json() + assert 1 == len(json_data) + + resp = c.get(apibase+'/subscriptions/'+subscriptionId1) + assert resp.status_code == 200 + json_data = resp.get_json() + assert sub_callback in json_data['callback'] + + resp = c.delete(apibase+'/subscriptions/'+subscriptionId1) + assert resp.status_code == 204 + + resp = c.get(apibase+'/subscriptions') + assert resp.status_code == 200 + json_data = resp.get_json() + assert 0 == len(json_data) diff --git a/tests/unit/test_ocloud.py b/tests/unit/test_ocloud.py index a2904b7..95bd9cd 100644 --- a/tests/unit/test_ocloud.py +++ b/tests/unit/test_ocloud.py @@ -288,170 +288,207 @@ def test_view_subscription_one(mock_uow): def test_flask_get_list(mock_flask_uow): - session, client = mock_flask_uow + session, app = mock_flask_uow session.return_value.execute.return_value = [] apibase = config.get_o2ims_api_base() - # Get list and return empty list - ########################## - resp = client.get(apibase+"/resourceTypes") - assert resp.get_data() == b'[]\n' + with app.test_client() as client: + # Get list and return empty list + ########################## + resp = client.get(apibase+"/resourceTypes") + assert resp.get_data() == b'[]\n' - resp = client.get(apibase+"/resourcePools") - assert resp.get_data() == b'[]\n' + resp = client.get(apibase+"/resourcePools") + assert resp.get_data() == b'[]\n' - resource_pool_id1 = str(uuid.uuid4()) - resp = client.get(apibase+"/resourcePools/"+resource_pool_id1+"/resources") - assert resp.get_data() == b'[]\n' + resource_pool_id1 = str(uuid.uuid4()) + resp = client.get(apibase+"/resourcePools/" + + resource_pool_id1+"/resources") + assert resp.get_data() == b'[]\n' + + resp = client.get(apibase+"/deploymentManagers") + assert resp.get_data() == b'[]\n' - resp = client.get(apibase+"/deploymentManagers") - assert resp.get_data() == b'[]\n' + resp = client.get(apibase+"/subscriptions") + assert resp.get_data() == b'[]\n' def test_flask_get_one(mock_flask_uow): - session, client = mock_flask_uow + session, app = mock_flask_uow session.return_value.execute.return_value.first.return_value = None apibase = config.get_o2ims_api_base() - # Get one and return 404 - ########################### - resp = client.get(apibase+"/") - assert resp.status_code == 404 + with app.test_client() as client: + # Get one and return 404 + ########################### + resp = client.get(apibase+"/") + assert resp.status_code == 404 - resource_type_id1 = str(uuid.uuid4()) - resp = client.get(apibase+"/resourceTypes/"+resource_type_id1) - assert resp.status_code == 404 + resource_type_id1 = str(uuid.uuid4()) + resp = client.get(apibase+"/resourceTypes/"+resource_type_id1) + assert resp.status_code == 404 - resource_pool_id1 = str(uuid.uuid4()) - resp = client.get(apibase+"/resourcePools/"+resource_pool_id1) - assert resp.status_code == 404 + resource_pool_id1 = str(uuid.uuid4()) + resp = client.get(apibase+"/resourcePools/"+resource_pool_id1) + assert resp.status_code == 404 - resource_id1 = str(uuid.uuid4()) - resp = client.get(apibase+"/resourcePools/" + - resource_pool_id1+"/resources/"+resource_id1) - assert resp.status_code == 404 + resource_id1 = str(uuid.uuid4()) + resp = client.get(apibase+"/resourcePools/" + + resource_pool_id1+"/resources/"+resource_id1) + assert resp.status_code == 404 - deployment_manager_id1 = str(uuid.uuid4()) - resp = client.get(apibase+"/deploymentManagers/"+deployment_manager_id1) - assert resp.status_code == 404 + deployment_manager_id1 = str(uuid.uuid4()) + resp = client.get(apibase+"/deploymentManagers/" + + deployment_manager_id1) + assert resp.status_code == 404 - subscription_id1 = str(uuid.uuid4()) - resp = client.get(apibase+"/subscriptions/"+subscription_id1) - assert resp.status_code == 404 + subscription_id1 = str(uuid.uuid4()) + resp = client.get(apibase+"/subscriptions/"+subscription_id1) + assert resp.status_code == 404 -def test_flask_not_allowed(mock_flask_uow): - _, client = mock_flask_uow +def test_flask_post(mock_flask_uow): + session, app = mock_flask_uow apibase = config.get_o2ims_api_base() - # Testing resource type not support method - ########################## - uri = apibase + "/resourceTypes" - resp = client.post(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - resp = client.put(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - resp = client.patch(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - resp = client.delete(uri) - assert resp.status == '405 METHOD NOT ALLOWED' + with app.test_client() as client: + session.return_value.execute.return_value = [] - resource_type_id1 = str(uuid.uuid4()) - uri = apibase + "/resourceTypes/" + resource_type_id1 - resp = client.post(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - resp = client.put(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - resp = client.patch(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - resp = client.delete(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - - # Testing resource pool not support method - ########################## - uri = apibase + "/resourcePools" - resp = client.post(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - resp = client.put(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - resp = client.patch(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - resp = client.delete(uri) - assert resp.status == '405 METHOD NOT ALLOWED' + sub_callback = 'http://subscription/callback/url' + resp = client.post(apibase+'/subscriptions', json={ + 'callback': sub_callback, + 'consumerSubscriptionId': 'consumerSubId1', + 'filter': 'empty' + }) + assert resp.status_code == 201 + assert 'subscriptionId' in resp.get_json() - resource_pool_id1 = str(uuid.uuid4()) - uri = apibase + "/resourcePools/" + resource_pool_id1 - resp = client.post(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - resp = client.put(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - resp = client.patch(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - resp = client.delete(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - - # Testing resource not support method - ########################## - uri = apibase + "/resourcePools/" + resource_pool_id1 + "/resources" - resp = client.post(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - resp = client.put(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - resp = client.patch(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - resp = client.delete(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - resource_id1 = str(uuid.uuid4()) - uri = apibase + "/resourcePools/" + \ - resource_pool_id1 + "/resources/" + resource_id1 - resp = client.post(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - resp = client.put(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - resp = client.patch(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - resp = client.delete(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - - # Testing deployment managers not support method - ########################## - uri = apibase + "/deploymentManagers" - resp = client.post(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - resp = client.put(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - resp = client.patch(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - resp = client.delete(uri) - assert resp.status == '405 METHOD NOT ALLOWED' +def test_flask_delete(mock_flask_uow): + session, app = mock_flask_uow + apibase = config.get_o2ims_api_base() - deployment_manager_id1 = str(uuid.uuid4()) - uri = apibase + "/deploymentManagers/" + deployment_manager_id1 - resp = client.post(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - resp = client.put(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - resp = client.patch(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - resp = client.delete(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - - # Testing subscriptions not support method - ########################## - uri = apibase + "/subscriptions" - resp = client.put(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - resp = client.patch(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - resp = client.delete(uri) - assert resp.status == '405 METHOD NOT ALLOWED' + with app.test_client() as client: + session.return_value.execute.return_value.first.return_value = {} - subscription_id1 = str(uuid.uuid4()) - uri = apibase + "/subscriptions/" + subscription_id1 - resp = client.post(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - resp = client.put(uri) - assert resp.status == '405 METHOD NOT ALLOWED' - resp = client.patch(uri) - assert resp.status == '405 METHOD NOT ALLOWED' + subscription_id1 = str(uuid.uuid4()) + resp = client.delete(apibase+"/subscriptions/"+subscription_id1) + assert resp.status_code == 204 + + +def test_flask_not_allowed(mock_flask_uow): + _, app = mock_flask_uow + apibase = config.get_o2ims_api_base() + + with app.test_client() as client: + # Testing resource type not support method + ########################## + uri = apibase + "/resourceTypes" + resp = client.post(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + resp = client.put(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + resp = client.patch(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + resp = client.delete(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + + resource_type_id1 = str(uuid.uuid4()) + uri = apibase + "/resourceTypes/" + resource_type_id1 + resp = client.post(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + resp = client.put(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + resp = client.patch(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + resp = client.delete(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + + # Testing resource pool not support method + ########################## + uri = apibase + "/resourcePools" + resp = client.post(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + resp = client.put(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + resp = client.patch(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + resp = client.delete(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + + resource_pool_id1 = str(uuid.uuid4()) + uri = apibase + "/resourcePools/" + resource_pool_id1 + resp = client.post(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + resp = client.put(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + resp = client.patch(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + resp = client.delete(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + + # Testing resource not support method + ########################## + uri = apibase + "/resourcePools/" + resource_pool_id1 + "/resources" + resp = client.post(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + resp = client.put(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + resp = client.patch(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + resp = client.delete(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + + resource_id1 = str(uuid.uuid4()) + uri = apibase + "/resourcePools/" + \ + resource_pool_id1 + "/resources/" + resource_id1 + resp = client.post(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + resp = client.put(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + resp = client.patch(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + resp = client.delete(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + + # Testing deployment managers not support method + ########################## + uri = apibase + "/deploymentManagers" + resp = client.post(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + resp = client.put(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + resp = client.patch(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + resp = client.delete(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + + deployment_manager_id1 = str(uuid.uuid4()) + uri = apibase + "/deploymentManagers/" + deployment_manager_id1 + resp = client.post(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + resp = client.put(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + resp = client.patch(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + resp = client.delete(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + + # Testing subscriptions not support method + ########################## + uri = apibase + "/subscriptions" + resp = client.put(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + resp = client.patch(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + resp = client.delete(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + + subscription_id1 = str(uuid.uuid4()) + uri = apibase + "/subscriptions/" + subscription_id1 + resp = client.post(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + resp = client.put(uri) + assert resp.status == '405 METHOD NOT ALLOWED' + resp = client.patch(uri) + assert resp.status == '405 METHOD NOT ALLOWED' -- 2.16.6