| 1 |
|
#! /usr/bin/python3 |
| 2 |
|
# -*- coding: utf-8 -*- |
| 3 |
|
|
| 4 |
|
# Licensed under the Apache License, Version 2.0 (the "License"); |
| 5 |
|
# you may not use this file except in compliance with the License. |
| 6 |
|
# You may obtain a copy of the License at |
| 7 |
|
# |
| 8 |
|
# http://www.apache.org/licenses/LICENSE-2.0 |
| 9 |
|
# |
| 10 |
|
# Unless required by applicable law or agreed to in writing, software |
| 11 |
|
# distributed under the License is distributed on an "AS IS" BASIS, |
| 12 |
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or |
| 13 |
|
# implied. |
| 14 |
|
# See the License for the specific language governing permissions and |
| 15 |
|
# limitations under the License. |
| 16 |
|
|
| 17 |
1 |
__author__ = "Alfonso Tierno, alfonso.tiernosepulveda@telefonica.com" |
| 18 |
1 |
__date__ = "2020-06-17" |
| 19 |
|
|
| 20 |
1 |
from copy import deepcopy |
| 21 |
1 |
import unittest |
| 22 |
1 |
from unittest import TestCase |
| 23 |
1 |
from unittest.mock import patch, Mock |
| 24 |
1 |
from osm_nbi.base_topic import ( |
| 25 |
|
BaseTopic, |
| 26 |
|
EngineException, |
| 27 |
|
NBIBadArgumentsException, |
| 28 |
|
detect_descriptor_usage, |
| 29 |
|
update_descriptor_usage_state, |
| 30 |
|
) |
| 31 |
1 |
from osm_common import dbbase |
| 32 |
1 |
from osm_nbi.tests.test_pkg_descriptors import db_vnfds_text, db_nsds_text |
| 33 |
1 |
import yaml |
| 34 |
|
|
| 35 |
1 |
db_vnfd_content = yaml.safe_load(db_vnfds_text)[0] |
| 36 |
1 |
db_nsd_content = yaml.safe_load(db_nsds_text)[0] |
| 37 |
|
|
| 38 |
|
|
| 39 |
1 |
class Test_BaseTopic(TestCase): |
| 40 |
1 |
@classmethod |
| 41 |
1 |
def setUpClass(cls): |
| 42 |
1 |
cls.test_name = "test-base-topic" |
| 43 |
|
|
| 44 |
1 |
@classmethod |
| 45 |
1 |
def tearDownClass(cls): |
| 46 |
1 |
pass |
| 47 |
|
|
| 48 |
1 |
def setUp(self): |
| 49 |
1 |
self.db = Mock(dbbase.DbBase()) |
| 50 |
|
|
| 51 |
1 |
def test_update_input_with_kwargs(self): |
| 52 |
1 |
test_set = ( |
| 53 |
|
# (descriptor content, kwargs, expected descriptor (None=fails), message) |
| 54 |
|
( |
| 55 |
|
{"a": {"none": None}}, |
| 56 |
|
{"a.b.num": "v"}, |
| 57 |
|
{"a": {"none": None, "b": {"num": "v"}}}, |
| 58 |
|
"create dict", |
| 59 |
|
), |
| 60 |
|
( |
| 61 |
|
{"a": {"none": None}}, |
| 62 |
|
{"a.none.num": "v"}, |
| 63 |
|
{"a": {"none": {"num": "v"}}}, |
| 64 |
|
"create dict over none", |
| 65 |
|
), |
| 66 |
|
( |
| 67 |
|
{"a": {"b": {"num": 4}}}, |
| 68 |
|
{"a.b.num": "v"}, |
| 69 |
|
{"a": {"b": {"num": "v"}}}, |
| 70 |
|
"replace_number", |
| 71 |
|
), |
| 72 |
|
( |
| 73 |
|
{"a": {"b": {"num": 4}}}, |
| 74 |
|
{"a.b.num.c.d": "v"}, |
| 75 |
|
{"a": {"b": {"num": {"c": {"d": "v"}}}}}, |
| 76 |
|
"create dict over number", |
| 77 |
|
), |
| 78 |
|
( |
| 79 |
|
{"a": {"b": {"num": 4}}}, |
| 80 |
|
{"a.b": "v"}, |
| 81 |
|
{"a": {"b": "v"}}, |
| 82 |
|
"replace dict with a string", |
| 83 |
|
), |
| 84 |
|
( |
| 85 |
|
{"a": {"b": {"num": 4}}}, |
| 86 |
|
{"a.b": None}, |
| 87 |
|
{"a": {}}, |
| 88 |
|
"replace dict with None", |
| 89 |
|
), |
| 90 |
|
( |
| 91 |
|
{"a": [{"b": {"num": 4}}]}, |
| 92 |
|
{"a.b.num": "v"}, |
| 93 |
|
None, |
| 94 |
|
"create dict over list should fail", |
| 95 |
|
), |
| 96 |
|
( |
| 97 |
|
{"a": [{"b": {"num": 4}}]}, |
| 98 |
|
{"a.0.b.num": "v"}, |
| 99 |
|
{"a": [{"b": {"num": "v"}}]}, |
| 100 |
|
"set list", |
| 101 |
|
), |
| 102 |
|
( |
| 103 |
|
{"a": [{"b": {"num": 4}}]}, |
| 104 |
|
{"a.3.b.num": "v"}, |
| 105 |
|
{"a": [{"b": {"num": 4}}, None, None, {"b": {"num": "v"}}]}, |
| 106 |
|
"expand list", |
| 107 |
|
), |
| 108 |
|
({"a": [[4]]}, {"a.0.0": "v"}, {"a": [["v"]]}, "set nested list"), |
| 109 |
|
( |
| 110 |
|
{"a": [[4]]}, |
| 111 |
|
{"a.0.2": "v"}, |
| 112 |
|
{"a": [[4, None, "v"]]}, |
| 113 |
|
"expand nested list", |
| 114 |
|
), |
| 115 |
|
( |
| 116 |
|
{"a": [[4]]}, |
| 117 |
|
{"a.2.2": "v"}, |
| 118 |
|
{"a": [[4], None, {"2": "v"}]}, |
| 119 |
|
"expand list and add number key", |
| 120 |
|
), |
| 121 |
|
({"a": None}, {"b.c": "v"}, {"a": None, "b": {"c": "v"}}, "expand at root"), |
| 122 |
|
) |
| 123 |
1 |
for desc, kwargs, expected, message in test_set: |
| 124 |
1 |
if expected is None: |
| 125 |
1 |
self.assertRaises( |
| 126 |
|
EngineException, BaseTopic._update_input_with_kwargs, desc, kwargs |
| 127 |
|
) |
| 128 |
|
else: |
| 129 |
1 |
BaseTopic._update_input_with_kwargs(desc, kwargs) |
| 130 |
1 |
self.assertEqual(desc, expected, message) |
| 131 |
|
|
| 132 |
1 |
def test_detect_descriptor_usage_empty_descriptor(self): |
| 133 |
1 |
descriptor = {} |
| 134 |
1 |
db_collection = "vnfds" |
| 135 |
1 |
with self.assertRaises(EngineException) as error: |
| 136 |
1 |
detect_descriptor_usage(descriptor, db_collection, self.db) |
| 137 |
0 |
self.assertIn( |
| 138 |
|
"Argument is mandatory and can not be empty, Bad arguments: descriptor", |
| 139 |
|
error, |
| 140 |
|
"Error message is wrong.", |
| 141 |
|
) |
| 142 |
1 |
self.db.get_list.assert_not_called() |
| 143 |
|
|
| 144 |
1 |
def test_detect_descriptor_usage_empty_db_argument(self): |
| 145 |
1 |
descriptor = deepcopy(db_vnfd_content) |
| 146 |
1 |
db_collection = "vnfds" |
| 147 |
1 |
db = None |
| 148 |
1 |
with self.assertRaises(EngineException) as error: |
| 149 |
1 |
detect_descriptor_usage(descriptor, db_collection, db) |
| 150 |
0 |
self.assertIn( |
| 151 |
|
"A valid DB object should be provided, Bad arguments: db", |
| 152 |
|
error, |
| 153 |
|
"Error message is wrong.", |
| 154 |
|
) |
| 155 |
1 |
self.db.get_list.assert_not_called() |
| 156 |
|
|
| 157 |
1 |
def test_detect_descriptor_usage_which_is_in_use(self): |
| 158 |
1 |
descriptor = deepcopy(db_vnfd_content) |
| 159 |
1 |
db_collection = "vnfds" |
| 160 |
1 |
self.db.get_list.side_effect = [deepcopy(db_vnfd_content)] |
| 161 |
1 |
expected = True |
| 162 |
1 |
result = detect_descriptor_usage(descriptor, db_collection, self.db) |
| 163 |
1 |
self.assertEqual(result, expected, "wrong result") |
| 164 |
1 |
self.db.get_list.assert_called_once_with( |
| 165 |
|
"vnfrs", {"vnfd-id": descriptor["_id"]} |
| 166 |
|
) |
| 167 |
|
|
| 168 |
1 |
def test_detect_descriptor_usage_which_is_not_in_use(self): |
| 169 |
1 |
descriptor = deepcopy(db_nsd_content) |
| 170 |
1 |
self.db.get_list.return_value = [] |
| 171 |
1 |
db_collection = "nsds" |
| 172 |
1 |
expected = None |
| 173 |
1 |
result = detect_descriptor_usage(descriptor, db_collection, self.db) |
| 174 |
1 |
self.assertEqual(result, expected, "wrong result") |
| 175 |
1 |
self.db.get_list.assert_called_once_with("nsrs", {"nsd-id": descriptor["_id"]}) |
| 176 |
|
|
| 177 |
1 |
def test_detect_descriptor_usage_wrong_desc_format(self): |
| 178 |
1 |
descriptor = deepcopy(db_nsd_content) |
| 179 |
1 |
descriptor.pop("_id") |
| 180 |
1 |
db_collection = "nsds" |
| 181 |
1 |
with self.assertRaises(EngineException) as error: |
| 182 |
1 |
detect_descriptor_usage(descriptor, db_collection, self.db) |
| 183 |
0 |
self.assertIn("KeyError", error, "wrong error type") |
| 184 |
1 |
self.db.get_list.assert_not_called() |
| 185 |
|
|
| 186 |
1 |
def test_detect_descriptor_usage_wrong_db_collection(self): |
| 187 |
1 |
descriptor = deepcopy(db_vnfd_content) |
| 188 |
1 |
descriptor.pop("_id") |
| 189 |
1 |
db_collection = "vnf" |
| 190 |
1 |
with self.assertRaises(EngineException) as error: |
| 191 |
1 |
detect_descriptor_usage(descriptor, db_collection, self.db) |
| 192 |
0 |
self.assertIn( |
| 193 |
|
"db_collection should be equal to vnfds or nsds, db_collection", |
| 194 |
|
error, |
| 195 |
|
"wrong error type", |
| 196 |
|
) |
| 197 |
|
|
| 198 |
1 |
self.db.get_list.assert_not_called() |
| 199 |
|
|
| 200 |
1 |
@patch("osm_nbi.base_topic.detect_descriptor_usage") |
| 201 |
1 |
def test_update_descriptor_usage_state_to_in_use(self, mock_descriptor_usage): |
| 202 |
1 |
db_collection = "vnfds" |
| 203 |
1 |
descriptor = deepcopy(db_vnfd_content) |
| 204 |
1 |
mock_descriptor_usage.return_value = True |
| 205 |
1 |
descriptor_update = {"_admin.usageState": "IN_USE"} |
| 206 |
1 |
update_descriptor_usage_state(descriptor, db_collection, self.db) |
| 207 |
1 |
self.db.set_one.assert_called_once_with( |
| 208 |
|
db_collection, {"_id": descriptor["_id"]}, update_dict=descriptor_update |
| 209 |
|
) |
| 210 |
|
|
| 211 |
1 |
@patch("osm_nbi.base_topic.detect_descriptor_usage") |
| 212 |
1 |
def test_update_descriptor_usage_state_to_not_in_use(self, mock_descriptor_usage): |
| 213 |
1 |
db_collection = "nsds" |
| 214 |
1 |
descriptor = deepcopy(db_nsd_content) |
| 215 |
1 |
mock_descriptor_usage.return_value = False |
| 216 |
1 |
descriptor_update = {"_admin.usageState": "NOT_IN_USE"} |
| 217 |
1 |
update_descriptor_usage_state(descriptor, db_collection, self.db) |
| 218 |
1 |
self.db.set_one.assert_called_once_with( |
| 219 |
|
db_collection, {"_id": descriptor["_id"]}, update_dict=descriptor_update |
| 220 |
|
) |
| 221 |
|
|
| 222 |
1 |
@patch("osm_nbi.base_topic.detect_descriptor_usage") |
| 223 |
1 |
def test_update_descriptor_usage_state_db_exception(self, mock_descriptor_usage): |
| 224 |
1 |
db_collection = "nsd" |
| 225 |
1 |
descriptor = deepcopy(db_nsd_content) |
| 226 |
1 |
mock_descriptor_usage.side_effect = NBIBadArgumentsException |
| 227 |
1 |
with self.assertRaises(EngineException) as error: |
| 228 |
1 |
update_descriptor_usage_state(descriptor, db_collection, self.db) |
| 229 |
0 |
self.assertIn( |
| 230 |
|
"db_collection should be equal to vnfds or nsds, db_collection", |
| 231 |
|
error, |
| 232 |
|
"wrong error type", |
| 233 |
|
) |
| 234 |
1 |
self.db.set_one.assert_not_called() |
| 235 |
|
|
| 236 |
|
|
| 237 |
1 |
if __name__ == "__main__": |
| 238 |
0 |
unittest.main() |
![[Jenkins]](/jenkins/static/c06f6c43/images/jenkins-header-logo-v2.svg){kind=logo}
