osm_lcm/tests/test_paas.py

#!/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)