import unittest
from unittest import TestCase
-from unittest.mock import Mock
+from unittest.mock import Mock, patch, call
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
+from osm_nbi.admin_topics import (
+ ProjectTopicAuth,
+ RoleTopicAuth,
+ UserTopicAuth,
+ CommonVimWimSdn,
+ VcaTopic,
+)
from osm_nbi.engine import EngineException
from osm_nbi.authconn import AuthconnNotFoundException
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