diff options
Diffstat (limited to 'components/pm-subscription-handler/tests')
5 files changed, 91 insertions, 47 deletions
diff --git a/components/pm-subscription-handler/tests/test_controller.py b/components/pm-subscription-handler/tests/test_controller.py new file mode 100755 index 00000000..8ef39460 --- /dev/null +++ b/components/pm-subscription-handler/tests/test_controller.py @@ -0,0 +1,69 @@ +# ============LICENSE_START=================================================== +# Copyright (C) 2019-2020 Nordix Foundation. +# ============================================================================ +# 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. +# +# SPDX-License-Identifier: Apache-2.0 +# ============LICENSE_END===================================================== +import json +import os +import unittest +from test.support import EnvironmentVarGuard +from unittest.mock import patch + +from requests import Session + +from mod import aai_client, create_app, db +from mod.api.controller import status, get_all_sub_to_nf_relations +from mod.network_function import NetworkFunction + + +class ControllerTestCase(unittest.TestCase): + @patch('mod.get_db_connection_url') + @patch.object(Session, 'put') + def setUp(self, mock_session, mock_get_db_url): + mock_get_db_url.return_value = 'sqlite://' + with open(os.path.join(os.path.dirname(__file__), 'data/aai_xnfs.json'), 'r') as data: + self.aai_response_data = data.read() + mock_session.return_value.status_code = 200 + mock_session.return_value.text = self.aai_response_data + self.env = EnvironmentVarGuard() + self.env.set('AAI_SERVICE_HOST', '1.2.3.4') + self.env.set('AAI_SERVICE_PORT', '8443') + self.env.set('TESTING', 'True') + self.env.set('LOGS_PATH', './unit_test_logs') + with open(os.path.join(os.path.dirname(__file__), 'data/cbs_data_1.json'), 'r') as data: + self.cbs_data_1 = json.load(data) + self.sub_1, self.xnfs = aai_client.get_pmsh_subscription_data(self.cbs_data_1) + self.nf_1 = NetworkFunction(nf_name='pnf_1', orchestration_status='Inventoried') + self.nf_2 = NetworkFunction(nf_name='pnf_2', orchestration_status='Active') + self.app = create_app() + self.app_context = self.app.app_context() + self.app_context.push() + db.create_all() + + def tearDown(self): + db.session.remove() + db.drop_all() + self.app_context.pop() + + def test_status_response_healthy(self): + self.assertEqual(status()['status'], 'healthy') + + def test_get_all_sub_to_nf_relations(self): + self.sub_1.create() + for nf in [self.nf_1, self.nf_2]: + self.sub_1.add_network_function_to_subscription(nf) + all_subs = get_all_sub_to_nf_relations() + self.assertEqual(len(all_subs[0]['network_functions']), 2) + self.assertEqual(all_subs[0]['subscription_name'], 'ExtraPM-All-gNB-R2B') diff --git a/components/pm-subscription-handler/tests/test_healthcheck.py b/components/pm-subscription-handler/tests/test_healthcheck.py deleted file mode 100755 index 1c40c3ff..00000000 --- a/components/pm-subscription-handler/tests/test_healthcheck.py +++ /dev/null @@ -1,27 +0,0 @@ -# ============LICENSE_START=================================================== -# Copyright (C) 2019-2020 Nordix Foundation. -# ============================================================================ -# 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. -# -# SPDX-License-Identifier: Apache-2.0 -# ============LICENSE_END===================================================== - -import unittest - -from mod.healthcheck import status - - -class HealthcheckTestCase(unittest.TestCase): - - def test_status_response_healthy(self): - self.assertEqual(status()['status'], 'healthy') diff --git a/components/pm-subscription-handler/tests/test_network_function.py b/components/pm-subscription-handler/tests/test_network_function.py index 4fca0778..138d99ad 100755 --- a/components/pm-subscription-handler/tests/test_network_function.py +++ b/components/pm-subscription-handler/tests/test_network_function.py @@ -71,7 +71,8 @@ class NetworkFunctionTests(TestCase): self.nf_1.create() self.nf_2.create() sub = Subscription(**{"subscriptionName": "sub"}) - sub.add_network_functions_to_subscription([self.nf_1, self.nf_2]) + for nf in [self.nf_1, self.nf_2]: + sub.add_network_function_to_subscription(nf) NetworkFunction.delete(nf_name=self.nf_1.nf_name) diff --git a/components/pm-subscription-handler/tests/test_policy_response_handler.py b/components/pm-subscription-handler/tests/test_policy_response_handler.py index 1cf947fb..6de33632 100644 --- a/components/pm-subscription-handler/tests/test_policy_response_handler.py +++ b/components/pm-subscription-handler/tests/test_policy_response_handler.py @@ -22,7 +22,7 @@ from unittest.mock import patch from tenacity import stop_after_attempt -from mod.db_models import SubscriptionModel +from mod.api.db_models import SubscriptionModel from mod.network_function import NetworkFunction from mod.subscription import AdministrativeState, SubNfState from mod.policy_response_handler import PolicyResponseHandler, policy_response_handle_functions diff --git a/components/pm-subscription-handler/tests/test_subscription.py b/components/pm-subscription-handler/tests/test_subscription.py index e6ee2b57..dc549c94 100755 --- a/components/pm-subscription-handler/tests/test_subscription.py +++ b/components/pm-subscription-handler/tests/test_subscription.py @@ -26,7 +26,7 @@ from tenacity import stop_after_attempt import mod.aai_client as aai_client from mod import db, create_app -from mod.db_models import NetworkFunctionModel +from mod.api.db_models import NetworkFunctionModel from mod.network_function import NetworkFunction, NetworkFunctionFilter, OrchestrationStatus from mod.pmsh_utils import AppConfig from mod.subscription import Subscription @@ -46,7 +46,7 @@ class SubscriptionTest(TestCase): mock_session.return_value.text = self.aai_response_data self.env = EnvironmentVarGuard() self.env.set('AAI_SERVICE_HOST', '1.2.3.4') - self.env.set('AAI_SERVICE_PORT_AAI_SSL', '8443') + self.env.set('AAI_SERVICE_PORT', '8443') self.env.set('TESTING', 'True') self.env.set('LOGS_PATH', './unit_test_logs') with open(os.path.join(os.path.dirname(__file__), 'data/cbs_data_1.json'), 'r') as data: @@ -110,22 +110,21 @@ class SubscriptionTest(TestCase): self.assertEqual(1, len(self.sub_1.get_all())) def test_add_network_functions_per_subscription(self): - nf_array = [self.nf_1, self.nf_2] - self.sub_1.add_network_functions_to_subscription(nf_array) + for nf in [self.nf_1, self.nf_2]: + self.sub_1.add_network_function_to_subscription(nf) nfs_for_sub_1 = Subscription.get_all_nfs_subscription_relations() self.assertEqual(2, len(nfs_for_sub_1)) - new_nf_array = [NetworkFunction(nf_name='vnf_3', orchestration_status='Inventoried')] - self.sub_1.add_network_functions_to_subscription(new_nf_array) + new_nf = NetworkFunction(nf_name='vnf_3', orchestration_status='Inventoried') + self.sub_1.add_network_function_to_subscription(new_nf) nf_subs = Subscription.get_all_nfs_subscription_relations() print(nf_subs) self.assertEqual(3, len(nf_subs)) def test_add_duplicate_network_functions_per_subscription(self): - nf_array = [self.nf_1] - self.sub_1.add_network_functions_to_subscription(nf_array) + self.sub_1.add_network_function_to_subscription(self.nf_1) nf_subs = Subscription.get_all_nfs_subscription_relations() self.assertEqual(1, len(nf_subs)) - self.sub_1.add_network_functions_to_subscription(nf_array) + self.sub_1.add_network_function_to_subscription(self.nf_1) nf_subs = Subscription.get_all_nfs_subscription_relations() self.assertEqual(1, len(nf_subs)) @@ -139,8 +138,10 @@ class SubscriptionTest(TestCase): self.assertEqual('new_status', sub.status) def test_delete_subscription(self): - self.sub_1.add_network_functions_to_subscription([self.nf_1, self.nf_2]) - self.sub_2.add_network_functions_to_subscription([self.nf_2]) + for nf in [self.nf_1, self.nf_2]: + self.sub_1.add_network_function_to_subscription(nf) + for nf in [self.nf_2]: + self.sub_2.add_network_function_to_subscription(nf) self.sub_1.delete_subscription() @@ -152,8 +153,8 @@ class SubscriptionTest(TestCase): def test_update_sub_nf_status(self): sub_name = 'ExtraPM-All-gNB-R2B' - nf_array = [self.nf_1, self.nf_2] - self.sub_1.add_network_functions_to_subscription(nf_array) + for nf in [self.nf_1, self.nf_2]: + self.sub_1.add_network_function_to_subscription(nf) sub_nfs = Subscription.get_all_nfs_subscription_relations() self.assertEqual('PENDING_CREATE', sub_nfs[0].nf_sub_status) @@ -162,7 +163,7 @@ class SubscriptionTest(TestCase): self.assertEqual('Active', sub_nfs[0].nf_sub_status) self.assertEqual('PENDING_CREATE', sub_nfs[1].nf_sub_status) - @patch('mod.subscription.Subscription.add_network_functions_to_subscription') + @patch('mod.subscription.Subscription.add_network_function_to_subscription') @patch('mod.subscription.Subscription.update_sub_nf_status') @patch('mod.subscription.Subscription.update_subscription_status') def test_process_activate_subscription(self, mock_update_sub_status, @@ -204,16 +205,16 @@ class SubscriptionTest(TestCase): self.assertEqual(expected_sub_event, actual_sub_event) def test_get_nf_models(self): - nf_array = [self.nf_1, self.nf_2] - self.sub_1.add_network_functions_to_subscription(nf_array) + for nf in [self.nf_1, self.nf_2]: + self.sub_1.add_network_function_to_subscription(nf) nf_models = self.sub_1._get_nf_models() self.assertEqual(2, len(nf_models)) self.assertIsInstance(nf_models[0], NetworkFunctionModel) def test_get_network_functions(self): - nf_array = [self.nf_1, self.nf_2] - self.sub_1.add_network_functions_to_subscription(nf_array) + for nf in [self.nf_1, self.nf_2]: + self.sub_1.add_network_function_to_subscription(nf) nfs = self.sub_1.get_network_functions() self.assertEqual(2, len(nfs)) |