| #!/usr/bin/python3 |
| |
| # Copyright 2022 Canonical Ltd. |
| # |
| # 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 asyncio |
| from unittest import TestCase |
| from unittest.mock import Mock, patch |
| from osm_common import msgbase |
| from osm_common.dbbase import DbException |
| from osm_lcm.paas import PaasLcm |
| |
| |
| class TestPaasLcm(TestCase): |
| @patch("osm_lcm.lcm_utils.Database") |
| @patch("osm_lcm.lcm_utils.Filesystem") |
| def setUp(self, mock_filesystem, mock_database): |
| self.loop = asyncio.get_event_loop() |
| self.msg = Mock(msgbase.MsgBase()) |
| self.lcm_tasks = Mock() |
| self.lcm_tasks.lock_HA.return_value = True |
| self.config = {"database": {"driver": "mongo"}} |
| self.paas_lcm = PaasLcm(self.msg, self.lcm_tasks, self.config, self.loop) |
| self.paas_lcm.db = Mock() |
| self.paas_lcm.fs = Mock() |
| self.paas_lcm.update_db_2 = Mock() |
| self.op_id = "op-id" |
| self.paas_id = "paas-id" |
| self.order_id = "order-id" |
| self.paas_content = {"op_id": self.op_id, "_id": self.paas_id} |
| self.db_paas = { |
| "_id": "_id", |
| "name": "paas-name", |
| "secret": "secret", |
| "schema_version": "1.11", |
| } |
| |
| def check_assert_not_called_when_legacy_nbi(self): |
| self.paas_lcm.db.get_one.assert_not_called() |
| self.paas_lcm.db.encrypt_decrypt_fields.assert_not_called() |
| self.paas_lcm.update_db_2.assert_not_called() |
| self.lcm_tasks.unlock_HA.assert_not_called() |
| self.lcm_tasks.remove.assert_not_called() |
| |
| def check_db_update_when_successful_connectivity(self): |
| self.paas_lcm.update_db_2.assert_called_with( |
| "paas", |
| self.paas_id, |
| { |
| "_admin.operationalState": "ENABLED", |
| "_admin.detailed-status": "Connectivity: ok", |
| }, |
| ) |
| |
| def check_db_update_when_db_exception(self): |
| self.paas_lcm.update_db_2.assert_called_with( |
| "paas", |
| self.paas_id, |
| { |
| "_admin.operationalState": "ERROR", |
| "_admin.detailed-status": "Failed with exception: database exception failed", |
| }, |
| ) |
| |
| def test_paas_lcm_create_legacy_nbi(self): |
| self.lcm_tasks.lock_HA.return_value = False |
| self.loop.run_until_complete( |
| self.paas_lcm.create(self.paas_content, self.order_id) |
| ) |
| self.check_assert_not_called_when_legacy_nbi() |
| |
| def test_paas_lcm_create(self): |
| self.paas_lcm.db.get_one.return_value = self.db_paas |
| self.loop.run_until_complete( |
| self.paas_lcm.create(self.paas_content, self.order_id) |
| ) |
| |
| self.lcm_tasks.lock_HA.assert_called_with("paas", "create", self.op_id) |
| self.paas_lcm.db.encrypt_decrypt_fields.assert_called_with( |
| self.db_paas, "decrypt", ["secret"], schema_version="1.11", salt="_id" |
| ) |
| self.check_db_update_when_successful_connectivity() |
| self.lcm_tasks.unlock_HA.assert_called_with( |
| "paas", |
| "create", |
| self.op_id, |
| operationState="COMPLETED", |
| detailed_status="PaaS validated", |
| ) |
| self.lcm_tasks.remove.assert_called_with("paas", self.paas_id, self.order_id) |
| |
| def test_paas_lcm_create_exception_getting_and_updating_db(self): |
| self.paas_lcm.db.get_one.side_effect = DbException("failed") |
| self.paas_lcm.update_db_2.side_effect = DbException("failed") |
| self.loop.run_until_complete( |
| self.paas_lcm.create(self.paas_content, self.order_id) |
| ) |
| |
| self.lcm_tasks.lock_HA.assert_called_with("paas", "create", self.op_id) |
| self.paas_lcm.db.encrypt_decrypt_fields.assert_not_called() |
| |
| self.check_db_update_when_db_exception() |
| self.lcm_tasks.unlock_HA.assert_not_called() |
| self.lcm_tasks.remove.assert_called_with("paas", self.paas_id, self.order_id) |
| |
| def test_paas_lcm_create_exception_updating_db(self): |
| |
| self.paas_lcm.db.get_one.return_value = self.db_paas |
| self.paas_lcm.update_db_2.side_effect = DbException("failed") |
| self.loop.run_until_complete( |
| self.paas_lcm.create(self.paas_content, self.order_id) |
| ) |
| |
| self.lcm_tasks.lock_HA.assert_called_with("paas", "create", self.op_id) |
| self.paas_lcm.db.encrypt_decrypt_fields.assert_called_with( |
| self.db_paas, "decrypt", ["secret"], schema_version="1.11", salt="_id" |
| ) |
| self.check_db_update_when_successful_connectivity() |
| self.lcm_tasks.unlock_HA.assert_not_called() |
| self.lcm_tasks.remove.assert_called_with("paas", self.paas_id, self.order_id) |
| |
| def test_paas_lcm_create_exception_getting_from_db(self): |
| self.paas_lcm.db.get_one.side_effect = DbException("failed") |
| self.loop.run_until_complete( |
| self.paas_lcm.create(self.paas_content, self.order_id) |
| ) |
| self.lcm_tasks.lock_HA.assert_called_with("paas", "create", self.op_id) |
| self.paas_lcm.db.encrypt_decrypt_fields.assert_not_called() |
| self.check_db_update_when_db_exception() |
| self.lcm_tasks.unlock_HA.assert_called_with( |
| "paas", |
| "create", |
| self.op_id, |
| operationState="FAILED", |
| detailed_status="Failed with exception: database exception failed", |
| ) |
| self.lcm_tasks.remove.assert_called_with("paas", self.paas_id, self.order_id) |
| |
| def test_paas_lcm_edit_legacy_nbi(self): |
| self.lcm_tasks.lock_HA.return_value = False |
| self.loop.run_until_complete( |
| self.paas_lcm.edit(self.paas_content, self.order_id) |
| ) |
| self.check_assert_not_called_when_legacy_nbi() |
| |
| def test_paas_lcm_edit(self): |
| |
| self.paas_lcm.db.get_one.return_value = self.db_paas |
| self.loop.run_until_complete( |
| self.paas_lcm.edit(self.paas_content, self.order_id) |
| ) |
| |
| self.lcm_tasks.lock_HA.assert_called_with("paas", "edit", self.op_id) |
| self.paas_lcm.db.encrypt_decrypt_fields.assert_called_with( |
| self.db_paas, "decrypt", ["secret"], schema_version="1.11", salt="_id" |
| ) |
| self.check_db_update_when_successful_connectivity() |
| self.lcm_tasks.unlock_HA.assert_called_with( |
| "paas", |
| "edit", |
| self.op_id, |
| operationState="COMPLETED", |
| detailed_status="PaaS validated", |
| ) |
| self.lcm_tasks.remove.assert_called_with("paas", self.paas_id, self.order_id) |
| |
| def test_paas_lcm_edit_exception_getting_and_updating_db(self): |
| self.paas_lcm.db.get_one.side_effect = DbException("failed") |
| self.paas_lcm.update_db_2.side_effect = DbException("failed") |
| self.loop.run_until_complete( |
| self.paas_lcm.edit(self.paas_content, self.order_id) |
| ) |
| |
| self.lcm_tasks.lock_HA.assert_called_with("paas", "edit", self.op_id) |
| self.paas_lcm.db.encrypt_decrypt_fields.assert_not_called() |
| |
| self.check_db_update_when_db_exception() |
| self.lcm_tasks.unlock_HA.assert_not_called() |
| self.lcm_tasks.remove.assert_called_with("paas", self.paas_id, self.order_id) |
| |
| def test_paas_lcm_edit_exception_updating_db(self): |
| self.paas_lcm.db.get_one.return_value = self.db_paas |
| self.paas_lcm.update_db_2.side_effect = DbException("failed") |
| self.loop.run_until_complete( |
| self.paas_lcm.edit(self.paas_content, self.order_id) |
| ) |
| |
| self.lcm_tasks.lock_HA.assert_called_with("paas", "edit", self.op_id) |
| self.paas_lcm.db.encrypt_decrypt_fields.assert_called_with( |
| self.db_paas, "decrypt", ["secret"], schema_version="1.11", salt="_id" |
| ) |
| self.check_db_update_when_successful_connectivity() |
| self.lcm_tasks.unlock_HA.assert_not_called() |
| self.lcm_tasks.remove.assert_called_with("paas", self.paas_id, self.order_id) |
| |
| def test_paas_lcm_edit_exception_getting_from_db(self): |
| self.paas_lcm.db.get_one.side_effect = DbException("failed") |
| self.loop.run_until_complete( |
| self.paas_lcm.edit(self.paas_content, self.order_id) |
| ) |
| self.lcm_tasks.lock_HA.assert_called_with("paas", "edit", self.op_id) |
| self.paas_lcm.db.encrypt_decrypt_fields.assert_not_called() |
| self.check_db_update_when_db_exception() |
| self.lcm_tasks.unlock_HA.assert_called_with( |
| "paas", |
| "edit", |
| self.op_id, |
| operationState="FAILED", |
| detailed_status="Failed with exception: database exception failed", |
| ) |
| self.lcm_tasks.remove.assert_called_with("paas", self.paas_id, self.order_id) |
| |
| def test_paas_lcm_delete_legacy_nbi(self): |
| self.lcm_tasks.lock_HA.return_value = False |
| self.loop.run_until_complete( |
| self.paas_lcm.delete(self.paas_content, self.order_id) |
| ) |
| self.check_assert_not_called_when_legacy_nbi() |
| |
| def test_paas_lcm_delete(self): |
| self.loop.run_until_complete( |
| self.paas_lcm.delete(self.paas_content, self.order_id) |
| ) |
| |
| self.lcm_tasks.lock_HA.assert_called_with("paas", "delete", self.op_id) |
| self.paas_lcm.db.del_one.assert_called_with("paas", {"_id": self.paas_id}) |
| self.paas_lcm.update_db_2.assert_called_with("paas", self.paas_id, None) |
| self.lcm_tasks.unlock_HA.assert_called_with( |
| "paas", |
| "delete", |
| self.op_id, |
| operationState="COMPLETED", |
| detailed_status="deleted", |
| ) |
| self.lcm_tasks.remove.assert_called_with("paas", self.paas_id, self.order_id) |
| |
| def test_paas_lcm_delete_exception_deleting_from_db(self): |
| self.paas_lcm.db.del_one.side_effect = Exception("failed deleting") |
| self.loop.run_until_complete( |
| self.paas_lcm.delete(self.paas_content, self.order_id) |
| ) |
| |
| self.lcm_tasks.lock_HA.assert_called_with("paas", "delete", self.op_id) |
| self.paas_lcm.db.del_one.assert_called_with("paas", {"_id": self.paas_id}) |
| self.paas_lcm.update_db_2.assert_called_with( |
| "paas", |
| self.paas_id, |
| { |
| "_admin.operationalState": "ERROR", |
| "_admin.detailed-status": "Failed with exception: failed deleting", |
| }, |
| ) |
| self.lcm_tasks.unlock_HA.assert_called_with( |
| "paas", |
| "delete", |
| self.op_id, |
| operationState="FAILED", |
| detailed_status="Failed with exception: failed deleting", |
| ) |
| self.lcm_tasks.remove.assert_called_with("paas", self.paas_id, self.order_id) |
| |
| def test_paas_lcm_delete_exception_updating_db(self): |
| self.loop.run_until_complete( |
| self.paas_lcm.delete(self.paas_content, self.order_id) |
| ) |
| |
| self.lcm_tasks.lock_HA.assert_called_with("paas", "delete", self.op_id) |
| self.paas_lcm.db.del_one.assert_called_with("paas", {"_id": self.paas_id}) |
| self.paas_lcm.update_db_2.assert_called_with("paas", self.paas_id, None) |
| self.lcm_tasks.unlock_HA.not_called() |
| self.lcm_tasks.remove.assert_called_with("paas", self.paas_id, self.order_id) |
| |
| def test_paas_lcm_delete_exception_deleting_and_updating_db(self): |
| self.paas_lcm.db.del_one.side_effect = Exception("failed deleting") |
| self.paas_lcm.update_db_2.side_effect = DbException("failed") |
| |
| self.loop.run_until_complete( |
| self.paas_lcm.delete(self.paas_content, self.order_id) |
| ) |
| |
| self.lcm_tasks.lock_HA.assert_called_with("paas", "delete", self.op_id) |
| self.paas_lcm.db.del_one.assert_called_with("paas", {"_id": self.paas_id}) |
| self.paas_lcm.update_db_2.assert_called_with( |
| "paas", |
| self.paas_id, |
| { |
| "_admin.operationalState": "ERROR", |
| "_admin.detailed-status": "Failed with exception: failed deleting", |
| }, |
| ) |
| self.lcm_tasks.unlock_HA.not_called() |
| self.lcm_tasks.remove.assert_called_with("paas", self.paas_id, self.order_id) |