__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,
+ VimAccountTopic,
)
from osm_nbi.engine import EngineException
from osm_nbi.authconn import AuthconnNotFoundException
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(
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):
)
+@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