osm_nbi/tests/test_admin_topics.py

#! /usr/bin/python3
# -*- coding: utf-8 -*-

# 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.

__author__ = "Pedro de la Cruz Ramos, pedro.delacruzramos@altran.com"
__date__ = "$2019-10-019"

import unittest
from unittest import TestCase
from unittest.mock import Mock, patch, call, ANY
from uuid import uuid4
from http import HTTPStatus
from time import time
from random import randint
from osm_common import dbbase, fsbase, msgbase
from osm_nbi import authconn, validation
from osm_nbi.admin_topics import (
    ProjectTopicAuth,
    RoleTopicAuth,
    UserTopicAuth,
    CommonVimWimSdn,
    VcaTopic,
    VimAccountTopic,
)
from osm_nbi.engine import EngineException
from osm_nbi.authconn import AuthconnNotFoundException


test_pid = str(uuid4())
test_name = "test-user"


def norm(str):
    """Normalize string for checking"""
    return " ".join(str.strip().split()).lower()


class TestVcaTopic(TestCase):
    def setUp(self):
        self.db = Mock(dbbase.DbBase())
        self.fs = Mock(fsbase.FsBase())
        self.msg = Mock(msgbase.MsgBase())
        self.auth = Mock(authconn.Authconn(None, None, None))
        self.vca_topic = VcaTopic(self.db, self.fs, self.msg, self.auth)

    @patch("osm_nbi.admin_topics.CommonVimWimSdn.format_on_new")
    def test_format_on_new(self, mock_super_format_on_new):
        content = {
            "_id": "id",
            "secret": "encrypted_secret",
            "cacert": "encrypted_cacert",
        }
        self.db.encrypt.side_effect = ["secret", "cacert"]
        mock_super_format_on_new.return_value = "1234"

        oid = self.vca_topic.format_on_new(content)

        self.assertEqual(oid, "1234")
        self.assertEqual(content["secret"], "secret")
        self.assertEqual(content["cacert"], "cacert")
        self.db.encrypt.assert_has_calls(
            [
                call("encrypted_secret", schema_version="1.11", salt="id"),
                call("encrypted_cacert", schema_version="1.11", salt="id"),
            ]
        )
        mock_super_format_on_new.assert_called_with(content, None, False)

    @patch("osm_nbi.admin_topics.CommonVimWimSdn.format_on_edit")
    def test_format_on_edit(self, mock_super_format_on_edit):
        edit_content = {
            "_id": "id",
            "secret": "encrypted_secret",
            "cacert": "encrypted_cacert",
        }
        final_content = {
            "_id": "id",
            "schema_version": "1.11",
        }
        self.db.encrypt.side_effect = ["secret", "cacert"]
        mock_super_format_on_edit.return_value = "1234"

        oid = self.vca_topic.format_on_edit(final_content, edit_content)

        self.assertEqual(oid, "1234")
        self.assertEqual(final_content["secret"], "secret")
        self.assertEqual(final_content["cacert"], "cacert")
        self.db.encrypt.assert_has_calls(
            [
                call("encrypted_secret", schema_version="1.11", salt="id"),
                call("encrypted_cacert", schema_version="1.11", salt="id"),
            ]
        )
        mock_super_format_on_edit.assert_called()

    @patch("osm_nbi.admin_topics.CommonVimWimSdn.check_conflict_on_del")
    def test_check_conflict_on_del(self, mock_check_conflict_on_del):
        session = {
            "project_id": "project-id",
            "force": False,
        }
        _id = "vca-id"
        db_content = {}

        self.db.get_list.return_value = None

        self.vca_topic.check_conflict_on_del(session, _id, db_content)

        self.db.get_list.assert_called_with(
            "vim_accounts",
            {"vca": _id, "_admin.projects_read.cont": "project-id"},
        )
        mock_check_conflict_on_del.assert_called_with(session, _id, db_content)

    @patch("osm_nbi.admin_topics.CommonVimWimSdn.check_conflict_on_del")
    def test_check_conflict_on_del_force(self, mock_check_conflict_on_del):
        session = {
            "project_id": "project-id",
            "force": True,
        }
        _id = "vca-id"
        db_content = {}

        self.vca_topic.check_conflict_on_del(session, _id, db_content)

        self.db.get_list.assert_not_called()
        mock_check_conflict_on_del.assert_not_called()

    @patch("osm_nbi.admin_topics.CommonVimWimSdn.check_conflict_on_del")
    def test_check_conflict_on_del_with_conflict(self, mock_check_conflict_on_del):
        session = {
            "project_id": "project-id",
            "force": False,
        }
        _id = "vca-id"
        db_content = {}

        self.db.get_list.return_value = {"_id": "vim", "vca": "vca-id"}

        with self.assertRaises(EngineException) as context:
            self.vca_topic.check_conflict_on_del(session, _id, db_content)
            self.assertEqual(
                context.exception,
                EngineException(
                    "There is at least one VIM account using this vca",
                    http_code=HTTPStatus.CONFLICT,
                ),
            )

        self.db.get_list.assert_called_with(
            "vim_accounts",
            {"vca": _id, "_admin.projects_read.cont": "project-id"},
        )
        mock_check_conflict_on_del.assert_not_called()


class Test_ProjectTopicAuth(TestCase):
    @classmethod
    def setUpClass(cls):
        cls.test_name = "test-project-topic"

    def setUp(self):
        self.db = Mock(dbbase.DbBase())
        self.fs = Mock(fsbase.FsBase())
        self.msg = Mock(msgbase.MsgBase())
        self.auth = Mock(authconn.Authconn(None, None, None))
        self.topic = ProjectTopicAuth(self.db, self.fs, self.msg, self.auth)
        self.fake_session = {
            "username": self.test_name,
            "project_id": (test_pid,),
            "method": None,
            "admin": True,
            "force": False,
            "public": False,
            "allow_show_user_project_role": True,
        }
        self.topic.check_quota = Mock(return_value=None)  # skip quota

    def test_new_project(self):
        with self.subTest(i=1):
            rollback = []
            pid1 = str(uuid4())
            self.auth.get_project_list.return_value = []
            self.auth.create_project.return_value = pid1
            pid2, oid = self.topic.new(
                rollback, self.fake_session, {"name": self.test_name, "quotas": {}}
            )
            self.assertEqual(len(rollback), 1, "Wrong rollback length")
            self.assertEqual(pid2, pid1, "Wrong project identifier")
            content = self.auth.create_project.call_args[0][0]
            self.assertEqual(content["name"], self.test_name, "Wrong project name")
            self.assertEqual(content["quotas"], {}, "Wrong quotas")
            self.assertIsNotNone(content["_admin"]["created"], "Wrong creation time")
            self.assertEqual(
                content["_admin"]["modified"],
                content["_admin"]["created"],
                "Wrong modification time",
            )
        with self.subTest(i=2):
            rollback = []
            with self.assertRaises(EngineException, msg="Accepted wrong quotas") as e:
                self.topic.new(
                    rollback,
                    self.fake_session,
                    {"name": "other-project-name", "quotas": {"baditems": 10}},
                )
            self.assertEqual(len(rollback), 0, "Wrong rollback length")
            self.assertEqual(
                e.exception.http_code,
                HTTPStatus.UNPROCESSABLE_ENTITY,
                "Wrong HTTP status code",
            )
            self.assertIn(
                "format error at 'quotas' 'additional properties are not allowed ('{}' was unexpected)'".format(
                    "baditems"
                ),
                norm(str(e.exception)),
                "Wrong exception text",
            )

    def test_edit_project(self):
        now = time()
        pid = str(uuid4())
        proj = {
            "_id": pid,
            "name": self.test_name,
            "_admin": {"created": now, "modified": now},
        }
        with self.subTest(i=1):
            self.auth.get_project_list.side_effect = [[proj], []]
            new_name = "new-project-name"
            quotas = {"vnfds": randint(0, 100), "nsds": randint(0, 100)}
            self.topic.edit(
                self.fake_session, pid, {"name": new_name, "quotas": quotas}
            )
            _id, content = self.auth.update_project.call_args[0]
            self.assertEqual(_id, pid, "Wrong project identifier")
            self.assertEqual(content["_id"], pid, "Wrong project identifier")
            self.assertEqual(content["_admin"]["created"], now, "Wrong creation time")
            self.assertGreater(
                content["_admin"]["modified"], now, "Wrong modification time"
            )
            self.assertEqual(content["name"], new_name, "Wrong project name")
            self.assertEqual(content["quotas"], quotas, "Wrong quotas")
        with self.subTest(i=2):
            new_name = "other-project-name"
            quotas = {"baditems": randint(0, 100)}
            self.auth.get_project_list.side_effect = [[proj], []]
            with self.assertRaises(EngineException, msg="Accepted wrong quotas") as e:
                self.topic.edit(
                    self.fake_session, pid, {"name": new_name, "quotas": quotas}
                )
            self.assertEqual(
                e.exception.http_code,
                HTTPStatus.UNPROCESSABLE_ENTITY,
                "Wrong HTTP status code",
            )
            self.assertIn(
                "format error at 'quotas' 'additional properties are not allowed ('{}' was unexpected)'".format(
                    "baditems"
                ),
                norm(str(e.exception)),
                "Wrong exception text",
            )

    def test_conflict_on_new(self):
        with self.subTest(i=1):
            rollback = []
            pid = str(uuid4())
            with self.assertRaises(
                EngineException, msg="Accepted uuid as project name"
            ) as e:
                self.topic.new(rollback, self.fake_session, {"name": pid})
            self.assertEqual(len(rollback), 0, "Wrong rollback length")
            self.assertEqual(
                e.exception.http_code,
                HTTPStatus.UNPROCESSABLE_ENTITY,
                "Wrong HTTP status code",
            )
            self.assertIn(
                "project name '{}' cannot have an uuid format".format(pid),
                norm(str(e.exception)),
                "Wrong exception text",
            )
        with self.subTest(i=2):
            rollback = []
            self.auth.get_project_list.return_value = [
                {"_id": test_pid, "name": self.test_name}
            ]
            with self.assertRaises(
                EngineException, msg="Accepted existing project name"
            ) as e:
                self.topic.new(rollback, self.fake_session, {"name": self.test_name})
            self.assertEqual(len(rollback), 0, "Wrong rollback length")
            self.assertEqual(
                e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
            )
            self.assertIn(
                "project '{}' exists".format(self.test_name),
                norm(str(e.exception)),
                "Wrong exception text",
            )

    def test_conflict_on_edit(self):
        with self.subTest(i=1):
            self.auth.get_project_list.return_value = [
                {"_id": test_pid, "name": self.test_name}
            ]
            new_name = str(uuid4())
            with self.assertRaises(
                EngineException, msg="Accepted uuid as project name"
            ) as e:
                self.topic.edit(self.fake_session, test_pid, {"name": new_name})
            self.assertEqual(
                e.exception.http_code,
                HTTPStatus.UNPROCESSABLE_ENTITY,
                "Wrong HTTP status code",
            )
            self.assertIn(
                "project name '{}' cannot have an uuid format".format(new_name),
                norm(str(e.exception)),
                "Wrong exception text",
            )
        with self.subTest(i=2):
            pid = str(uuid4())
            self.auth.get_project_list.return_value = [{"_id": pid, "name": "admin"}]
            with self.assertRaises(
                EngineException, msg="Accepted renaming of project 'admin'"
            ) as e:
                self.topic.edit(self.fake_session, pid, {"name": "new-name"})
            self.assertEqual(
                e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
            )
            self.assertIn(
                "you cannot rename project 'admin'",
                norm(str(e.exception)),
                "Wrong exception text",
            )
        with self.subTest(i=3):
            new_name = "new-project-name"
            self.auth.get_project_list.side_effect = [
                [{"_id": test_pid, "name": self.test_name}],
                [{"_id": str(uuid4()), "name": new_name}],
            ]
            with self.assertRaises(
                EngineException, msg="Accepted existing project name"
            ) as e:
                self.topic.edit(self.fake_session, pid, {"name": new_name})
            self.assertEqual(
                e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
            )
            self.assertIn(
                "project '{}' is already used".format(new_name),
                norm(str(e.exception)),
                "Wrong exception text",
            )

    def test_delete_project(self):
        with self.subTest(i=1):
            pid = str(uuid4())
            self.auth.get_project.return_value = {
                "_id": pid,
                "name": "other-project-name",
            }
            self.auth.delete_project.return_value = {"deleted": 1}
            self.auth.get_user_list.return_value = []
            self.db.get_list.return_value = []
            rc = self.topic.delete(self.fake_session, pid)
            self.assertEqual(rc, {"deleted": 1}, "Wrong project deletion return info")
            self.assertEqual(
                self.auth.get_project.call_args[0][0], pid, "Wrong project identifier"
            )
            self.assertEqual(
                self.auth.delete_project.call_args[0][0],
                pid,
                "Wrong project identifier",
            )

    def test_conflict_on_del(self):
        with self.subTest(i=1):
            self.auth.get_project.return_value = {
                "_id": test_pid,
                "name": self.test_name,
            }
            with self.assertRaises(
                EngineException, msg="Accepted deletion of own project"
            ) as e:
                self.topic.delete(self.fake_session, self.test_name)
            self.assertEqual(
                e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
            )
            self.assertIn(
                "you cannot delete your own project",
                norm(str(e.exception)),
                "Wrong exception text",
            )
        with self.subTest(i=2):
            self.auth.get_project.return_value = {"_id": str(uuid4()), "name": "admin"}
            with self.assertRaises(
                EngineException, msg="Accepted deletion of project 'admin'"
            ) as e:
                self.topic.delete(self.fake_session, "admin")
            self.assertEqual(
                e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
            )
            self.assertIn(
                "you cannot delete project 'admin'",
                norm(str(e.exception)),
                "Wrong exception text",
            )
        with self.subTest(i=3):
            pid = str(uuid4())
            name = "other-project-name"
            self.auth.get_project.return_value = {"_id": pid, "name": name}
            self.auth.get_user_list.return_value = [
                {
                    "_id": str(uuid4()),
                    "username": self.test_name,
                    "project_role_mappings": [{"project": pid, "role": str(uuid4())}],
                }
            ]
            with self.assertRaises(
                EngineException, msg="Accepted deletion of used project"
            ) as e:
                self.topic.delete(self.fake_session, pid)
            self.assertEqual(
                e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
            )
            self.assertIn(
                "project '{}' ({}) is being used by user '{}'".format(
                    name, pid, self.test_name
                ),
                norm(str(e.exception)),
                "Wrong exception text",
            )
        with self.subTest(i=4):
            self.auth.get_user_list.return_value = []
            self.db.get_list.return_value = [
                {
                    "_id": str(uuid4()),
                    "id": self.test_name,
                    "_admin": {"projects_read": [pid], "projects_write": []},
                }
            ]
            with self.assertRaises(
                EngineException, msg="Accepted deletion of used project"
            ) as e:
                self.topic.delete(self.fake_session, pid)
            self.assertEqual(
                e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
            )
            self.assertIn(
                "project '{}' ({}) is being used by {} '{}'".format(
                    name, pid, "vnf descriptor", self.test_name
                ),
                norm(str(e.exception)),
                "Wrong exception text",
            )


class Test_RoleTopicAuth(TestCase):
    @classmethod
    def setUpClass(cls):
        cls.test_name = "test-role-topic"
        cls.test_operations = ["tokens:get"]

    def setUp(self):
        self.db = Mock(dbbase.DbBase())
        self.fs = Mock(fsbase.FsBase())
        self.msg = Mock(msgbase.MsgBase())
        self.auth = Mock(authconn.Authconn(None, None, None))
        self.auth.role_permissions = self.test_operations
        self.topic = RoleTopicAuth(self.db, self.fs, self.msg, self.auth)
        self.fake_session = {
            "username": test_name,
            "project_id": (test_pid,),
            "method": None,
            "admin": True,
            "force": False,
            "public": False,
            "allow_show_user_project_role": True,
        }
        self.topic.check_quota = Mock(return_value=None)  # skip quota

    def test_new_role(self):
        with self.subTest(i=1):
            rollback = []
            rid1 = str(uuid4())
            perms_in = {"tokens": True}
            perms_out = {"default": False, "admin": False, "tokens": True}
            self.auth.get_role_list.return_value = []
            self.auth.create_role.return_value = rid1
            rid2, oid = self.topic.new(
                rollback,
                self.fake_session,
                {"name": self.test_name, "permissions": perms_in},
            )
            self.assertEqual(len(rollback), 1, "Wrong rollback length")
            self.assertEqual(rid2, rid1, "Wrong project identifier")
            content = self.auth.create_role.call_args[0][0]
            self.assertEqual(content["name"], self.test_name, "Wrong role name")
            self.assertEqual(content["permissions"], perms_out, "Wrong permissions")
            self.assertIsNotNone(content["_admin"]["created"], "Wrong creation time")
            self.assertEqual(
                content["_admin"]["modified"],
                content["_admin"]["created"],
                "Wrong modification time",
            )
        with self.subTest(i=2):
            rollback = []
            with self.assertRaises(
                EngineException, msg="Accepted wrong permissions"
            ) as e:
                self.topic.new(
                    rollback,
                    self.fake_session,
                    {"name": "other-role-name", "permissions": {"projects": True}},
                )
            self.assertEqual(len(rollback), 0, "Wrong rollback length")
            self.assertEqual(
                e.exception.http_code,
                HTTPStatus.UNPROCESSABLE_ENTITY,
                "Wrong HTTP status code",
            )
            self.assertIn(
                "invalid permission '{}'".format("projects"),
                norm(str(e.exception)),
                "Wrong exception text",
            )

    def test_edit_role(self):
        now = time()
        rid = str(uuid4())
        role = {
            "_id": rid,
            "name": self.test_name,
            "permissions": {"tokens": True},
            "_admin": {"created": now, "modified": now},
        }
        with self.subTest(i=1):
            self.auth.get_role_list.side_effect = [[role], []]
            self.auth.get_role.return_value = role
            new_name = "new-role-name"
            perms_in = {"tokens": False, "tokens:get": True}
            perms_out = {
                "default": False,
                "admin": False,
                "tokens": False,
                "tokens:get": True,
            }
            self.topic.edit(
                self.fake_session, rid, {"name": new_name, "permissions": perms_in}
            )
            content = self.auth.update_role.call_args[0][0]
            self.assertEqual(content["_id"], rid, "Wrong role identifier")
            self.assertEqual(content["_admin"]["created"], now, "Wrong creation time")
            self.assertGreater(
                content["_admin"]["modified"], now, "Wrong modification time"
            )
            self.assertEqual(content["name"], new_name, "Wrong role name")
            self.assertEqual(content["permissions"], perms_out, "Wrong permissions")
        with self.subTest(i=2):
            new_name = "other-role-name"
            perms_in = {"tokens": False, "tokens:post": True}
            self.auth.get_role_list.side_effect = [[role], []]
            with self.assertRaises(
                EngineException, msg="Accepted wrong permissions"
            ) as e:
                self.topic.edit(
                    self.fake_session, rid, {"name": new_name, "permissions": perms_in}
                )
            self.assertEqual(
                e.exception.http_code,
                HTTPStatus.UNPROCESSABLE_ENTITY,
                "Wrong HTTP status code",
            )
            self.assertIn(
                "invalid permission '{}'".format("tokens:post"),
                norm(str(e.exception)),
                "Wrong exception text",
            )

    def test_delete_role(self):
        with self.subTest(i=1):
            rid = str(uuid4())
            role = {"_id": rid, "name": "other-role-name"}
            self.auth.get_role_list.return_value = [role]
            self.auth.get_role.return_value = role
            self.auth.delete_role.return_value = {"deleted": 1}
            self.auth.get_user_list.return_value = []
            rc = self.topic.delete(self.fake_session, rid)
            self.assertEqual(rc, {"deleted": 1}, "Wrong role deletion return info")
            self.assertEqual(
                self.auth.get_role_list.call_args[0][0]["_id"],
                rid,
                "Wrong role identifier",
            )
            self.assertEqual(
                self.auth.get_role.call_args[0][0], rid, "Wrong role identifier"
            )
            self.assertEqual(
                self.auth.delete_role.call_args[0][0], rid, "Wrong role identifier"
            )

    def test_conflict_on_new(self):
        with self.subTest(i=1):
            rollback = []
            rid = str(uuid4())
            with self.assertRaises(
                EngineException, msg="Accepted uuid as role name"
            ) as e:
                self.topic.new(rollback, self.fake_session, {"name": rid})
            self.assertEqual(len(rollback), 0, "Wrong rollback length")
            self.assertEqual(
                e.exception.http_code,
                HTTPStatus.UNPROCESSABLE_ENTITY,
                "Wrong HTTP status code",
            )
            self.assertIn(
                "role name '{}' cannot have an uuid format".format(rid),
                norm(str(e.exception)),
                "Wrong exception text",
            )
        with self.subTest(i=2):
            rollback = []
            self.auth.get_role_list.return_value = [
                {"_id": str(uuid4()), "name": self.test_name}
            ]
            with self.assertRaises(
                EngineException, msg="Accepted existing role name"
            ) as e:
                self.topic.new(rollback, self.fake_session, {"name": self.test_name})
            self.assertEqual(len(rollback), 0, "Wrong rollback length")
            self.assertEqual(
                e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
            )
            self.assertIn(
                "role name '{}' exists".format(self.test_name),
                norm(str(e.exception)),
                "Wrong exception text",
            )

    def test_conflict_on_edit(self):
        rid = str(uuid4())
        with self.subTest(i=1):
            self.auth.get_role_list.return_value = [
                {"_id": rid, "name": self.test_name, "permissions": {}}
            ]
            new_name = str(uuid4())
            with self.assertRaises(
                EngineException, msg="Accepted uuid as role name"
            ) as e:
                self.topic.edit(self.fake_session, rid, {"name": new_name})
            self.assertEqual(
                e.exception.http_code,
                HTTPStatus.UNPROCESSABLE_ENTITY,
                "Wrong HTTP status code",
            )
            self.assertIn(
                "role name '{}' cannot have an uuid format".format(new_name),
                norm(str(e.exception)),
                "Wrong exception text",
            )
        for i, role_name in enumerate(["system_admin", "project_admin"], start=2):
            with self.subTest(i=i):
                rid = str(uuid4())
                self.auth.get_role.return_value = {
                    "_id": rid,
                    "name": role_name,
                    "permissions": {},
                }
                with self.assertRaises(
                    EngineException,
                    msg="Accepted renaming of role '{}'".format(role_name),
                ) as e:
                    self.topic.edit(self.fake_session, rid, {"name": "new-name"})
                self.assertEqual(
                    e.exception.http_code,
                    HTTPStatus.FORBIDDEN,
                    "Wrong HTTP status code",
                )
                self.assertIn(
                    "you cannot rename role '{}'".format(role_name),
                    norm(str(e.exception)),
                    "Wrong exception text",
                )
        with self.subTest(i=i + 1):
            new_name = "new-role-name"
            self.auth.get_role_list.side_effect = [
                [{"_id": rid, "name": self.test_name, "permissions": {}}],
                [{"_id": str(uuid4()), "name": new_name, "permissions": {}}],
            ]
            self.auth.get_role.return_value = {
                "_id": rid,
                "name": self.test_name,
                "permissions": {},
            }
            with self.assertRaises(
                EngineException, msg="Accepted existing role name"
            ) as e:
                self.topic.edit(self.fake_session, rid, {"name": new_name})
            self.assertEqual(
                e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
            )
            self.assertIn(
                "role name '{}' exists".format(new_name),
                norm(str(e.exception)),
                "Wrong exception text",
            )

    def test_conflict_on_del(self):
        for i, role_name in enumerate(["system_admin", "project_admin"], start=1):
            with self.subTest(i=i):
                rid = str(uuid4())
                role = {"_id": rid, "name": role_name}
                self.auth.get_role_list.return_value = [role]
                self.auth.get_role.return_value = role
                with self.assertRaises(
                    EngineException,
                    msg="Accepted deletion of role '{}'".format(role_name),
                ) as e:
                    self.topic.delete(self.fake_session, rid)
                self.assertEqual(
                    e.exception.http_code,
                    HTTPStatus.FORBIDDEN,
                    "Wrong HTTP status code",
                )
                self.assertIn(
                    "you cannot delete role '{}'".format(role_name),
                    norm(str(e.exception)),
                    "Wrong exception text",
                )
        with self.subTest(i=i + 1):
            rid = str(uuid4())
            name = "other-role-name"
            role = {"_id": rid, "name": name}
            self.auth.get_role_list.return_value = [role]
            self.auth.get_role.return_value = role
            self.auth.get_user_list.return_value = [
                {
                    "_id": str(uuid4()),
                    "username": self.test_name,
                    "project_role_mappings": [{"project": str(uuid4()), "role": rid}],
                }
            ]
            with self.assertRaises(
                EngineException, msg="Accepted deletion of used role"
            ) as e:
                self.topic.delete(self.fake_session, rid)
            self.assertEqual(
                e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
            )
            self.assertIn(
                "role '{}' ({}) is being used by user '{}'".format(
                    name, rid, self.test_name
                ),
                norm(str(e.exception)),
                "Wrong exception text",
            )


class Test_UserTopicAuth(TestCase):
    @classmethod
    def setUpClass(cls):
        cls.test_name = "test-user-topic"

    def setUp(self):
        self.db = Mock(dbbase.DbBase())
        self.fs = Mock(fsbase.FsBase())
        self.msg = Mock(msgbase.MsgBase())
        self.auth = Mock(authconn.Authconn(None, None, None))
        self.topic = UserTopicAuth(self.db, self.fs, self.msg, self.auth)
        self.fake_session = {
            "username": test_name,
            "project_id": (test_pid,),
            "method": None,
            "admin": True,
            "force": False,
            "public": False,
            "allow_show_user_project_role": True,
        }
        self.topic.check_quota = Mock(return_value=None)  # skip quota

    def test_new_user(self):
        uid1 = str(uuid4())
        pid = str(uuid4())
        self.auth.get_user_list.return_value = []
        self.auth.get_project.return_value = {"_id": pid, "name": "some_project"}
        self.auth.create_user.return_value = {"_id": uid1, "username": self.test_name}
        with self.subTest(i=1):
            rollback = []
            rid = str(uuid4())
            self.auth.get_role.return_value = {"_id": rid, "name": "some_role"}
            prms_in = [{"project": "some_project", "role": "some_role"}]
            prms_out = [{"project": pid, "role": rid}]
            uid2, oid = self.topic.new(
                rollback,
                self.fake_session,
                {
                    "username": self.test_name,
                    "password": self.test_name,
                    "project_role_mappings": prms_in,
                },
            )
            self.assertEqual(len(rollback), 1, "Wrong rollback length")
            self.assertEqual(uid2, uid1, "Wrong project identifier")
            content = self.auth.create_user.call_args[0][0]
            self.assertEqual(content["username"], self.test_name, "Wrong project name")
            self.assertEqual(content["password"], self.test_name, "Wrong password")
            self.assertEqual(
                content["project_role_mappings"],
                prms_out,
                "Wrong project-role mappings",
            )
            self.assertIsNotNone(content["_admin"]["created"], "Wrong creation time")
            self.assertEqual(
                content["_admin"]["modified"],
                content["_admin"]["created"],
                "Wrong modification time",
            )
        with self.subTest(i=2):
            rollback = []
            def_rid = str(uuid4())
            def_role = {"_id": def_rid, "name": "project_admin"}
            self.auth.get_role.return_value = def_role
            self.auth.get_role_list.return_value = [def_role]
            prms_out = [{"project": pid, "role": def_rid}]
            uid2, oid = self.topic.new(
                rollback,
                self.fake_session,
                {
                    "username": self.test_name,
                    "password": self.test_name,
                    "projects": ["some_project"],
                },
            )
            self.assertEqual(len(rollback), 1, "Wrong rollback length")
            self.assertEqual(uid2, uid1, "Wrong project identifier")
            content = self.auth.create_user.call_args[0][0]
            self.assertEqual(content["username"], self.test_name, "Wrong project name")
            self.assertEqual(content["password"], self.test_name, "Wrong password")
            self.assertEqual(
                content["project_role_mappings"],
                prms_out,
                "Wrong project-role mappings",
            )
            self.assertIsNotNone(content["_admin"]["created"], "Wrong creation time")
            self.assertEqual(
                content["_admin"]["modified"],
                content["_admin"]["created"],
                "Wrong modification time",
            )
        with self.subTest(i=3):
            rollback = []
            with self.assertRaises(
                EngineException, msg="Accepted wrong project-role mappings"
            ) as e:
                self.topic.new(
                    rollback,
                    self.fake_session,
                    {
                        "username": "other-project-name",
                        "password": "other-password",
                        "project_role_mappings": [{}],
                    },
                )
            self.assertEqual(len(rollback), 0, "Wrong rollback length")
            self.assertEqual(
                e.exception.http_code,
                HTTPStatus.UNPROCESSABLE_ENTITY,
                "Wrong HTTP status code",
            )
            self.assertIn(
                "format error at '{}' '{}'".format(
                    "project_role_mappings:{}", "'{}' is a required property"
                ).format(0, "project"),
                norm(str(e.exception)),
                "Wrong exception text",
            )
        with self.subTest(i=4):
            rollback = []
            with self.assertRaises(EngineException, msg="Accepted wrong projects") as e:
                self.topic.new(
                    rollback,
                    self.fake_session,
                    {
                        "username": "other-project-name",
                        "password": "other-password",
                        "projects": [],
                    },
                )
            self.assertEqual(len(rollback), 0, "Wrong rollback length")
            self.assertEqual(
                e.exception.http_code,
                HTTPStatus.UNPROCESSABLE_ENTITY,
                "Wrong HTTP status code",
            )
            self.assertIn(
                "format error at '{}' '{}'".format(
                    "projects", "{} is too short"
                ).format([]),
                norm(str(e.exception)),
                "Wrong exception text",
            )

    def test_edit_user(self):
        now = time()
        uid = str(uuid4())
        pid1 = str(uuid4())
        rid1 = str(uuid4())
        prms = [
            {
                "project": pid1,
                "project_name": "project-1",
                "role": rid1,
                "role_name": "role-1",
            }
        ]
        user = {
            "_id": uid,
            "username": self.test_name,
            "project_role_mappings": prms,
            "_admin": {"created": now, "modified": now},
        }
        with self.subTest(i=1):
            self.auth.get_user_list.side_effect = [[user], []]
            self.auth.get_user.return_value = user
            pid2 = str(uuid4())
            rid2 = str(uuid4())
            self.auth.get_project.side_effect = [
                {"_id": pid2, "name": "project-2"},
                {"_id": pid1, "name": "project-1"},
            ]
            self.auth.get_role.side_effect = [
                {"_id": rid2, "name": "role-2"},
                {"_id": rid1, "name": "role-1"},
            ]
            new_name = "new-user-name"
            new_pasw = "new-password"
            add_prms = [{"project": pid2, "role": rid2}]
            rem_prms = [{"project": pid1, "role": rid1}]
            self.topic.edit(
                self.fake_session,
                uid,
                {
                    "username": new_name,
                    "password": new_pasw,
                    "add_project_role_mappings": add_prms,
                    "remove_project_role_mappings": rem_prms,
                },
            )
            content = self.auth.update_user.call_args[0][0]
            self.assertEqual(content["_id"], uid, "Wrong user identifier")
            self.assertEqual(content["username"], new_name, "Wrong user name")
            self.assertEqual(content["password"], new_pasw, "Wrong user password")
            self.assertEqual(
                content["add_project_role_mappings"],
                add_prms,
                "Wrong project-role mappings to add",
            )
            self.assertEqual(
                content["remove_project_role_mappings"],
                prms,
                "Wrong project-role mappings to remove",
            )
        with self.subTest(i=2):
            new_name = "other-user-name"
            new_prms = [{}]
            self.auth.get_role_list.side_effect = [[user], []]
            self.auth.get_user_list.side_effect = [[user]]
            with self.assertRaises(
                EngineException, msg="Accepted wrong project-role mappings"
            ) as e:
                self.topic.edit(
                    self.fake_session,
                    uid,
                    {"username": new_name, "project_role_mappings": new_prms},
                )
            self.assertEqual(
                e.exception.http_code,
                HTTPStatus.UNPROCESSABLE_ENTITY,
                "Wrong HTTP status code",
            )
            self.assertIn(
                "format error at '{}' '{}'".format(
                    "project_role_mappings:{}", "'{}' is a required property"
                ).format(0, "project"),
                norm(str(e.exception)),
                "Wrong exception text",
            )
        with self.subTest(i=3):
            self.auth.get_user_list.side_effect = [[user], []]
            self.auth.get_user.return_value = user
            old_password = self.test_name
            new_pasw = "new-password"
            self.topic.edit(
                self.fake_session,
                uid,
                {
                    "old_password": old_password,
                    "password": new_pasw,
                },
            )
            content = self.auth.update_user.call_args[0][0]
            self.assertEqual(
                content["old_password"], old_password, "Wrong old password"
            )
            self.assertEqual(content["password"], new_pasw, "Wrong user password")

    def test_delete_user(self):
        with self.subTest(i=1):
            uid = str(uuid4())
            self.fake_session["username"] = self.test_name
            user = user = {
                "_id": uid,
                "username": "other-user-name",
                "project_role_mappings": [],
            }
            self.auth.get_user.return_value = user
            self.auth.delete_user.return_value = {"deleted": 1}
            rc = self.topic.delete(self.fake_session, uid)
            self.assertEqual(rc, {"deleted": 1}, "Wrong user deletion return info")
            self.assertEqual(
                self.auth.get_user.call_args[0][0], uid, "Wrong user identifier"
            )
            self.assertEqual(
                self.auth.delete_user.call_args[0][0], uid, "Wrong user identifier"
            )

    def test_conflict_on_new(self):
        with self.subTest(i=1):
            rollback = []
            uid = str(uuid4())
            with self.assertRaises(
                EngineException, msg="Accepted uuid as username"
            ) as e:
                self.topic.new(
                    rollback,
                    self.fake_session,
                    {
                        "username": uid,
                        "password": self.test_name,
                        "projects": [test_pid],
                    },
                )
            self.assertEqual(len(rollback), 0, "Wrong rollback length")
            self.assertEqual(
                e.exception.http_code,
                HTTPStatus.UNPROCESSABLE_ENTITY,
                "Wrong HTTP status code",
            )
            self.assertIn(
                "username '{}' cannot have a uuid format".format(uid),
                norm(str(e.exception)),
                "Wrong exception text",
            )
        with self.subTest(i=2):
            rollback = []
            self.auth.get_user_list.return_value = [
                {"_id": str(uuid4()), "username": self.test_name}
            ]
            with self.assertRaises(
                EngineException, msg="Accepted existing username"
            ) as e:
                self.topic.new(
                    rollback,
                    self.fake_session,
                    {
                        "username": self.test_name,
                        "password": self.test_name,
                        "projects": [test_pid],
                    },
                )
            self.assertEqual(len(rollback), 0, "Wrong rollback length")
            self.assertEqual(
                e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
            )
            self.assertIn(
                "username '{}' is already used".format(self.test_name),
                norm(str(e.exception)),
                "Wrong exception text",
            )
        with self.subTest(i=3):
            rollback = []
            self.auth.get_user_list.return_value = []
            self.auth.get_role_list.side_effect = [[], []]
            with self.assertRaises(
                AuthconnNotFoundException, msg="Accepted user without default role"
            ) as e:
                self.topic.new(
                    rollback,
                    self.fake_session,
                    {
                        "username": self.test_name,
                        "password": self.test_name,
                        "projects": [str(uuid4())],
                    },
                )
            self.assertEqual(len(rollback), 0, "Wrong rollback length")
            self.assertEqual(
                e.exception.http_code, HTTPStatus.NOT_FOUND, "Wrong HTTP status code"
            )
            self.assertIn(
                "can't find default role for user '{}'".format(self.test_name),
                norm(str(e.exception)),
                "Wrong exception text",
            )

    def test_conflict_on_edit(self):
        uid = str(uuid4())
        with self.subTest(i=1):
            self.auth.get_user_list.return_value = [
                {"_id": uid, "username": self.test_name}
            ]
            new_name = str(uuid4())
            with self.assertRaises(
                EngineException, msg="Accepted uuid as username"
            ) as e:
                self.topic.edit(self.fake_session, uid, {"username": new_name})
            self.assertEqual(
                e.exception.http_code,
                HTTPStatus.UNPROCESSABLE_ENTITY,
                "Wrong HTTP status code",
            )
            self.assertIn(
                "username '{}' cannot have an uuid format".format(new_name),
                norm(str(e.exception)),
                "Wrong exception text",
            )
        with self.subTest(i=2):
            self.auth.get_user_list.return_value = [
                {"_id": uid, "username": self.test_name}
            ]
            self.auth.get_role_list.side_effect = [[], []]
            with self.assertRaises(
                AuthconnNotFoundException, msg="Accepted user without default role"
            ) as e:
                self.topic.edit(self.fake_session, uid, {"projects": [str(uuid4())]})
            self.assertEqual(
                e.exception.http_code, HTTPStatus.NOT_FOUND, "Wrong HTTP status code"
            )
            self.assertIn(
                "can't find a default role for user '{}'".format(self.test_name),
                norm(str(e.exception)),
                "Wrong exception text",
            )
        with self.subTest(i=3):
            admin_uid = str(uuid4())
            self.auth.get_user_list.return_value = [
                {"_id": admin_uid, "username": "admin"}
            ]
            with self.assertRaises(
                EngineException,
                msg="Accepted removing system_admin role from admin user",
            ) as e:
                self.topic.edit(
                    self.fake_session,
                    admin_uid,
                    {
                        "remove_project_role_mappings": [
                            {"project": "admin", "role": "system_admin"}
                        ]
                    },
                )
            self.assertEqual(
                e.exception.http_code, HTTPStatus.FORBIDDEN, "Wrong HTTP status code"
            )
            self.assertIn(
                "you cannot remove system_admin role from admin user",
                norm(str(e.exception)),
                "Wrong exception text",
            )
        with self.subTest(i=4):
            new_name = "new-user-name"
            self.auth.get_user_list.side_effect = [
                [{"_id": uid, "name": self.test_name}],
                [{"_id": str(uuid4()), "name": new_name}],
            ]
            with self.assertRaises(
                EngineException, msg="Accepted existing username"
            ) as e:
                self.topic.edit(self.fake_session, uid, {"username": new_name})
            self.assertEqual(
                e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
            )
            self.assertIn(
                "username '{}' is already used".format(new_name),
                norm(str(e.exception)),
                "Wrong exception text",
            )

    def test_conflict_on_del(self):
        with self.subTest(i=1):
            uid = str(uuid4())
            self.fake_session["username"] = self.test_name
            user = user = {
                "_id": uid,
                "username": self.test_name,
                "project_role_mappings": [],
            }
            self.auth.get_user.return_value = user
            with self.assertRaises(
                EngineException, msg="Accepted deletion of own user"
            ) as e:
                self.topic.delete(self.fake_session, uid)
            self.assertEqual(
                e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
            )
            self.assertIn(
                "you cannot delete your own login user",
                norm(str(e.exception)),
                "Wrong exception text",
            )


class Test_CommonVimWimSdn(TestCase):
    @classmethod
    def setUpClass(cls):
        cls.test_name = "test-cim-topic"  # CIM = Common Infrastructure Manager

    def setUp(self):
        self.db = Mock(dbbase.DbBase())
        self.fs = Mock(fsbase.FsBase())
        self.msg = Mock(msgbase.MsgBase())
        self.auth = Mock(authconn.Authconn(None, None, None))
        self.topic = CommonVimWimSdn(self.db, self.fs, self.msg, self.auth)
        # Use WIM schemas for testing because they are the simplest
        self.topic._send_msg = Mock()
        self.topic.topic = "wims"
        self.topic.schema_new = validation.wim_account_new_schema
        self.topic.schema_edit = validation.wim_account_edit_schema
        self.fake_session = {
            "username": test_name,
            "project_id": (test_pid,),
            "method": None,
            "admin": True,
            "force": False,
            "public": False,
            "allow_show_user_project_role": True,
        }
        self.topic.check_quota = Mock(return_value=None)  # skip quota

    def test_new_cvws(self):
        test_url = "http://0.0.0.0:0"
        with self.subTest(i=1):
            rollback = []
            test_type = "fake"
            self.db.get_one.return_value = None
            self.db.create.side_effect = lambda self, content: content["_id"]
            cid, oid = self.topic.new(
                rollback,
                self.fake_session,
                {"name": self.test_name, "wim_url": test_url, "wim_type": test_type},
            )
            self.assertEqual(len(rollback), 1, "Wrong rollback length")
            args = self.db.create.call_args[0]
            content = args[1]
            self.assertEqual(args[0], self.topic.topic, "Wrong topic")
            self.assertEqual(content["_id"], cid, "Wrong CIM identifier")
            self.assertEqual(content["name"], self.test_name, "Wrong CIM name")
            self.assertEqual(content["wim_url"], test_url, "Wrong URL")
            self.assertEqual(content["wim_type"], test_type, "Wrong CIM type")
            self.assertEqual(content["schema_version"], "1.11", "Wrong schema version")
            self.assertEqual(content["op_id"], oid, "Wrong operation identifier")
            self.assertIsNotNone(content["_admin"]["created"], "Wrong creation time")
            self.assertEqual(
                content["_admin"]["modified"],
                content["_admin"]["created"],
                "Wrong modification time",
            )
            self.assertEqual(
                content["_admin"]["operationalState"],
                "PROCESSING",
                "Wrong operational state",
            )
            self.assertEqual(
                content["_admin"]["projects_read"],
                [test_pid],
                "Wrong read-only projects",
            )
            self.assertEqual(
                content["_admin"]["projects_write"],
                [test_pid],
                "Wrong read/write projects",
            )
            self.assertIsNone(
                content["_admin"]["current_operation"], "Wrong current operation"
            )
            self.assertEqual(
                len(content["_admin"]["operations"]), 1, "Wrong number of operations"
            )
            operation = content["_admin"]["operations"][0]
            self.assertEqual(
                operation["lcmOperationType"], "create", "Wrong operation type"
            )
            self.assertEqual(
                operation["operationState"], "PROCESSING", "Wrong operation state"
            )
            self.assertGreater(
                operation["startTime"],
                content["_admin"]["created"],
                "Wrong operation start time",
            )
            self.assertGreater(
                operation["statusEnteredTime"],
                content["_admin"]["created"],
                "Wrong operation status enter time",
            )
            self.assertEqual(
                operation["detailed-status"], "", "Wrong operation detailed status info"
            )
            self.assertIsNone(
                operation["operationParams"], "Wrong operation parameters"
            )
        # This test is disabled. From Feature 8030 we admit all WIM/SDN types
        # with self.subTest(i=2):
        #     rollback = []
        #     test_type = "bad_type"
        #     with self.assertRaises(EngineException, msg="Accepted wrong CIM type") as e:
        #         self.topic.new(rollback, self.fake_session,
        #                        {"name": self.test_name, "wim_url": test_url, "wim_type": test_type})
        #     self.assertEqual(len(rollback), 0, "Wrong rollback length")
        #     self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
        #     self.assertIn("format error at '{}' '{}".format("wim_type", "'{}' is not one of {}").format(test_type,""),
        #                   norm(str(e.exception)), "Wrong exception text")

    def test_conflict_on_new(self):
        with self.subTest(i=1):
            rollback = []
            test_url = "http://0.0.0.0:0"
            test_type = "fake"
            self.db.get_one.return_value = {"_id": str(uuid4()), "name": self.test_name}
            with self.assertRaises(
                EngineException, msg="Accepted existing CIM name"
            ) as e:
                self.topic.new(
                    rollback,
                    self.fake_session,
                    {
                        "name": self.test_name,
                        "wim_url": test_url,
                        "wim_type": test_type,
                    },
                )
            self.assertEqual(len(rollback), 0, "Wrong rollback length")
            self.assertEqual(
                e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
            )
            self.assertIn(
                "name '{}' already exists for {}".format(
                    self.test_name, self.topic.topic
                ),
                norm(str(e.exception)),
                "Wrong exception text",
            )

    def test_edit_cvws(self):
        now = time()
        cid = str(uuid4())
        test_url = "http://0.0.0.0:0"
        test_type = "fake"
        cvws = {
            "_id": cid,
            "name": self.test_name,
            "wim_url": test_url,
            "wim_type": test_type,
            "_admin": {
                "created": now,
                "modified": now,
                "operations": [{"lcmOperationType": "create"}],
            },
        }
        with self.subTest(i=1):
            new_name = "new-cim-name"
            new_url = "https://1.1.1.1:1"
            new_type = "onos"
            self.db.get_one.side_effect = [cvws, None]
            self.db.replace.return_value = {"updated": 1}
            # self.db.encrypt.side_effect = [b64str(), b64str()]
            self.topic.edit(
                self.fake_session,
                cid,
                {"name": new_name, "wim_url": new_url, "wim_type": new_type},
            )
            args = self.db.replace.call_args[0]
            content = args[2]
            self.assertEqual(args[0], self.topic.topic, "Wrong topic")
            self.assertEqual(args[1], cid, "Wrong CIM identifier")
            self.assertEqual(content["_id"], cid, "Wrong CIM identifier")
            self.assertEqual(content["name"], new_name, "Wrong CIM name")
            self.assertEqual(content["wim_type"], new_type, "Wrong CIM type")
            self.assertEqual(content["wim_url"], new_url, "Wrong URL")
            self.assertEqual(content["_admin"]["created"], now, "Wrong creation time")
            self.assertGreater(
                content["_admin"]["modified"],
                content["_admin"]["created"],
                "Wrong modification time",
            )
            self.assertEqual(
                len(content["_admin"]["operations"]), 2, "Wrong number of operations"
            )
            operation = content["_admin"]["operations"][1]
            self.assertEqual(
                operation["lcmOperationType"], "edit", "Wrong operation type"
            )
            self.assertEqual(
                operation["operationState"], "PROCESSING", "Wrong operation state"
            )
            self.assertGreater(
                operation["startTime"],
                content["_admin"]["modified"],
                "Wrong operation start time",
            )
            self.assertGreater(
                operation["statusEnteredTime"],
                content["_admin"]["modified"],
                "Wrong operation status enter time",
            )
            self.assertEqual(
                operation["detailed-status"], "", "Wrong operation detailed status info"
            )
            self.assertIsNone(
                operation["operationParams"], "Wrong operation parameters"
            )
        with self.subTest(i=2):
            self.db.get_one.side_effect = [cvws]
            with self.assertRaises(EngineException, msg="Accepted wrong property") as e:
                self.topic.edit(
                    self.fake_session,
                    str(uuid4()),
                    {"name": "new-name", "extra_prop": "anything"},
                )
            self.assertEqual(
                e.exception.http_code,
                HTTPStatus.UNPROCESSABLE_ENTITY,
                "Wrong HTTP status code",
            )
            self.assertIn(
                "format error '{}'".format(
                    "additional properties are not allowed ('{}' was unexpected)"
                ).format("extra_prop"),
                norm(str(e.exception)),
                "Wrong exception text",
            )

    def test_conflict_on_edit(self):
        with self.subTest(i=1):
            cid = str(uuid4())
            new_name = "new-cim-name"
            self.db.get_one.side_effect = [
                {"_id": cid, "name": self.test_name},
                {"_id": str(uuid4()), "name": new_name},
            ]
            with self.assertRaises(
                EngineException, msg="Accepted existing CIM name"
            ) as e:
                self.topic.edit(self.fake_session, cid, {"name": new_name})
            self.assertEqual(
                e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
            )
            self.assertIn(
                "name '{}' already exists for {}".format(new_name, self.topic.topic),
                norm(str(e.exception)),
                "Wrong exception text",
            )

    def test_delete_cvws(self):
        cid = str(uuid4())
        ro_pid = str(uuid4())
        rw_pid = str(uuid4())
        cvws = {"_id": cid, "name": self.test_name}
        self.db.get_list.return_value = []
        with self.subTest(i=1):
            cvws["_admin"] = {
                "projects_read": [test_pid, ro_pid, rw_pid],
                "projects_write": [test_pid, rw_pid],
            }
            self.db.get_one.return_value = cvws
            oid = self.topic.delete(self.fake_session, cid)
            self.assertIsNone(oid, "Wrong operation identifier")
            self.assertEqual(
                self.db.get_one.call_args[0][0], self.topic.topic, "Wrong topic"
            )
            self.assertEqual(
                self.db.get_one.call_args[0][1]["_id"], cid, "Wrong CIM identifier"
            )
            self.assertEqual(
                self.db.set_one.call_args[0][0], self.topic.topic, "Wrong topic"
            )
            self.assertEqual(
                self.db.set_one.call_args[0][1]["_id"], cid, "Wrong CIM identifier"
            )
            self.assertEqual(
                self.db.set_one.call_args[1]["update_dict"],
                None,
                "Wrong read-only projects update",
            )
            self.assertEqual(
                self.db.set_one.call_args[1]["pull_list"],
                {
                    "_admin.projects_read": (test_pid,),
                    "_admin.projects_write": (test_pid,),
                },
                "Wrong read/write projects update",
            )
            self.topic._send_msg.assert_not_called()
        with self.subTest(i=2):
            now = time()
            cvws["_admin"] = {
                "projects_read": [test_pid],
                "projects_write": [test_pid],
                "operations": [],
            }
            self.db.get_one.return_value = cvws
            oid = self.topic.delete(self.fake_session, cid)
            self.assertEqual(oid, cid + ":0", "Wrong operation identifier")
            self.assertEqual(
                self.db.get_one.call_args[0][0], self.topic.topic, "Wrong topic"
            )
            self.assertEqual(
                self.db.get_one.call_args[0][1]["_id"], cid, "Wrong CIM identifier"
            )
            self.assertEqual(
                self.db.set_one.call_args[0][0], self.topic.topic, "Wrong topic"
            )
            self.assertEqual(
                self.db.set_one.call_args[0][1]["_id"], cid, "Wrong user identifier"
            )
            self.assertEqual(
                self.db.set_one.call_args[1]["update_dict"],
                {"_admin.to_delete": True},
                "Wrong _admin.to_delete update",
            )
            operation = self.db.set_one.call_args[1]["push"]["_admin.operations"]
            self.assertEqual(
                operation["lcmOperationType"], "delete", "Wrong operation type"
            )
            self.assertEqual(
                operation["operationState"], "PROCESSING", "Wrong operation state"
            )
            self.assertEqual(
                operation["detailed-status"], "", "Wrong operation detailed status"
            )
            self.assertIsNone(
                operation["operationParams"], "Wrong operation parameters"
            )
            self.assertGreater(
                operation["startTime"], now, "Wrong operation start time"
            )
            self.assertGreater(
                operation["statusEnteredTime"], now, "Wrong operation status enter time"
            )
            self.topic._send_msg.assert_called_once_with(
                "delete", {"_id": cid, "op_id": cid + ":0"}, not_send_msg=None
            )
        with self.subTest(i=3):
            cvws["_admin"] = {
                "projects_read": [],
                "projects_write": [],
                "operations": [],
            }
            self.db.get_one.return_value = cvws
            self.topic._send_msg.reset_mock()
            self.db.get_one.reset_mock()
            self.db.del_one.reset_mock()
            self.fake_session["force"] = True  # to force deletion
            self.fake_session["admin"] = True  # to force deletion
            self.fake_session["project_id"] = []  # to force deletion
            oid = self.topic.delete(self.fake_session, cid)
            self.assertIsNone(oid, "Wrong operation identifier")
            self.assertEqual(
                self.db.get_one.call_args[0][0], self.topic.topic, "Wrong topic"
            )
            self.assertEqual(
                self.db.get_one.call_args[0][1]["_id"], cid, "Wrong CIM identifier"
            )
            self.assertEqual(
                self.db.del_one.call_args[0][0], self.topic.topic, "Wrong topic"
            )
            self.assertEqual(
                self.db.del_one.call_args[0][1]["_id"], cid, "Wrong CIM identifier"
            )
            self.topic._send_msg.assert_called_once_with(
                "deleted", {"_id": cid, "op_id": None}, not_send_msg=None
            )


@patch("osm_nbi.admin_topics.CommonVimWimSdn.check_conflict_on_new")
class TestVimAccountTopic(TestCase):
    def setUp(self):
        self.db = Mock(dbbase.DbBase())
        self.fs = Mock(fsbase.FsBase())
        self.msg = Mock(msgbase.MsgBase())
        self.auth = Mock(authconn.Authconn(None, None, None))
        self.topic = VimAccountTopic(self.db, self.fs, self.msg, self.auth)
        self.topic.check_quota = Mock(return_value=None)  # skip quota

        self.fake_session = {
            "username": test_name,
            "project_id": (test_pid,),
            "method": None,
            "admin": True,
            "force": False,
            "public": False,
            "allow_show_user_project_role": True,
        }

    def check_invalid_indata_raises_exception(self, indata, mock_common_vim_wim_sdn):
        with self.assertRaises(EngineException) as error:
            self.topic.check_conflict_on_new(self.fake_session, indata)
        mock_common_vim_wim_sdn.assert_called_with(self.fake_session, indata)
        error_msg = "Invalid config for VIM account '{}'.".format(indata["name"])
        self.assertEqual(str(error.exception), error_msg)

    def test_check_conflict_on_new_vim_type_paas(self, mock_common_vim_wim_sdn):
        valid_juju_paas_config = {
            "paas_provider": "juju",
            "ca_cert_content": "file_content",
            "cloud": "microk8s",
            "cloud_credentials": "cloud_credential_name",
            "authorized_keys": "keys",
        }
        indata = {
            "name": "juju_paas",
            "vim_type": "paas",
            "description": None,
            "vim_url": "http://0.0.0.0:0",
            "vim_user": "some_user",
            "vim_password": "some_password",
            "vim_tenant_name": "null",
            "config": valid_juju_paas_config,
        }
        self.topic.check_conflict_on_new(self.fake_session, indata)
        mock_common_vim_wim_sdn.assert_called_once_with(self.fake_session, indata)

    def test_check_conflict_on_new_vim_type_paas_config_missing(
        self, mock_common_vim_wim_sdn
    ):
        indata = {
            "name": "juju_paas",
            "vim_type": "paas",
            "description": None,
            "vim_url": "http://0.0.0.0:0",
            "vim_user": "some_user",
            "vim_password": "some_password",
            "vim_tenant_name": "null",
        }
        self.check_invalid_indata_raises_exception(indata, mock_common_vim_wim_sdn)

    def test_check_conflict_on_new_vim_type_paas_invalid_config(
        self, mock_common_vim_wim_sdn
    ):
        invalid_configs = [
            {
                "paas_provider": "some_paas_provider",
                "ca_cert_content": "file_content",
                "cloud": "microk8s",
                "cloud_credentials": "cloud_credential_name",
            },
            {
                "ca_cert_content": "file_content",
                "cloud": "microk8s",
                "cloud_credentials": "cloud_credential_name",
            },
            {
                "paas_provider": "juju",
                "cloud": "microk8s",
                "cloud_credentials": "cloud_credential_name",
            },
            {
                "paas_provider": "juju",
                "ca_cert_content": "file_content",
                "cloud_credentials": "cloud_credential_name",
            },
            {
                "paas_provider": "juju",
                "ca_cert_content": "file_content",
                "cloud": "microk8s",
            },
            {
                "some_param": None,
            },
            {},
        ]
        for config in invalid_configs:
            with self.subTest():
                indata = {
                    "name": "juju_paas",
                    "vim_type": "paas",
                    "description": None,
                    "vim_url": "http://0.0.0.0:0",
                    "vim_user": "some_user",
                    "vim_password": "some_password",
                    "vim_tenant_name": "null",
                    "config": config,
                }
                self.check_invalid_indata_raises_exception(
                    indata, mock_common_vim_wim_sdn
                )

    def test_kafka_message_is_not_sent_if_paas_vim(self, mock_common_vim_wim_sdn):
        valid_juju_paas_config = {
            "paas_provider": "juju",
            "ca_cert_content": "file_content",
            "cloud": "microk8s",
            "cloud_credentials": "cloud_credential_name",
            "authorized_keys": "keys",
        }
        indata = {
            "name": "juju_paas",
            "vim_type": "paas",
            "description": None,
            "vim_url": "http://0.0.0.0:0",
            "vim_user": "some_user",
            "vim_password": "some_password",
            "vim_tenant_name": "null",
            "config": valid_juju_paas_config,
        }
        rollback = []
        self.topic.temporal = Mock()

        self.topic.new(rollback, self.fake_session, indata)
        self.assertEqual(len(rollback), 1, "Wrong rollback length")
        self.msg.write.assert_not_called()
        self.topic.temporal.start_vim_workflow.assert_called_once()

    def test_kafka_message_is_sent_if_not_paas_vim(self, mock_common_vim_wim_sdn):
        indata = {
            "name": "juju_paas",
            "vim_type": "openstack",
            "description": None,
            "vim_url": "http://0.0.0.0:0",
            "vim_user": "some_user",
            "vim_password": "some_password",
            "vim_tenant_name": "null",
        }
        rollback = []

        self.topic.new(rollback, self.fake_session, indata)
        self.assertEqual(len(rollback), 1, "Wrong rollback length")
        mock_common_vim_wim_sdn.assert_called_once_with(self.fake_session, indata)
        self.msg.write.assert_called_once_with("vim_account", "created", ANY)


if __name__ == "__main__":
    unittest.main()