Coverage for osm_nbi/tests/test_base_topic.py: 94%

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__author__ = "Alfonso Tierno, alfonso.tiernosepulveda@telefonica.com" 

18__date__ = "2020-06-17" 

19 

20from copy import deepcopy 

21import unittest 

22from unittest import TestCase 

23from unittest.mock import patch, Mock 

24from osm_nbi.base_topic import ( 

25 BaseTopic, 

26 EngineException, 

27 NBIBadArgumentsException, 

28 detect_descriptor_usage, 

29 update_descriptor_usage_state, 

30) 

31from osm_common import dbbase 

32from osm_nbi.tests.test_pkg_descriptors import db_vnfds_text, db_nsds_text 

33import yaml 

34 

35db_vnfd_content = yaml.safe_load(db_vnfds_text)[0] 

36db_nsd_content = yaml.safe_load(db_nsds_text)[0] 

37 

38 

39class Test_BaseTopic(TestCase): 

40 @classmethod 

41 def setUpClass(cls): 

42 cls.test_name = "test-base-topic" 

43 

44 @classmethod 

45 def tearDownClass(cls): 

46 pass 

47 

48 def setUp(self): 

49 self.db = Mock(dbbase.DbBase()) 

50 

51 def test_update_input_with_kwargs(self): 

52 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 for desc, kwargs, expected, message in test_set: 

124 if expected is None: 

125 self.assertRaises( 

126 EngineException, BaseTopic._update_input_with_kwargs, desc, kwargs 

127 ) 

128 else: 

129 BaseTopic._update_input_with_kwargs(desc, kwargs) 

130 self.assertEqual(desc, expected, message) 

131 

132 def test_detect_descriptor_usage_empty_descriptor(self): 

133 descriptor = {} 

134 db_collection = "vnfds" 

135 with self.assertRaises(EngineException) as error: 

136 detect_descriptor_usage(descriptor, db_collection, self.db) 

137 self.assertIn( 

138 "Argument is mandatory and can not be empty, Bad arguments: descriptor", 

139 error, 

140 "Error message is wrong.", 

141 ) 

142 self.db.get_list.assert_not_called() 

143 

144 def test_detect_descriptor_usage_empty_db_argument(self): 

145 descriptor = deepcopy(db_vnfd_content) 

146 db_collection = "vnfds" 

147 db = None 

148 with self.assertRaises(EngineException) as error: 

149 detect_descriptor_usage(descriptor, db_collection, db) 

150 self.assertIn( 

151 "A valid DB object should be provided, Bad arguments: db", 

152 error, 

153 "Error message is wrong.", 

154 ) 

155 self.db.get_list.assert_not_called() 

156 

157 def test_detect_descriptor_usage_which_is_in_use(self): 

158 descriptor = deepcopy(db_vnfd_content) 

159 db_collection = "vnfds" 

160 self.db.get_list.side_effect = [deepcopy(db_vnfd_content)] 

161 expected = True 

162 result = detect_descriptor_usage(descriptor, db_collection, self.db) 

163 self.assertEqual(result, expected, "wrong result") 

164 self.db.get_list.assert_called_once_with( 

165 "vnfrs", {"vnfd-id": descriptor["_id"]} 

166 ) 

167 

168 def test_detect_descriptor_usage_which_is_not_in_use(self): 

169 descriptor = deepcopy(db_nsd_content) 

170 self.db.get_list.return_value = [] 

171 db_collection = "nsds" 

172 expected = None 

173 result = detect_descriptor_usage(descriptor, db_collection, self.db) 

174 self.assertEqual(result, expected, "wrong result") 

175 self.db.get_list.assert_called_once_with("nsrs", {"nsd-id": descriptor["_id"]}) 

176 

177 def test_detect_descriptor_usage_wrong_desc_format(self): 

178 descriptor = deepcopy(db_nsd_content) 

179 descriptor.pop("_id") 

180 db_collection = "nsds" 

181 with self.assertRaises(EngineException) as error: 

182 detect_descriptor_usage(descriptor, db_collection, self.db) 

183 self.assertIn("KeyError", error, "wrong error type") 

184 self.db.get_list.assert_not_called() 

185 

186 def test_detect_descriptor_usage_wrong_db_collection(self): 

187 descriptor = deepcopy(db_vnfd_content) 

188 descriptor.pop("_id") 

189 db_collection = "vnf" 

190 with self.assertRaises(EngineException) as error: 

191 detect_descriptor_usage(descriptor, db_collection, self.db) 

192 self.assertIn( 

193 "db_collection should be equal to vnfds or nsds, db_collection", 

194 error, 

195 "wrong error type", 

196 ) 

197 

198 self.db.get_list.assert_not_called() 

199 

200 @patch("osm_nbi.base_topic.detect_descriptor_usage") 

201 def test_update_descriptor_usage_state_to_in_use(self, mock_descriptor_usage): 

202 db_collection = "vnfds" 

203 descriptor = deepcopy(db_vnfd_content) 

204 mock_descriptor_usage.return_value = True 

205 descriptor_update = {"_admin.usageState": "IN_USE"} 

206 update_descriptor_usage_state(descriptor, db_collection, self.db) 

207 self.db.set_one.assert_called_once_with( 

208 db_collection, {"_id": descriptor["_id"]}, update_dict=descriptor_update 

209 ) 

210 

211 @patch("osm_nbi.base_topic.detect_descriptor_usage") 

212 def test_update_descriptor_usage_state_to_not_in_use(self, mock_descriptor_usage): 

213 db_collection = "nsds" 

214 descriptor = deepcopy(db_nsd_content) 

215 mock_descriptor_usage.return_value = False 

216 descriptor_update = {"_admin.usageState": "NOT_IN_USE"} 

217 update_descriptor_usage_state(descriptor, db_collection, self.db) 

218 self.db.set_one.assert_called_once_with( 

219 db_collection, {"_id": descriptor["_id"]}, update_dict=descriptor_update 

220 ) 

221 

222 @patch("osm_nbi.base_topic.detect_descriptor_usage") 

223 def test_update_descriptor_usage_state_db_exception(self, mock_descriptor_usage): 

224 db_collection = "nsd" 

225 descriptor = deepcopy(db_nsd_content) 

226 mock_descriptor_usage.side_effect = NBIBadArgumentsException 

227 with self.assertRaises(EngineException) as error: 

228 update_descriptor_usage_state(descriptor, db_collection, self.db) 

229 self.assertIn( 

230 "db_collection should be equal to vnfds or nsds, db_collection", 

231 error, 

232 "wrong error type", 

233 ) 

234 self.db.set_one.assert_not_called() 

235 

236 

237if __name__ == "__main__": 

238 unittest.main()