__date__ = "$2019-10-019"
import unittest
+import random
from unittest import TestCase
-from unittest.mock import Mock, patch, call
+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 (
UserTopicAuth,
CommonVimWimSdn,
VcaTopic,
- PaasTopic,
+ VimAccountTopic,
)
from osm_nbi.engine import EngineException
from osm_nbi.authconn import AuthconnNotFoundException
mock_check_conflict_on_del.assert_not_called()
-class TestPaaSTopic(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.paas_topic = PaasTopic(self.db, self.fs, self.msg, self.auth)
-
- def test_format_on_new(self):
- content = {
- "_id": "id",
- "secret": "secret_to_encrypt",
- }
- self.db.encrypt.side_effect = ["encrypted_secret"]
-
- expecte_oid = "id:0"
- expected_num_operations = 1
- oid = self.paas_topic.format_on_new(content)
-
- self.assertEqual(oid, expecte_oid)
- self.assertEqual(content["secret"], "encrypted_secret")
- self.assertEqual(content["_admin"]["operationalState"], "PROCESSING")
- self.assertEqual(content["_admin"]["current_operation"], None)
- self.assertEqual(len(content["_admin"]["operations"]), expected_num_operations)
- self.assertEqual(
- content["_admin"]["operations"][0]["lcmOperationType"], "create"
- )
- self.db.encrypt.assert_called_with(
- "secret_to_encrypt", schema_version="1.11", salt="id"
- )
-
- @patch("osm_nbi.base_topic.BaseTopic._get_project_filter")
- def test_check_conflict_on_new(self, mock_get_project_filter):
- indata = {"name": "new_paas_name"}
- session = {}
- mock_get_project_filter.return_value = {}
- self.db.get_one.return_value = None
- self.paas_topic.check_conflict_on_new(session, indata)
-
- @patch("osm_nbi.base_topic.BaseTopic._get_project_filter")
- def test_check_conflict_on_new_raise_exception(self, mock_get_project_filter):
- indata = {"name": "new_paas_name"}
- session = {}
- mock_get_project_filter.return_value = {}
- self.db.get_one.return_value = ["Found_PaaS"]
- with self.assertRaises(EngineException):
- self.paas_topic.check_conflict_on_new(session, indata)
-
- @patch("osm_nbi.base_topic.BaseTopic._get_project_filter")
- def test_check_conflict_on_edit(self, mock_get_project_filter):
- edit_content = {"name": "new_paas_name"}
- final_content = {}
- session = {"force": None}
- mock_get_project_filter.return_value = {}
- self.db.get_one.return_value = None
- self.paas_topic.check_conflict_on_edit(
- session, final_content, edit_content, "id"
- )
-
- @patch("osm_nbi.base_topic.BaseTopic._get_project_filter")
- def test_check_conflict_on_edit_raise_exception(self, mock_get_project_filter):
- edit_content = {"name": "new_paas_name"}
- final_content = {}
- session = {"force": None}
- mock_get_project_filter.return_value = {}
- self.db.get_one.return_value = ["Found_PaaS"]
- with self.assertRaises(EngineException):
- self.paas_topic.check_conflict_on_edit(
- session, final_content, edit_content, "id"
- )
-
- def test_format_on_edit(self):
- edit_content = {
- "_id": "id",
- "secret": "secret_to_encrypt",
- }
- final_content = {
- "_id": "id",
- "_admin": {"operations": [{"lcmOperationType": "create"}]},
- "schema_version": "1.11",
- }
- self.db.encrypt.side_effect = ["encrypted_secret"]
- expected_oid = "id:1"
- expected_num_operations = 2
- print(self.paas_topic.password_to_encrypt)
- oid = self.paas_topic.format_on_edit(final_content, edit_content)
-
- self.assertEqual(oid, expected_oid)
- self.assertEqual(final_content["secret"], "encrypted_secret")
- self.assertEqual(
- len(final_content["_admin"]["operations"]), expected_num_operations
- )
- self.assertEqual(final_content["_admin"]["operationalState"], "PROCESSING")
- self.assertEqual(final_content["_admin"]["detailed-status"], "Editing")
- self.db.encrypt.assert_called_with(
- "secret_to_encrypt", schema_version="1.11", salt="id"
- )
-
-
class Test_ProjectTopicAuth(TestCase):
@classmethod
def setUpClass(cls):
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)}
+ quotas = {
+ "vnfds": random.SystemRandom().randint(0, 100),
+ "nsds": random.SystemRandom().randint(0, 100),
+ }
self.topic.edit(
self.fake_session, pid, {"name": new_name, "quotas": quotas}
)
self.assertEqual(content["quotas"], quotas, "Wrong quotas")
with self.subTest(i=2):
new_name = "other-project-name"
- quotas = {"baditems": randint(0, 100)}
+ quotas = {"baditems": random.SystemRandom().randint(0, 100)}
self.auth.get_project_list.side_effect = [[proj], []]
with self.assertRaises(EngineException, msg="Accepted wrong quotas") as e:
self.topic.edit(
)
+@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()
projects / osm / NBI.git / blobdiff