Coverage for NG-RO/osm_ng_ro/tests/test_monitor.py: 99%
1421 statements
« prev ^ index » next coverage.py v7.3.2, created at 2024-06-30 09:51 +0000
1#######################################################################################
2# Copyright ETSI Contributors and Others.
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#######################################################################################
17from copy import deepcopy
18import logging
19import threading
20import unittest
21from unittest.mock import MagicMock, mock_open, patch
23from novaclient.v2.servers import Server as NovaServer
24from osm_common import dbmongo
25from osm_common.dbbase import DbException
26from osm_ng_ro.monitor import (
27 MonitorDbException,
28 MonitorVimException,
29 MonitorVms,
30 MonitorVmsException,
31 start_monitoring,
32 stop_monitoring,
33 VimToMonitor,
34 VmToMonitor,
35)
36from osm_ng_ro.tests.sample_data import (
37 config,
38 db_vim_cacert,
39 db_vim_collection,
40 deleted_externally,
41 file_name,
42 interface_info2,
43 interface_with_binding,
44 ip1_addr,
45 mac1_addr,
46 mac2_addr,
47 net1_id,
48 old_interface,
49 old_interface2,
50 plugin_name,
51 port1,
52 port2,
53 ro_task1,
54 ro_task2,
55 sample_vim,
56 sample_vim_info,
57 sample_vnfr,
58 serialized_interface_info,
59 serialized_server_info,
60 server_other_info,
61 target_id,
62 target_record,
63 target_record2,
64 vdur_path,
65 vim1_id,
66 vim2_id,
67 vim3_id,
68 vim4_id,
69 vim_info_path,
70 vims,
71 vims_to_monitor,
72 vm1_id,
73 vm2_id,
74 vnfr_id,
75 wrong_ro_task,
76)
77from osm_ro_plugin.vimconn import VimConnector
78import yaml
81def create_server(id: str, name: str, status: str = "ACTIVE", info: dict = {}):
82 instance = NovaServer(manager="manager", info=info)
83 instance.id = id
84 instance.name = name
85 instance.status = status
86 return instance
89# The preparation for the tests
90sample_vim_connector_instance = VimConnector(
91 uuid=sample_vim["_id"],
92 name=sample_vim["name"],
93 tenant_id=sample_vim.get("vim_tenant_id"),
94 tenant_name=sample_vim.get("vim_tenant_name"),
95 url=sample_vim["vim_url"],
96)
97sample_vm = VmToMonitor(vm1_id, target_record)
98sample_vm2 = VmToMonitor(vm2_id, target_record)
99sample_vm3 = VmToMonitor("deleted-vm-id", target_record)
100server1 = create_server(vm1_id, "server1")
101server2 = create_server(vm2_id, "server2")
102server3 = create_server("other-vm-id3", "other-vm3")
103server4 = create_server("other-vm-id4", "other-vm4")
104all_server_info = deepcopy(server_other_info)
105server7 = create_server(vm1_id, "server7", info=all_server_info)
108class CopyingMock(MagicMock):
109 def __call__(self, *args, **kwargs):
110 args = deepcopy(args)
111 kwargs = deepcopy(kwargs)
112 return super(CopyingMock, self).__call__(*args, **kwargs)
115def check_if_assert_not_called(mocks: list):
116 for mocking in mocks:
117 mocking.assert_not_called()
120class TestMonitorVms(unittest.TestCase):
121 @patch("osm_ng_ro.monitor.MonitorVms.__init__")
122 @patch("osm_ng_ro.ns_thread.ConfigValidate")
123 @patch("osm_ng_ro.monitor.MonitorVms.get_db_vims")
124 @patch("osm_ng_ro.monitor.MonitorVms.load_vims")
125 @patch("logging.getLogger", autospec=True)
126 def setUp(
127 self,
128 mock_logger,
129 mock_load_vims,
130 mock_get_db_vims,
131 mock_config_validate,
132 mock_init,
133 ):
134 # We are disabling the logging of exception not to print them to console.
135 mock_logger = logging.getLogger()
136 mock_logger.disabled = True
137 mock_init.return_value = None
138 self.monitor = MonitorVms(config=config)
139 self.monitor.db_vims = []
140 self.monitor.db = CopyingMock(dbmongo.DbMongo(), autospec=True)
141 self.monitor.config = config
142 self.monitor.logger = mock_logger
143 self.monitor.my_vims = {}
144 self.monitor.refresh_config = mock_config_validate
146 @patch("osm_ng_ro.ns_thread.ConfigValidate.__init__")
147 @patch("osm_ng_ro.monitor.MonitorVms.get_db_vims")
148 @patch("osm_ng_ro.monitor.MonitorVms.load_vims")
149 @patch("logging.getLogger", autospec=True)
150 @patch("osm_ng_ro.monitor.MonitorVms.connect_db")
151 def test_init(
152 self,
153 mock_connect_db,
154 mock_logger,
155 mock_load_vims,
156 mock_get_db_vims,
157 mock_config_validate_init,
158 ):
159 mock_config_validate_init.return_value = None
160 mock_get_db_vims.return_value = vims
161 instance = MonitorVms(config)
162 mock_config_validate_init.assert_called_once_with(config)
163 self.assertDictEqual(instance.config, config)
164 mock_load_vims.assert_called_once()
165 self.assertEqual(instance.db_vims, vims)
166 mock_connect_db.assert_called_once()
167 self.assertIsNone(instance.db)
168 self.assertIsNotNone(instance.db_vims)
169 mock_logger.assert_called_once_with("ro.monitor")
171 @patch("osm_ng_ro.monitor.MonitorVms._load_vim")
172 def test_load_vims_empty_db_vims(self, mock_load_vim):
173 self.monitor.load_vims()
174 mock_load_vim.assert_not_called()
176 @patch("osm_ng_ro.monitor.MonitorVms._load_vim")
177 def test_load_vims_vim_id_not_in_my_vims(self, mock_load_vim):
178 self.monitor.db_vims = vims
179 self.monitor.my_vims = {vim3_id: "vim-obj3"}
180 self.monitor.load_vims()
181 _call_mock_load_vim = mock_load_vim.call_args_list
182 self.assertEqual(mock_load_vim.call_count, 2)
183 self.assertEqual(
184 _call_mock_load_vim[0][0],
185 (vim1_id,),
186 )
187 self.assertEqual(
188 _call_mock_load_vim[1][0],
189 (vim2_id,),
190 )
192 @patch("osm_ng_ro.monitor.MonitorVms._load_vim")
193 def test_load_vims_vim_id_in_my_vims(self, mock_load_vim):
194 self.monitor.db_vims = vims
195 self.monitor.my_vims = {vim1_id: "vim-obj1", vim2_id: "vim-obj2"}
196 self.monitor.load_vims()
197 mock_load_vim.assert_not_called()
199 @patch("osm_common.dbmongo.DbMongo.db_connect")
200 @patch("osm_common.dbmongo.DbMongo.__init__")
201 @patch("osm_common.dbmemory.DbMemory.db_connect")
202 @patch("osm_common.dbmemory.DbMemory.__init__")
203 def test_connect_db_type_mongo(
204 self,
205 mock_dbmemory_init,
206 mock_dbmemory_connect,
207 mock_dbmongo_init,
208 mock_dbmongo_connect,
209 ):
210 self.monitor.db = None
211 self.monitor.config["database"]["driver"] = "mongo"
212 mock_dbmongo_init.return_value = None
213 self.monitor.connect_db()
214 mock_dbmongo_init.assert_called_once()
215 mock_dbmongo_connect.assert_called_once()
216 self.monitor.db.db_connect.assert_called_once_with(
217 self.monitor.config["database"]
218 )
219 check_if_assert_not_called([mock_dbmemory_init, mock_dbmemory_connect])
221 @patch("osm_common.dbmongo.DbMongo.db_connect")
222 @patch("osm_common.dbmongo.DbMongo.__init__")
223 @patch("osm_common.dbmemory.DbMemory.db_connect")
224 @patch("osm_common.dbmemory.DbMemory.__init__")
225 def test_connect_db_type_mongo_initialize_exception(
226 self,
227 mock_dbmemory_init,
228 mock_dbmemory_connect,
229 mock_dbmongo_init,
230 mock_dbmongo_connect,
231 ):
232 self.monitor.db = None
233 self.monitor.config["database"]["driver"] = "mongo"
234 mock_dbmongo_init.side_effect = ValueError("Db object could not be created.")
235 with self.assertRaises(MonitorDbException) as err:
236 self.monitor.connect_db()
237 self.assertEqual(str(err.exception), "Db object could not be created.")
238 mock_dbmongo_init.assert_called_once()
239 check_if_assert_not_called(
240 [mock_dbmongo_connect, mock_dbmemory_init, mock_dbmemory_connect]
241 )
243 @patch("osm_common.dbmongo.DbMongo.db_connect")
244 @patch("osm_common.dbmongo.DbMongo.__init__")
245 @patch("osm_common.dbmemory.DbMemory.db_connect")
246 @patch("osm_common.dbmemory.DbMemory.__init__")
247 def test_connect_db_type_mongo_connection_exception(
248 self,
249 mock_dbmemory_init,
250 mock_dbmemory_connect,
251 mock_dbmongo_init,
252 mock_dbmongo_connect,
253 ):
254 self.monitor.db = None
255 self.monitor.config["database"]["driver"] = "mongo"
256 mock_dbmongo_init.return_value = None
257 mock_dbmongo_connect.side_effect = DbException("Connection failed")
258 with self.assertRaises(MonitorDbException) as err:
259 self.monitor.connect_db()
260 self.assertEqual(str(err.exception), "database exception Connection failed")
261 mock_dbmongo_init.assert_called_once()
262 mock_dbmongo_connect.assert_called_once_with(self.monitor.config["database"])
263 check_if_assert_not_called([mock_dbmemory_init, mock_dbmemory_connect])
265 @patch("osm_common.dbmongo.DbMongo.db_connect")
266 @patch("osm_common.dbmongo.DbMongo.__init__")
267 @patch("osm_common.dbmemory.DbMemory.db_connect")
268 @patch("osm_common.dbmemory.DbMemory.__init__")
269 def test_connect_db_type_memory(
270 self,
271 mock_dbmemory_init,
272 mock_dbmemory_connect,
273 mock_dbmongo_init,
274 mock_dbmongo_connect,
275 ):
276 self.monitor.db = None
277 self.monitor.config["database"]["driver"] = "memory"
278 mock_dbmemory_init.return_value = None
279 self.monitor.connect_db()
280 mock_dbmemory_init.assert_called_once()
281 mock_dbmemory_connect.assert_called_once_with(self.monitor.config["database"])
282 check_if_assert_not_called([mock_dbmongo_init, mock_dbmongo_connect])
284 @patch("osm_common.dbmongo.DbMongo.db_connect")
285 @patch("osm_common.dbmongo.DbMongo.__init__")
286 @patch("osm_common.dbmemory.DbMemory.db_connect")
287 @patch("osm_common.dbmemory.DbMemory.__init__")
288 def test_connect_db_existing_db(
289 self,
290 mock_dbmemory_init,
291 mock_dbmemory_connect,
292 mock_dbmongo_init,
293 mock_dbmongo_connect,
294 ):
295 self.monitor.connect_db()
296 check_if_assert_not_called(
297 [
298 mock_dbmemory_init,
299 mock_dbmongo_init,
300 mock_dbmemory_connect,
301 mock_dbmongo_connect,
302 ]
303 )
305 @patch("osm_common.dbmongo.DbMongo.db_connect")
306 @patch("osm_common.dbmongo.DbMongo.__init__")
307 @patch("osm_common.dbmemory.DbMemory.db_connect")
308 @patch("osm_common.dbmemory.DbMemory.__init__")
309 def test_connect_db_wrong_driver_type(
310 self,
311 mock_dbmemory_init,
312 mock_dbmemory_connect,
313 mock_dbmongo_init,
314 mock_dbmongo_connect,
315 ):
316 self.monitor.db = None
317 self.monitor.config["database"]["driver"] = "posgresql"
318 with self.assertRaises(MonitorDbException) as err:
319 self.monitor.connect_db()
320 self.assertEqual(
321 str(err.exception),
322 "Invalid configuration param 'posgresql' at '[database]':'driver'",
323 )
324 check_if_assert_not_called(
325 [
326 mock_dbmemory_init,
327 mock_dbmongo_init,
328 mock_dbmemory_connect,
329 mock_dbmongo_connect,
330 ]
331 )
333 def test_get_db_vims(self):
334 self.monitor.db.get_list.return_value = vims
335 result = self.monitor.get_db_vims()
336 self.assertEqual(result, vims)
337 self.monitor.db.get_list.assert_called_once_with(
338 "vim_accounts", {"vim_type": "openstack"}
339 )
341 def test_get_db_vims_db_raises(self):
342 self.monitor.db.get_list.side_effect = DbException("Connection failed.")
343 with self.assertRaises(DbException) as err:
344 result = self.monitor.get_db_vims()
345 self.assertEqual(result, None)
346 self.assertEqual(str(err.exception), "database exception Connection failed.")
347 self.monitor.db.get_list.assert_called_once_with(
348 "vim_accounts", {"vim_type": "openstack"}
349 )
351 def test_find_ro_tasks_to_monitor(self):
352 self.monitor.db.get_list.return_value = [ro_task1]
353 result = self.monitor.find_ro_tasks_to_monitor()
354 self.assertEqual(result, [ro_task1])
355 self.monitor.db.get_list.assert_called_once_with(
356 "ro_tasks",
357 q_filter={
358 "tasks.status": ["DONE"],
359 "tasks.item": ["vdu"],
360 },
361 )
363 def test_find_ro_tasks_to_monitor_db_exception(self):
364 self.monitor.db.get_list.side_effect = DbException("Wrong database status")
365 with self.assertRaises(DbException) as err:
366 result = self.monitor.find_ro_tasks_to_monitor()
367 self.assertEqual(result, None)
368 self.assertEqual(str(err.exception), "database exception Wrong database status")
369 self.monitor.db.get_list.assert_called_once_with(
370 "ro_tasks",
371 q_filter={
372 "tasks.status": ["DONE"],
373 "tasks.item": ["vdu"],
374 },
375 )
377 def test_initialize_target_vim(self):
378 vim_module_conn = VimConnector
379 vim_connector_instance = self.monitor._initialize_target_vim(
380 vim_module_conn, sample_vim
381 )
382 self.assertIsInstance(vim_connector_instance, VimConnector)
383 self.assertListEqual(
384 [vim_connector_instance.id, vim_connector_instance.name],
385 [target_id, "openstackETSI1"],
386 )
388 def test_initialize_target_vim_invalid_vim_connector_input(self):
389 vim_module_conn = "openstack_vim_connector"
390 with self.assertRaises(TypeError) as err:
391 self.monitor._initialize_target_vim(vim_module_conn, sample_vim)
392 self.assertEqual(str(err.exception), "'str' object is not callable")
394 def test_initialize_target_vim_missing_vim_keys(self):
395 vim_module_conn = VimConnector
396 sample_vim = {
397 "_id": target_id,
398 "name": "openstackETSI1",
399 "vim_type": "openstack",
400 }
401 with self.assertRaises(KeyError) as err:
402 self.monitor._initialize_target_vim(vim_module_conn, sample_vim)
403 self.assertEqual(str(err.exception.args[0]), "vim_url")
405 def test_initialize_target_vim_invalid_vim_input_type(self):
406 vim_module_conn = VimConnector
407 sample_vim = [target_id, "openstackETSI1"]
408 with self.assertRaises(TypeError) as err:
409 self.monitor._initialize_target_vim(vim_module_conn, sample_vim)
410 self.assertEqual(
411 str(err.exception), "list indices must be integers or slices, not str"
412 )
414 @patch("osm_ng_ro.monitor.MonitorVms._process_vim_config")
415 @patch("osm_ng_ro.monitor.MonitorVms._load_plugin")
416 @patch("osm_ng_ro.monitor.MonitorVms._initialize_target_vim")
417 def test_load_vim(self, mock_target_vim, mock_load_plugin, mock_vim_config):
418 self.monitor.my_vims = {}
419 sample_vim["schema_version"] = "1.11"
420 self.monitor.db.get_one.return_value = sample_vim
421 mock_load_plugin.return_value = VimConnector
422 mock_target_vim.return_value = sample_vim_connector_instance
423 self.monitor._load_vim(target_id)
424 self.assertEqual(self.monitor.my_vims[target_id], sample_vim_connector_instance)
425 mock_vim_config.assert_called_once()
426 self.monitor.db.get_one.assert_called_once_with(
427 db_vim_collection, {"_id": target_id}
428 )
429 self.monitor.db.encrypt_decrypt_fields.assert_called_once_with(
430 sample_vim,
431 "decrypt",
432 fields=("password", "secret"),
433 schema_version="1.11",
434 salt=target_id,
435 )
436 mock_vim_config.assert_called_once_with(target_id, sample_vim)
437 mock_load_plugin.assert_called_once_with(plugin_name)
438 mock_target_vim.assert_called_once_with(VimConnector, sample_vim)
440 @patch("osm_ng_ro.monitor.MonitorVms._process_vim_config")
441 @patch("osm_ng_ro.monitor.MonitorVms._load_plugin")
442 @patch("osm_ng_ro.monitor.MonitorVms._initialize_target_vim")
443 def test_load_vim_target_vim_not_found(
444 self, mock_target_vim, mock_load_plugin, mock_vim_config
445 ):
446 self.monitor.my_vims = {}
447 self.monitor.db.get_one.return_value = None
448 with self.assertRaises(MonitorVimException) as err:
449 self.monitor._load_vim(target_id)
450 self.assertEqual(
451 str(err.exception),
452 "Cannot load 55b2219a-7bb9-4644-9612-980dada84e83 plugin=rovim_openstack: "
453 "'NoneType' object has no attribute 'get'",
454 )
455 self.monitor.db.get_one.assert_called_once_with(
456 db_vim_collection, {"_id": target_id}
457 )
458 check_if_assert_not_called(
459 [
460 self.monitor.db.encrypt_decrypt_fields,
461 mock_vim_config,
462 mock_load_plugin,
463 mock_target_vim,
464 ]
465 )
467 @patch("osm_ng_ro.monitor.MonitorVms._process_vim_config")
468 @patch("osm_ng_ro.monitor.MonitorVms._load_plugin")
469 @patch("osm_ng_ro.monitor.MonitorVms._initialize_target_vim")
470 def test_load_vim_decrypt_fields_raises(
471 self, mock_target_vim, mock_load_plugin, mock_vim_config
472 ):
473 self.monitor.my_vims = {}
474 sample_vim["schema_version"] = "1.11"
475 self.monitor.db.get_one.return_value = sample_vim
476 self.monitor.db.encrypt_decrypt_fields.side_effect = DbException(
477 "Value could not decrypted."
478 )
479 with self.assertRaises(MonitorVimException) as err:
480 self.monitor._load_vim(target_id)
481 self.assertEqual(
482 str(err.exception),
483 "Cannot load 55b2219a-7bb9-4644-9612-980dada84e83 plugin=rovim_openstack: "
484 "database exception Value could not decrypted.",
485 )
486 self.monitor.db.get_one.assert_called_once_with(
487 db_vim_collection, {"_id": target_id}
488 )
489 self.monitor.db.encrypt_decrypt_fields.assert_called_once_with(
490 sample_vim,
491 "decrypt",
492 fields=("password", "secret"),
493 schema_version="1.11",
494 salt=target_id,
495 )
496 check_if_assert_not_called([mock_vim_config, mock_load_plugin, mock_target_vim])
498 @patch("osm_ng_ro.monitor.MonitorVms._process_vim_config")
499 @patch("osm_ng_ro.monitor.MonitorVms._load_plugin")
500 @patch("osm_ng_ro.monitor.MonitorVms._initialize_target_vim")
501 def test_load_vim_process_vim_config_raises(
502 self, mock_target_vim, mock_load_plugin, mock_vim_config
503 ):
504 self.monitor.my_vims = {}
505 sample_vim["schema_version"] = "1.11"
506 self.monitor.db.get_one.return_value = sample_vim
507 mock_vim_config.side_effect = MonitorVimException(
508 "Error writing file config_1234"
509 )
510 with self.assertRaises(MonitorVimException) as err:
511 self.monitor._load_vim(target_id)
512 self.assertEqual(
513 str(err.exception),
514 "Cannot load 55b2219a-7bb9-4644-9612-980dada84e83 plugin=rovim_openstack: "
515 "Error writing file config_1234",
516 )
517 self.monitor.db.get_one.assert_called_once_with(
518 db_vim_collection, {"_id": target_id}
519 )
520 self.monitor.db.encrypt_decrypt_fields.assert_called_once_with(
521 sample_vim,
522 "decrypt",
523 fields=("password", "secret"),
524 schema_version="1.11",
525 salt=target_id,
526 )
527 mock_vim_config.assert_called_once_with(target_id, sample_vim)
528 check_if_assert_not_called([mock_load_plugin, mock_target_vim])
530 @patch("osm_ng_ro.monitor.MonitorVms._process_vim_config")
531 @patch("osm_ng_ro.monitor.MonitorVms._load_plugin")
532 @patch("osm_ng_ro.monitor.MonitorVms._initialize_target_vim")
533 def test_load_vim_load_plugin_raises(
534 self, mock_target_vim, mock_load_plugin, mock_vim_config
535 ):
536 self.monitor.my_vims = {}
537 sample_vim["schema_version"] = "1.11"
538 self.monitor.db.get_one.return_value = sample_vim
539 mock_load_plugin.side_effect = MonitorVimException(
540 "Cannot load plugin osm_rovim_openstack"
541 )
542 with self.assertRaises(MonitorVimException) as err:
543 self.monitor._load_vim(target_id)
544 self.assertEqual(
545 str(err.exception),
546 "Cannot load 55b2219a-7bb9-4644-9612-980dada84e83 plugin=rovim_openstack: "
547 "Cannot load plugin osm_rovim_openstack",
548 )
549 self.monitor.db.get_one.assert_called_once_with(
550 db_vim_collection, {"_id": target_id}
551 )
552 self.monitor.db.encrypt_decrypt_fields.assert_called_once_with(
553 sample_vim,
554 "decrypt",
555 fields=("password", "secret"),
556 schema_version="1.11",
557 salt=target_id,
558 )
559 mock_vim_config.assert_called_once_with(target_id, sample_vim)
560 mock_load_plugin.assert_called_once_with(plugin_name)
561 mock_target_vim.assert_not_called()
563 @patch("osm_ng_ro.monitor.MonitorVms._process_vim_config")
564 @patch("osm_ng_ro.monitor.MonitorVms._load_plugin")
565 @patch("osm_ng_ro.monitor.MonitorVms._initialize_target_vim")
566 def test_load_vim_initialize_target_vim_raises(
567 self, mock_target_vim, mock_load_plugin, mock_vim_config
568 ):
569 self.monitor.my_vims = {}
570 self.monitor.db.get_one.return_value = sample_vim
571 sample_vim["schema_version"] = "1.0"
572 mock_load_plugin.return_value = VimConnector
573 mock_target_vim.side_effect = TypeError("'module' object is not callable")
574 with self.assertRaises(MonitorVimException) as err:
575 self.monitor._load_vim(target_id)
576 self.assertEqual(
577 str(err.exception),
578 "Cannot load 55b2219a-7bb9-4644-9612-980dada84e83 plugin=rovim_openstack: "
579 "'module' object is not callable",
580 )
581 self.monitor.db.get_one.assert_called_once_with(
582 db_vim_collection, {"_id": target_id}
583 )
584 self.monitor.db.encrypt_decrypt_fields.assert_called_once_with(
585 sample_vim,
586 "decrypt",
587 fields=("password", "secret"),
588 schema_version="1.0",
589 salt=target_id,
590 )
591 mock_vim_config.assert_called_once_with(target_id, sample_vim)
592 mock_load_plugin.assert_called_once_with(plugin_name)
593 mock_target_vim.assert_called_once_with(VimConnector, sample_vim)
595 @patch("osm_ng_ro.monitor.makedirs")
596 @patch("osm_ng_ro.monitor.path")
597 @patch("builtins.open", new_callable=mock_open())
598 def test_process_vim_config_vim_without_config(
599 self, mock_open, mock_path, mock_makedirs
600 ):
601 db_vim = {}
602 self.monitor._process_vim_config(target_id, db_vim)
603 check_if_assert_not_called([mock_open, mock_path.isdir, mock_makedirs])
605 @patch("osm_ng_ro.monitor.random")
606 @patch("osm_ng_ro.monitor.makedirs")
607 @patch("osm_ng_ro.monitor.path")
608 @patch("builtins.open", new_callable=mock_open())
609 def test_process_vim_config_vim_with_ca_cert(
610 self, mock_open, mock_path, mock_makedirs, mock_random
611 ):
612 db_vim = {"config": {"ca_cert_content": "my_vim_cert"}}
613 mock_path.isdir.return_value = False
614 mock_random.randint.return_value = 23242
615 self.monitor._process_vim_config(target_id, db_vim)
616 self.assertEqual(db_vim["config"].get("ca_cert_content"), None)
617 self.assertEqual(
618 db_vim["config"].get("ca_cert"),
619 db_vim_cacert,
620 )
621 mock_path.isdir.asssert_called_once_with(file_name)
622 mock_makedirs.assert_called_once_with(file_name)
623 mock_random.randint.assert_called_once()
624 mock_open.assert_called_once_with(file_name + "/ca_cert", "w")
626 @patch("osm_ng_ro.monitor.random")
627 @patch("osm_ng_ro.monitor.makedirs")
628 @patch("osm_ng_ro.monitor.path")
629 @patch("builtins.open", new_callable=mock_open())
630 def test_process_vim_config_vim_with_cacert_path_is_dir(
631 self, mock_open, mock_path, mock_makedirs, mock_random
632 ):
633 db_vim = {"config": {"ca_cert_content": "my_vim_cert"}}
634 mock_path.isdir.return_value = True
635 mock_random.randint.return_value = 23242
636 self.monitor._process_vim_config(target_id, db_vim)
637 self.assertEqual(db_vim["config"].get("ca_cert_content"), None)
638 self.assertEqual(
639 db_vim["config"].get("ca_cert"),
640 db_vim_cacert,
641 )
642 mock_path.isdir.asssert_called_once_with(file_name)
643 mock_makedirs.assert_not_called()
644 mock_random.randint.assert_called_once()
645 mock_open.assert_called_once_with(file_name + "/ca_cert", "w")
647 @patch("osm_ng_ro.monitor.random")
648 @patch("osm_ng_ro.monitor.makedirs")
649 @patch("osm_ng_ro.monitor.path")
650 @patch("builtins.open", new_callable=mock_open())
651 def test_process_vim_config_vim_with_cacert_makedir_raises(
652 self, mock_open, mock_path, mock_makedirs, mock_random
653 ):
654 db_vim = {"config": {"ca_cert_content": "my_vim_cert"}}
655 mock_path.isdir.return_value = False
656 mock_random.randint.return_value = 23242
657 mock_makedirs.side_effect = OSError("Can not create directory")
658 with self.assertRaises(MonitorVimException) as err:
659 self.monitor._process_vim_config(target_id, db_vim)
660 self.assertEqual(
661 str(err.exception),
662 "Error writing to file '/app/osm_ro/certs/55b2219a-7bb9-4644-9612-980dada84e83:23242': "
663 "Can not create directory",
664 )
665 self.assertEqual(db_vim["config"].get("ca_cert_content"), "my_vim_cert")
666 self.assertEqual(db_vim["config"].get("ca_cert"), None)
667 mock_path.isdir.asssert_called_once_with(file_name)
668 mock_makedirs.assert_called_once_with(file_name)
669 mock_random.randint.assert_called_once()
670 mock_open.assert_not_called()
672 @patch("osm_ng_ro.monitor.random")
673 @patch("osm_ng_ro.monitor.makedirs")
674 @patch("osm_ng_ro.monitor.path")
675 @patch("builtins.open", new_callable=mock_open())
676 def test_process_vim_config_vim_with_cacert_mock_open_raises(
677 self, mock_open, mock_path, mock_makedirs, mock_random
678 ):
679 db_vim = {"config": {"ca_cert_content": "my_vim_cert"}}
680 mock_path.isdir.return_value = False
681 mock_random.randint.return_value = 23242
682 mock_open.side_effect = FileNotFoundError("File is not found.")
683 with self.assertRaises(MonitorVimException) as err:
684 self.monitor._process_vim_config(target_id, db_vim)
685 self.assertEqual(
686 str(err.exception),
687 "Error writing to file '/app/osm_ro/certs/55b2219a-7bb9-4644-9612-980dada84e83:23242/ca_cert': "
688 "File is not found.",
689 )
690 self.assertEqual(db_vim["config"].get("ca_cert_content"), "my_vim_cert")
691 self.assertEqual(db_vim["config"].get("ca_cert"), None)
692 mock_path.isdir.asssert_called_once_with(file_name)
693 mock_makedirs.assert_called_once_with(file_name)
694 mock_random.randint.assert_called_once()
695 mock_open.assert_called_once_with(file_name + "/ca_cert", "w")
697 @patch("osm_ng_ro.monitor.random")
698 @patch("osm_ng_ro.monitor.makedirs")
699 @patch("osm_ng_ro.monitor.path")
700 @patch("builtins.open", new_callable=mock_open())
701 def test_process_vim_config_vim_without_cacert(
702 self, mock_open, mock_path, mock_makedirs, mock_random
703 ):
704 db_vim = {"config": {}}
705 self.monitor._process_vim_config(target_id, db_vim)
706 self.assertEqual(db_vim["config"].get("ca_cert"), None)
707 check_if_assert_not_called(
708 [mock_path.isdir, mock_makedirs, mock_random.randint, mock_open]
709 )
711 @patch("osm_ng_ro.monitor.entry_points")
712 def test_load_plugin_name_exists(self, mock_entry_points):
713 self.monitor.plugins = {plugin_name: VimConnector}
714 result = self.monitor._load_plugin(plugin_name)
715 mock_entry_points.assert_not_called()
716 self.assertEqual(self.monitor.plugins, {plugin_name: VimConnector})
717 self.assertEqual(result, VimConnector)
719 @patch("osm_ng_ro.monitor.entry_points")
720 def test_load_plugin_name_does_not_exist(self, mock_entry_points):
721 self.monitor.plugins = {}
722 mock_ep = MagicMock()
723 mock_ep.load.return_value = sample_vim_connector_instance
724 mock_entry_points.return_value = [mock_ep]
725 result = self.monitor._load_plugin(plugin_name)
726 self.assertEqual(mock_entry_points.call_count, 1)
727 mock_entry_points.assert_called_once_with(
728 group="osm_rovim.plugins", name=plugin_name
729 )
730 self.assertEqual(
731 self.monitor.plugins, {plugin_name: sample_vim_connector_instance}
732 )
733 self.assertEqual(result, sample_vim_connector_instance)
735 @patch("osm_ng_ro.monitor.entry_points")
736 def test_load_plugin_load_raises(self, mock_entry_points):
737 self.monitor.plugins = {}
738 mock_entry_points.return_value = None
739 with self.assertRaises(MonitorVimException) as err:
740 self.monitor._load_plugin(plugin_name)
741 self.assertEqual(
742 str(err.exception),
743 "Cannot load plugin osm_rovim_openstack: 'NoneType' object is not iterable",
744 )
745 self.assertEqual(mock_entry_points.call_count, 1)
746 mock_entry_points.assert_called_once_with(
747 group="osm_rovim.plugins", name=plugin_name
748 )
749 self.assertEqual(self.monitor.plugins, {})
751 @patch("osm_ng_ro.monitor.VmToMonitor")
752 def test_create_vm_to_monitor_empty_ro_task(self, mock_vm_to_monitor):
753 ro_task = {}
754 result = self.monitor.create_vm_to_monitor(ro_task)
755 self.assertEqual(result, None)
756 mock_vm_to_monitor.assert_not_called()
758 @patch("osm_ng_ro.monitor.VmToMonitor")
759 def test_create_vm_to_monitor(self, mock_vm_to_monitor):
760 sample_vm = VmToMonitor("sample_id", "sample_target_record")
761 mock_vm_to_monitor.return_value = sample_vm
762 result = self.monitor.create_vm_to_monitor(ro_task1)
763 self.assertEqual(result, sample_vm)
764 mock_vm_to_monitor.assert_called_once_with(
765 "ebd39f37-e607-4bce-9f10-ea4c5635f726", target_record
766 )
768 @patch("osm_ng_ro.monitor.VmToMonitor")
769 def test_create_vm_to_monitor_wrong_ro_task_format(self, mock_vm_to_monitor):
770 mock_vm_to_monitor.return_value = "VmtoMonitor"
771 with self.assertRaises(KeyError) as err:
772 self.monitor.create_vm_to_monitor(wrong_ro_task)
773 self.assertEqual(str(err.exception.args[0]), "vim_info")
774 mock_vm_to_monitor.assert_not_called()
776 @patch("osm_ng_ro.monitor.MonitorVms.create_vm_to_monitor")
777 def test_add_vm_to_existing_vim(self, mock_create_vm_to_monitor):
778 sample_vim1 = VimToMonitor(vim1_id, [vm1_id])
779 vims_to_monitor = [sample_vim1]
780 result = self.monitor.add_vm_to_existing_vim(vims_to_monitor, ro_task2, vim1_id)
781 self.assertEqual(result, True)
782 mock_create_vm_to_monitor.assert_called_once_with(ro_task2)
783 self.assertEqual(2, len(sample_vim1.vms))
785 @patch("osm_ng_ro.monitor.MonitorVms.create_vm_to_monitor")
786 def test_add_vm_to_existing_vim_empty_vims_list(self, mock_create_vm_to_monitor):
787 vims_to_monitor = []
788 result = self.monitor.add_vm_to_existing_vim(vims_to_monitor, ro_task1, vim1_id)
789 self.assertEqual(result, False)
790 mock_create_vm_to_monitor.assert_not_called()
792 @patch("osm_ng_ro.monitor.MonitorVms.create_vm_to_monitor")
793 def test_add_vm_to_existing_vim_different_target_vim_id(
794 self, mock_create_vm_to_monitor
795 ):
796 sample_vim1 = VimToMonitor(vim1_id, [vm1_id])
797 vims_to_monitor = [sample_vim1]
798 result = self.monitor.add_vm_to_existing_vim(vims_to_monitor, ro_task2, vim2_id)
799 self.assertEqual(result, False)
800 mock_create_vm_to_monitor.assert_not_called()
801 self.assertEqual(1, len(sample_vim1.vms))
803 @patch("osm_ng_ro.monitor.MonitorVms.create_vm_to_monitor")
804 def test_add_vm_to_existing_vim_create_vm_to_monitor_raises(
805 self, mock_create_vm_to_monitor
806 ):
807 sample_vim1 = VimToMonitor(vim1_id, [vm1_id])
808 vims_to_monitor = [sample_vim1]
809 mock_create_vm_to_monitor.side_effect = KeyError(
810 "target_record does not exist."
811 )
812 with self.assertRaises(KeyError) as err:
813 self.monitor.add_vm_to_existing_vim(vims_to_monitor, ro_task2, vim1_id)
814 self.assertEqual(str(err.exception.args[0]), "target_record does not exist.")
815 mock_create_vm_to_monitor.assert_called_once_with(ro_task2)
816 self.assertEqual(1, len(sample_vim1.vms))
818 @patch("osm_ng_ro.monitor.MonitorVms.create_vm_to_monitor")
819 @patch("osm_ng_ro.monitor.VimToMonitor")
820 def test_add_new_vim_for_monitoring(
821 self, mock_vim_to_monitor, mock_create_vm_to_monitor
822 ):
823 sample_vim = VimToMonitor(vim1_id, [])
824 mock_vim_to_monitor.return_value = sample_vim
825 self.monitor.add_new_vim_for_monitoring(vims_to_monitor, ro_task1, vim1_id)
826 mock_vim_to_monitor.assert_called_once_with(vim1_id, [])
827 mock_create_vm_to_monitor.assert_called_once_with(ro_task1)
828 self.assertEqual(len(sample_vim.vms), 1)
829 self.assertEqual(len(vims_to_monitor), 1)
831 @patch("osm_ng_ro.monitor.MonitorVms.create_vm_to_monitor")
832 @patch("osm_ng_ro.monitor.VimToMonitor")
833 def test_add_new_vim_for_monitoring_vim_to_monitor_raises(
834 self, mock_vim_to_monitor, mock_create_vm_to_monitor
835 ):
836 vims_to_monitor = []
837 mock_vim_to_monitor.side_effect = TypeError(
838 "Missing required positional arguments"
839 )
840 with self.assertRaises(TypeError) as err:
841 self.monitor.add_new_vim_for_monitoring(vims_to_monitor, ro_task1, None)
842 self.assertEqual(
843 str(err.exception.args[0]), "Missing required positional arguments"
844 )
845 mock_vim_to_monitor.assert_called_once_with(None, [])
846 mock_create_vm_to_monitor.assert_not_called()
847 self.assertEqual(len(vims_to_monitor), 0)
849 @patch("osm_ng_ro.monitor.MonitorVms.create_vm_to_monitor")
850 @patch("osm_ng_ro.monitor.VimToMonitor")
851 def test_add_new_vim_for_monitoring_create_vm_to_monitor_raises(
852 self, mock_vim_to_monitor, mock_create_vm_to_monitor
853 ):
854 vims_to_monitor = []
855 mock_create_vm_to_monitor.side_effect = KeyError("target_record is not found.")
856 with self.assertRaises(KeyError) as err:
857 self.monitor.add_new_vim_for_monitoring(vims_to_monitor, ro_task1, vim1_id)
858 self.assertEqual(str(err.exception.args[0]), "target_record is not found.")
859 mock_vim_to_monitor.assert_called_once_with(vim1_id, [])
860 mock_create_vm_to_monitor.assert_called_once_with(ro_task1)
861 self.assertEqual(len(vims_to_monitor), 0)
863 @patch("osm_ng_ro.monitor.MonitorVms.add_vm_to_existing_vim")
864 @patch("osm_ng_ro.monitor.MonitorVms.add_new_vim_for_monitoring")
865 def test_prepare_vims_to_monitor_no_proper_existing_vim(
866 self, mock_add_new_vim_for_monitoring, mock_add_vm_to_existing_vim
867 ):
868 mock_add_vm_to_existing_vim.return_value = False
869 self.monitor.prepare_vims_to_monitor(vims_to_monitor, ro_task1, vim1_id)
870 mock_add_vm_to_existing_vim.assert_called_once_with(
871 vims_to_monitor, ro_task1, vim1_id
872 )
873 mock_add_new_vim_for_monitoring.assert_called_once_with(
874 vims_to_monitor, ro_task1, vim1_id
875 )
877 @patch("osm_ng_ro.monitor.MonitorVms.add_vm_to_existing_vim")
878 @patch("osm_ng_ro.monitor.MonitorVms.add_new_vim_for_monitoring")
879 def test_prepare_vims_to_monitor_proper_existing_vim(
880 self, mock_add_new_vim_for_monitoring, mock_add_vm_to_existing_vim
881 ):
882 mock_add_vm_to_existing_vim.return_value = True
883 self.monitor.prepare_vims_to_monitor(vims_to_monitor, ro_task1, vim1_id)
884 mock_add_vm_to_existing_vim.assert_called_once_with(
885 vims_to_monitor, ro_task1, vim1_id
886 )
887 mock_add_new_vim_for_monitoring.assert_not_called()
889 @patch("osm_ng_ro.monitor.MonitorVms.add_vm_to_existing_vim")
890 @patch("osm_ng_ro.monitor.MonitorVms.add_new_vim_for_monitoring")
891 def test_prepare_vims_to_monitor_add_vm_to_existing_vim_raises(
892 self, mock_add_new_vim_for_monitoring, mock_add_vm_to_existing_vim
893 ):
894 mock_add_vm_to_existing_vim.side_effect = KeyError(
895 "target_record is not found."
896 )
897 with self.assertRaises(KeyError) as err:
898 self.monitor.prepare_vims_to_monitor(vims_to_monitor, ro_task1, vim1_id)
899 self.assertEqual(str(err.exception.args[0]), "target_record is not found.")
900 mock_add_vm_to_existing_vim.assert_called_once_with(
901 vims_to_monitor, ro_task1, vim1_id
902 )
903 mock_add_new_vim_for_monitoring.assert_not_called()
905 @patch("osm_ng_ro.monitor.MonitorVms.add_vm_to_existing_vim")
906 @patch("osm_ng_ro.monitor.MonitorVms.add_new_vim_for_monitoring")
907 def test_prepare_vims_to_monitor_add_new_vim_for_monitoring_raises(
908 self, mock_add_new_vim_for_monitoring, mock_add_vm_to_existing_vim
909 ):
910 mock_add_vm_to_existing_vim.return_value = False
911 mock_add_new_vim_for_monitoring.side_effect = KeyError(
912 "target_record is not found."
913 )
914 with self.assertRaises(KeyError) as err:
915 self.monitor.prepare_vims_to_monitor(vims_to_monitor, ro_task1, vim1_id)
916 self.assertEqual(str(err.exception.args[0]), "target_record is not found.")
917 mock_add_vm_to_existing_vim.assert_called_once_with(
918 vims_to_monitor, ro_task1, vim1_id
919 )
920 mock_add_new_vim_for_monitoring.assert_called_once_with(
921 vims_to_monitor, ro_task1, vim1_id
922 )
924 def test_get_db_paths(self):
925 self.monitor.db.get_one.return_value = sample_vnfr
926 (
927 vim_info_path,
928 vim_id,
929 vnfr_id,
930 vdur_path,
931 vdur_index,
932 db_vnfr,
933 ) = self.monitor._get_db_paths(target_record)
934 self.assertEqual(
935 vim_info_path, "vdur.0.vim_info.vim:f239ed93-756b-408e-89f8-fcbf47a9d8f7"
936 )
937 self.assertEqual(vim_id, vim4_id)
938 self.assertEqual(vdur_path, "vdur.0")
939 self.assertEqual(vdur_index, 0)
940 self.assertEqual(vnfr_id, vnfr_id)
941 self.assertDictEqual(db_vnfr, sample_vnfr)
942 self.monitor.db.get_one.assert_called_once_with(
943 "vnfrs",
944 {"_id": vnfr_id},
945 fail_on_empty=False,
946 )
948 def test_get_db_paths_empty_vnfr(self):
949 self.monitor.db.get_one.return_value = None
950 (
951 vim_info_path,
952 vim_id,
953 vnfr_id,
954 vdur_path,
955 vdur_index,
956 db_vnfr,
957 ) = self.monitor._get_db_paths(target_record)
958 self.assertEqual(
959 vim_info_path, "vdur.0.vim_info.vim:f239ed93-756b-408e-89f8-fcbf47a9d8f7"
960 )
961 self.assertEqual(vim_id, vim4_id)
962 self.assertEqual(vdur_path, "vdur.0")
963 self.assertEqual(vdur_index, 0)
964 self.assertEqual(vnfr_id, vnfr_id)
965 self.assertEqual(db_vnfr, None)
966 self.monitor.db.get_one.assert_called_once_with(
967 "vnfrs",
968 {"_id": vnfr_id},
969 fail_on_empty=False,
970 )
972 def test_get_db_paths_invalid_target_record(self):
973 invalid_target_record = "vnfrs:35c034cc-8c5b-48c4-bfa2-17a71577ef19:f239ed93-756b-408e-89f8-fcbf47a9d8f7"
974 with self.assertRaises(MonitorVmsException) as err:
975 self.monitor._get_db_paths(invalid_target_record)
976 self.assertEqual(
977 str(err.exception.args[0]),
978 "not enough values to unpack (expected 4, got 3)",
979 )
980 self.monitor.db.get_one.assert_not_called()
982 def test_get_db_paths_db_raises(self):
983 self.monitor.db.get_one.side_effect = DbException("Connection Failed.")
984 with self.assertRaises(MonitorVmsException) as err:
985 self.monitor._get_db_paths(target_record)
986 self.assertEqual(
987 str(err.exception.args[0]), "database exception Connection Failed."
988 )
989 self.monitor.db.get_one.assert_called_once_with(
990 "vnfrs",
991 {"_id": vnfr_id},
992 fail_on_empty=False,
993 )
995 def test_check_if_vdur_vim_info_exists(self):
996 vdur_index = 0
997 result = self.monitor._check_if_vdur_vim_info_exists(sample_vnfr, vdur_index)
998 self.assertEqual(result, True)
1000 def test_check_if_vdur_vim_info_exists_wrong_vdur_index(self):
1001 vdur_index = 3
1002 result = self.monitor._check_if_vdur_vim_info_exists(sample_vnfr, vdur_index)
1003 self.assertEqual(result, None)
1005 def test_check_if_vdur_vim_info_exists_empty_vnfr(self):
1006 vdur_index = 2
1007 result = self.monitor._check_if_vdur_vim_info_exists({}, vdur_index)
1008 self.assertEqual(result, None)
1010 def test_check_if_vdur_vim_info_exists_str_vdur_index(self):
1011 vdur_index = "2"
1012 result = self.monitor._check_if_vdur_vim_info_exists({}, vdur_index)
1013 self.assertEqual(result, None)
1015 def test_check_if_vdur_vim_info_exists_none_vnfr(self):
1016 vdur_index = 2
1017 result = self.monitor._check_if_vdur_vim_info_exists(None, vdur_index)
1018 self.assertEqual(result, None)
1020 @patch("osm_ng_ro.monitor.MonitorVms._get_db_paths")
1021 @patch("osm_ng_ro.monitor.MonitorVms._check_if_vdur_vim_info_exists")
1022 @patch("osm_ng_ro.monitor.deepcopy")
1023 def test_get_vm_data_from_db(
1024 self, mock_deepcopy, mock_vim_info_exists, mock_get_db_paths
1025 ):
1026 vim_id = vim4_id
1027 vdur_index = 0
1028 db_vnfr = sample_vnfr
1029 mock_get_db_paths.return_value = (
1030 vim_info_path,
1031 vim_id,
1032 vnfr_id,
1033 vdur_path,
1034 vdur_index,
1035 db_vnfr,
1036 )
1037 mock_vim_info_exists.return_value = True
1038 mock_deepcopy.return_value = sample_vim_info
1039 (
1040 vdur_path_result,
1041 vdur_vim_info_update_result,
1042 db_vnfr_result,
1043 existing_vim_info_result,
1044 vnfr_id_result,
1045 vim_info_path_result,
1046 ) = self.monitor._get_vm_data_from_db(sample_vm)
1047 self.assertEqual(vdur_path_result, vdur_path)
1048 self.assertEqual(vdur_vim_info_update_result, sample_vim_info)
1049 self.assertEqual(db_vnfr_result, db_vnfr)
1050 self.assertEqual(existing_vim_info_result, sample_vim_info)
1051 self.assertEqual(vnfr_id_result, vnfr_id)
1052 self.assertEqual(vim_info_path_result, vim_info_path)
1053 mock_deepcopy.assert_called_once_with(sample_vim_info)
1054 mock_get_db_paths.assert_called_once_with(target_record)
1055 mock_vim_info_exists.assert_called_once_with(db_vnfr, vdur_index)
1057 @patch("osm_ng_ro.monitor.MonitorVms._get_db_paths")
1058 @patch("osm_ng_ro.monitor.MonitorVms._check_if_vdur_vim_info_exists")
1059 @patch("osm_ng_ro.monitor.deepcopy")
1060 def test_get_vm_data_from_db_no_vim_info(
1061 self, mock_deepcopy, mock_vim_info_exists, mock_get_db_paths
1062 ):
1063 vim_id = vim4_id
1064 vdur_index = 0
1065 db_vnfr = sample_vnfr
1066 mock_get_db_paths.return_value = (
1067 vim_info_path,
1068 vim_id,
1069 vnfr_id,
1070 vdur_path,
1071 vdur_index,
1072 db_vnfr,
1073 )
1074 mock_vim_info_exists.return_value = False
1075 result = self.monitor._get_vm_data_from_db(sample_vm)
1076 self.assertEqual(result, None)
1077 mock_deepcopy.assert_not_called()
1078 mock_get_db_paths.assert_called_once_with(target_record)
1079 mock_vim_info_exists.assert_called_once_with(db_vnfr, vdur_index)
1081 @patch("osm_ng_ro.monitor.MonitorVms._get_db_paths")
1082 @patch("osm_ng_ro.monitor.MonitorVms._check_if_vdur_vim_info_exists")
1083 @patch("osm_ng_ro.monitor.deepcopy")
1084 def test_get_vm_data_from_db_get_db_path_raises(
1085 self, mock_deepcopy, mock_vim_info_exists, mock_get_db_paths
1086 ):
1087 mock_get_db_paths.side_effect = DbException("Connection failed.")
1088 with self.assertRaises(DbException) as err:
1089 self.monitor._get_vm_data_from_db(sample_vm)
1090 self.assertEqual(
1091 str(err.exception.args[0]), "database exception Connection failed."
1092 )
1093 mock_get_db_paths.assert_called_once_with(target_record)
1094 check_if_assert_not_called([mock_deepcopy, mock_vim_info_exists])
1096 @patch("osm_ng_ro.monitor.MonitorVms._get_db_paths")
1097 @patch("osm_ng_ro.monitor.MonitorVms._check_if_vdur_vim_info_exists")
1098 @patch("osm_ng_ro.monitor.deepcopy")
1099 def test_get_vm_data_from_db_vnfr_without_correct_vdur_index(
1100 self, mock_deepcopy, mock_vim_info_exists, mock_get_db_paths
1101 ):
1102 vim_id = vim4_id
1103 vdur_index = 2
1104 db_vnfr = sample_vnfr
1105 mock_get_db_paths.return_value = (
1106 vim_info_path,
1107 vim_id,
1108 vnfr_id,
1109 vdur_path,
1110 vdur_index,
1111 db_vnfr,
1112 )
1113 mock_vim_info_exists.return_value = True
1114 with self.assertRaises(IndexError) as err:
1115 self.monitor._get_vm_data_from_db(sample_vm)
1116 self.assertEqual(str(err.exception.args[0]), "list index out of range")
1117 mock_deepcopy.assert_not_called()
1118 mock_get_db_paths.assert_called_once_with(target_record)
1119 mock_vim_info_exists.assert_called_once_with(db_vnfr, vdur_index)
1121 @patch("osm_ng_ro.monitor.MonitorVms._get_db_paths")
1122 @patch("osm_ng_ro.monitor.MonitorVms._check_if_vdur_vim_info_exists")
1123 @patch("osm_ng_ro.monitor.deepcopy")
1124 def test_get_vm_data_from_db_vnfr_without_proper_vim_id(
1125 self, mock_deepcopy, mock_vim_info_exists, mock_get_db_paths
1126 ):
1127 vim_id = "5239ed93-756b-408e-89f8-fcbf47a9d8f7"
1128 vdur_index = 0
1129 db_vnfr = sample_vnfr
1130 mock_get_db_paths.return_value = (
1131 vim_info_path,
1132 vim_id,
1133 vnfr_id,
1134 vdur_path,
1135 vdur_index,
1136 db_vnfr,
1137 )
1138 mock_vim_info_exists.return_value = True
1139 self.monitor._get_vm_data_from_db(sample_vm)
1140 mock_deepcopy.assert_not_called()
1141 mock_get_db_paths.assert_called_once_with(target_record)
1142 mock_vim_info_exists.assert_called_once_with(db_vnfr, vdur_index)
1144 def test_update_vim_info_for_deleted_vm_empty_input_dict(self):
1145 vdur_vim_info_update = {}
1146 self.monitor.update_vim_info_for_deleted_vm(vdur_vim_info_update)
1147 self.assertEqual(
1148 vdur_vim_info_update,
1149 deleted_externally,
1150 )
1152 def test_update_vim_info_for_deleted_vm_update_existing_info(self):
1153 vdur_vim_info_update = {
1154 "vim_status": "ACTIVE",
1155 "vim_message": None,
1156 "vim_id": vm1_id,
1157 "vim_name": "test7-vnf-hackfest_basic-VM-000000",
1158 }
1159 self.monitor.update_vim_info_for_deleted_vm(vdur_vim_info_update)
1160 self.assertEqual(
1161 vdur_vim_info_update,
1162 deleted_externally,
1163 )
1165 @patch("osm_ng_ro.monitor.MonitorVms._get_vm_data_from_db")
1166 @patch("osm_ng_ro.monitor.MonitorVms.update_vim_info_for_deleted_vm")
1167 @patch("osm_ng_ro.monitor.MonitorVms.backup_vdu_interfaces")
1168 @patch("osm_ng_ro.monitor.MonitorVms.update_in_database")
1169 def test_report_deleted_vdur_no_vm_data_in_db(
1170 self,
1171 mock_update_in_database,
1172 mock_backup_vdu_interfaces,
1173 mock_update_vim_info_for_deleted_vm,
1174 mock_get_vm_data_from_db,
1175 ):
1176 mock_get_vm_data_from_db.return_value = None
1177 self.monitor.report_deleted_vdur(sample_vm)
1178 self.assertEqual(mock_get_vm_data_from_db.call_count, 1)
1179 check_if_assert_not_called(
1180 [
1181 mock_update_in_database,
1182 mock_backup_vdu_interfaces,
1183 mock_update_vim_info_for_deleted_vm,
1184 ]
1185 )
1187 @patch("osm_ng_ro.monitor.MonitorVms._get_vm_data_from_db")
1188 @patch("osm_ng_ro.monitor.MonitorVms.update_vim_info_for_deleted_vm")
1189 @patch("osm_ng_ro.monitor.MonitorVms.backup_vdu_interfaces")
1190 @patch("osm_ng_ro.monitor.MonitorVms.update_in_database")
1191 def test_report_deleted_vdur(
1192 self,
1193 mock_update_in_database,
1194 mock_backup_vdu_interfaces,
1195 mock_update_vim_info_for_deleted_vm,
1196 mock_get_vm_data_from_db,
1197 ):
1198 existing_vim_info = sample_vim_info
1199 vdur_vim_info_update = deleted_externally
1200 mock_get_vm_data_from_db.return_value = (
1201 vdur_path,
1202 vdur_vim_info_update,
1203 None,
1204 existing_vim_info,
1205 vnfr_id,
1206 vim_info_path,
1207 )
1208 vdur_update = {
1209 vdur_path + ".status": "DELETED",
1210 }
1211 self.monitor.report_deleted_vdur(sample_vm)
1212 self.assertEqual(mock_get_vm_data_from_db.call_count, 1)
1213 mock_get_vm_data_from_db.assert_called_with(sample_vm)
1214 mock_update_vim_info_for_deleted_vm.assert_called_once_with(
1215 vdur_vim_info_update
1216 )
1217 mock_backup_vdu_interfaces.assert_called_once_with(vdur_vim_info_update)
1218 mock_update_in_database.assert_called_once_with(
1219 [vdur_update, {vim_info_path: vdur_vim_info_update}], vnfr_id
1220 )
1222 @patch("osm_ng_ro.monitor.MonitorVms._get_vm_data_from_db")
1223 @patch("osm_ng_ro.monitor.MonitorVms.update_vim_info_for_deleted_vm")
1224 @patch("osm_ng_ro.monitor.MonitorVms.backup_vdu_interfaces")
1225 @patch("osm_ng_ro.monitor.MonitorVms.update_in_database")
1226 def test_report_deleted_vdur_vm_db_already_updated(
1227 self,
1228 mock_update_in_database,
1229 mock_backup_vdu_interfaces,
1230 mock_update_vim_info_for_deleted_vm,
1231 mock_get_vm_data_from_db,
1232 ):
1233 vdur_vim_info_update = existing_vim_info = deleted_externally
1234 mock_get_vm_data_from_db.return_value = (
1235 vdur_path,
1236 vdur_vim_info_update,
1237 None,
1238 existing_vim_info,
1239 vnfr_id,
1240 vim_info_path,
1241 )
1242 self.monitor.report_deleted_vdur(sample_vm)
1243 self.assertEqual(mock_get_vm_data_from_db.call_count, 1)
1244 mock_get_vm_data_from_db.assert_called_with(sample_vm)
1245 mock_update_vim_info_for_deleted_vm.assert_called_once_with(
1246 vdur_vim_info_update
1247 )
1248 check_if_assert_not_called(
1249 [mock_backup_vdu_interfaces, mock_update_in_database]
1250 )
1252 @patch("osm_ng_ro.monitor.MonitorVms._get_vm_data_from_db")
1253 @patch("osm_ng_ro.monitor.MonitorVms.update_vim_info_for_deleted_vm")
1254 @patch("osm_ng_ro.monitor.MonitorVms.backup_vdu_interfaces")
1255 @patch("osm_ng_ro.monitor.MonitorVms.update_in_database")
1256 def test_report_deleted_vdur_get_vm_data_raises(
1257 self,
1258 mock_update_in_database,
1259 mock_backup_vdu_interfaces,
1260 mock_update_vim_info_for_deleted_vm,
1261 mock_get_vm_data_from_db,
1262 ):
1263 mock_get_vm_data_from_db.side_effect = IndexError("list index out of range.")
1264 with self.assertRaises(IndexError) as err:
1265 self.monitor.report_deleted_vdur(sample_vm)
1266 self.assertEqual(str(err.exception.args[0]), "list index out of range.")
1267 self.assertEqual(mock_get_vm_data_from_db.call_count, 1)
1268 mock_get_vm_data_from_db.assert_called_with(sample_vm)
1269 check_if_assert_not_called(
1270 [
1271 mock_update_vim_info_for_deleted_vm,
1272 mock_backup_vdu_interfaces,
1273 mock_update_in_database,
1274 ]
1275 )
1277 @patch("osm_ng_ro.monitor.MonitorVms._get_vm_data_from_db")
1278 @patch("osm_ng_ro.monitor.MonitorVms.update_vim_info_for_deleted_vm")
1279 @patch("osm_ng_ro.monitor.MonitorVms.backup_vdu_interfaces")
1280 @patch("osm_ng_ro.monitor.MonitorVms.update_in_database")
1281 def test_report_deleted_vdur_update_in_database_raises(
1282 self,
1283 mock_update_in_database,
1284 mock_backup_vdu_interfaces,
1285 mock_update_vim_info_for_deleted_vm,
1286 mock_get_vm_data_from_db,
1287 ):
1288 existing_vim_info = sample_vim_info
1289 vdur_vim_info_update = deleted_externally
1290 mock_update_in_database.side_effect = MonitorDbException(
1291 "Error while updating differences in VNFR."
1292 )
1293 mock_get_vm_data_from_db.return_value = (
1294 vdur_path,
1295 vdur_vim_info_update,
1296 None,
1297 existing_vim_info,
1298 vnfr_id,
1299 vim_info_path,
1300 )
1301 vdur_update = {
1302 vdur_path + ".status": "DELETED",
1303 }
1304 with self.assertRaises(MonitorDbException) as err:
1305 self.monitor.report_deleted_vdur(sample_vm)
1306 self.assertEqual(
1307 str(err.exception.args[0]), "Error while updating differences in VNFR."
1308 )
1309 self.assertEqual(mock_get_vm_data_from_db.call_count, 1)
1310 mock_get_vm_data_from_db.assert_called_with(sample_vm)
1311 mock_update_vim_info_for_deleted_vm.assert_called_once_with(
1312 vdur_vim_info_update
1313 )
1314 mock_backup_vdu_interfaces.assert_called_once_with(vdur_vim_info_update)
1315 mock_update_in_database.assert_called_once_with(
1316 [vdur_update, {vim_info_path: vdur_vim_info_update}], vnfr_id
1317 )
1319 @patch("osm_ng_ro.monitor.MonitorVms.report_vdur_updates")
1320 @patch("osm_ng_ro.monitor.MonitorVms.report_deleted_vdur")
1321 def test_update_vnfrs(self, mock_report_deleted_vdur, mock_report_vdur_updates):
1322 vms_to_monitor = [sample_vm, sample_vm2, sample_vm3]
1323 servers = [server1, server2, server3, server4]
1324 ports = {"ports": [port1, port2]}
1325 self.monitor.update_vnfrs(servers, ports, vms_to_monitor)
1326 self.assertEqual(mock_report_vdur_updates.call_count, 2)
1327 mock_report_deleted_vdur.assert_called_once_with(sample_vm3)
1328 _call_mock_report_vdur_updates = mock_report_vdur_updates.call_args_list
1329 self.assertEqual(
1330 _call_mock_report_vdur_updates[0].args,
1331 (server1, sample_vm, ports),
1332 )
1333 self.assertEqual(
1334 _call_mock_report_vdur_updates[1].args,
1335 (server2, sample_vm2, ports),
1336 )
1338 @patch("osm_ng_ro.monitor.MonitorVms.report_vdur_updates")
1339 @patch("osm_ng_ro.monitor.MonitorVms.report_deleted_vdur")
1340 def test_update_vnfrs_empty_vms_to_monitor(
1341 self, mock_report_deleted_vdur, mock_report_vdur_updates
1342 ):
1343 vms_to_monitor = []
1344 servers = [server1, server2, server3, server4]
1345 ports = {"ports": [port1, port2]}
1346 self.monitor.update_vnfrs(servers, ports, vms_to_monitor)
1347 check_if_assert_not_called([mock_report_deleted_vdur, mock_report_vdur_updates])
1349 @patch("osm_ng_ro.monitor.MonitorVms.report_vdur_updates")
1350 @patch("osm_ng_ro.monitor.MonitorVms.report_deleted_vdur")
1351 def test_update_vnfrs_empty_servers(
1352 self, mock_report_deleted_vdur, mock_report_vdur_updates
1353 ):
1354 vms_to_monitor = [sample_vm, sample_vm2, sample_vm3]
1355 servers = []
1356 ports = {"ports": [port1, port2]}
1357 self.monitor.update_vnfrs(servers, ports, vms_to_monitor)
1358 mock_report_vdur_updates.assert_not_called()
1359 self.assertEqual(mock_report_deleted_vdur.call_count, 3)
1360 _call_mock_report_deleted_vdur = mock_report_deleted_vdur.call_args_list
1361 self.assertEqual(
1362 _call_mock_report_deleted_vdur[0].args[0],
1363 (sample_vm),
1364 )
1365 self.assertEqual(
1366 _call_mock_report_deleted_vdur[1].args[0],
1367 (sample_vm2),
1368 )
1369 self.assertEqual(
1370 _call_mock_report_deleted_vdur[2].args[0],
1371 (sample_vm3),
1372 )
1374 @patch("osm_ng_ro.monitor.MonitorVms.report_vdur_updates")
1375 @patch("osm_ng_ro.monitor.MonitorVms.report_deleted_vdur")
1376 def test_update_vnfrs_report_vdur_updates_raises(
1377 self, mock_report_deleted_vdur, mock_report_vdur_updates
1378 ):
1379 vms_to_monitor = [sample_vm, sample_vm2, sample_vm3]
1380 servers = [server1, server2, server3, server4]
1381 ports = {"ports": [port1, port2]}
1382 mock_report_vdur_updates.side_effect = IndexError("list index out of range.")
1383 with self.assertRaises(IndexError) as err:
1384 self.monitor.update_vnfrs(servers, ports, vms_to_monitor)
1385 self.assertEqual(str(err.exception.args[0]), "list index out of range.")
1386 self.assertEqual(mock_report_vdur_updates.call_count, 1)
1387 mock_report_deleted_vdur.assert_not_called()
1388 _call_mock_report_vdur_updates = mock_report_vdur_updates.call_args_list
1389 self.assertEqual(
1390 _call_mock_report_vdur_updates[0].args,
1391 (server1, sample_vm, ports),
1392 )
1394 @patch("osm_ng_ro.monitor.MonitorVms.report_vdur_updates")
1395 @patch("osm_ng_ro.monitor.MonitorVms.report_deleted_vdur")
1396 def test_update_vnfrs_report_deleted_vdur_raises(
1397 self, mock_report_deleted_vdur, mock_report_vdur_updates
1398 ):
1399 vms_to_monitor = [sample_vm, sample_vm2, sample_vm3]
1400 servers = [server1, server2, server3, server4]
1401 ports = {"ports": [port1, port2]}
1402 mock_report_deleted_vdur.side_effect = DbException("DB is not in active state.")
1403 with self.assertRaises(DbException) as err:
1404 self.monitor.update_vnfrs(servers, ports, vms_to_monitor)
1405 self.assertEqual(
1406 str(err.exception.args[0]), "database exception DB is not in active state."
1407 )
1408 self.assertEqual(mock_report_vdur_updates.call_count, 2)
1409 mock_report_deleted_vdur.assert_called_once_with(sample_vm3)
1410 _call_mock_report_vdur_updates = mock_report_vdur_updates.call_args_list
1411 self.assertEqual(
1412 _call_mock_report_vdur_updates[0].args,
1413 (server1, sample_vm, ports),
1414 )
1415 self.assertEqual(
1416 _call_mock_report_vdur_updates[1].args,
1417 (server2, sample_vm2, ports),
1418 )
1420 @patch("osm_ng_ro.monitor.yaml")
1421 def test_serialize_string_value(self, mock_yaml):
1422 value = "some string"
1423 result = self.monitor.serialize(value)
1424 mock_yaml.dump.assert_not_called()
1425 self.assertEqual(result, value)
1427 @patch("osm_ng_ro.monitor.yaml")
1428 def test_serialize_list_value(self, mock_yaml):
1429 value = [
1430 {"version": 3.4},
1431 ["image", "ghcr.io/foo/mysvc"],
1432 {"MYSVC_ENV": "to_nice_yaml"},
1433 ]
1434 output = [
1435 {"version": 3.4},
1436 ["image", "ghcr.io/foo/mysvc"],
1437 {"MYSVC_ENV": "to_nice_yaml"},
1438 ]
1439 mock_yaml.dump.return_value = output
1440 result = self.monitor.serialize(value)
1441 mock_yaml.dump.assert_called_once()
1442 self.assertEqual(result, output)
1444 @patch("osm_ng_ro.monitor.yaml")
1445 def test_serialize_dict_value(self, mock_yaml):
1446 value = {
1447 "version": 3.4,
1448 "MYSVC_ENV": "to_nice_yaml_to_nice_yaml_to_nice_yaml_to_nice_yaml_to_nice_yaml",
1449 }
1450 output = {
1451 "MYSVC_ENV": "to_nice_yaml_to_nice_yaml_to_nice_yaml_to_nice_yaml_to_nice_yaml",
1452 "version": 3.4,
1453 }
1454 mock_yaml.dump.return_value = output
1455 result = self.monitor.serialize(value)
1456 mock_yaml.dump.assert_called_once()
1457 self.assertEqual(result, output)
1459 @patch("osm_ng_ro.monitor.yaml")
1460 def test_serialize_raise_representer_error(self, mock_yaml):
1461 value = {
1462 "name": {"firstname": str, "lastname": str},
1463 "age": int,
1464 }
1465 mock_yaml.dump.side_effect = yaml.representer.RepresenterError(
1466 "cannot represent an object"
1467 )
1468 result = self.monitor.serialize(value)
1469 mock_yaml.dump.assert_called_once()
1470 self.assertEqual(result, str(value))
1472 @patch("osm_ng_ro.monitor.yaml")
1473 def test_serialize_raise_yaml_error(self, mock_yaml):
1474 value = {
1475 "name": {"firstname": str, "lastname": str},
1476 "age": int,
1477 }
1479 mock_yaml.dump.side_effect = yaml.YAMLError("cannot represent an object.")
1480 with self.assertRaises(yaml.YAMLError) as err:
1481 result = self.monitor.serialize(value)
1482 self.assertEqual(result, None)
1483 self.assertEqual(str(err.exception.args[0]), "cannot represent an object.")
1484 mock_yaml.dump.assert_called_once()
1486 @patch("osm_ng_ro.monitor.MonitorVms.serialize")
1487 def test_get_server_info_with_user_data(self, mock_serialize):
1488 all_server_info = deepcopy(server_other_info)
1489 user_data = {
1490 "OS-EXT-SRV-ATTR:user_data": "EXT-USER-DATA",
1491 "user_data": "some-data",
1492 }
1493 mock_serialize.return_value = serialized_server_info
1494 all_server_info.update(user_data)
1495 server5 = create_server(vm1_id, "server5", info=all_server_info)
1496 result = self.monitor._get_server_info(server5)
1497 self.assertEqual(result, serialized_server_info)
1498 mock_serialize.assert_called_once_with(server_other_info)
1500 @patch("osm_ng_ro.monitor.MonitorVms.serialize")
1501 def test_get_server_info_without_user_data(self, mock_serialize):
1502 mock_serialize.return_value = serialized_server_info
1503 server5 = create_server(vm1_id, "server5", info=server_other_info)
1504 result = self.monitor._get_server_info(server5)
1505 self.assertEqual(result, serialized_server_info)
1506 mock_serialize.assert_called_once_with(server_other_info)
1508 @patch("osm_ng_ro.monitor.MonitorVms.serialize")
1509 def test_get_server_info_empty_server_info(self, mock_serialize):
1510 server_other_info = {}
1511 expected_result = {}
1512 mock_serialize.return_value = expected_result
1513 server5 = create_server(vm1_id, "server5", info=server_other_info)
1514 result = self.monitor._get_server_info(server5)
1515 self.assertEqual(result, expected_result)
1516 mock_serialize.assert_called_once_with(server_other_info)
1518 @patch("osm_ng_ro.monitor.MonitorVms.serialize")
1519 def test_get_server_info_serialize_raises(self, mock_serialize):
1520 server_other_info = {
1521 "admin_state_up": "true",
1522 "binding:host_id": int,
1523 "binding:profile": {},
1524 "binding:vif_type": str,
1525 "binding:vnic_type": "normal",
1526 "created_at": "2023-02-22T05:35:46Z",
1527 }
1528 mock_serialize.side_effect = yaml.YAMLError("cannot represent an object.")
1529 server5 = create_server(vm1_id, "server5", info=server_other_info)
1530 with self.assertRaises(yaml.YAMLError) as err:
1531 self.monitor._get_server_info(server5)
1532 self.assertEqual(str(err.exception.args[0]), "cannot represent an object.")
1533 mock_serialize.assert_called_once_with(server_other_info)
1535 @patch("osm_ng_ro.monitor.MonitorVms._get_server_info")
1536 def test_check_vm_status_updates_server_status_ok(self, mock_server_info):
1537 server6 = create_server("server6-id", "server6", status="PAUSED")
1538 mock_server_info.return_value = serialized_server_info
1539 vdur_vim_info_update = {}
1540 vdur_update = {}
1541 expected_vdur_vim_info_update = {
1542 "vim_status": "PAUSED",
1543 "vim_details": serialized_server_info,
1544 "vim_id": server6.id,
1545 "vim_name": server6.name,
1546 }
1547 expected_vdur_update = {
1548 "vdur.0.status": "PAUSED",
1549 "vdur.0.name": server6.name,
1550 }
1551 self.monitor.check_vm_status_updates(
1552 vdur_vim_info_update, vdur_update, server6, vdur_path
1553 )
1554 self.assertDictEqual(vdur_vim_info_update, expected_vdur_vim_info_update)
1555 self.assertDictEqual(vdur_update, expected_vdur_update)
1556 mock_server_info.assert_called_once_with(server6)
1558 @patch("osm_ng_ro.monitor.MonitorVms._get_server_info")
1559 def test_check_vm_status_updates_server_status_nok(self, mock_server_info):
1560 server8 = create_server("server8-id", "server8", status="FAILED")
1561 mock_server_info.return_value = serialized_server_info
1562 vdur_vim_info_update = {}
1563 vdur_update = {}
1564 expected_vdur_vim_info_update = {
1565 "vim_status": "FAILED",
1566 "vim_details": serialized_server_info,
1567 "vim_id": server8.id,
1568 "vim_name": server8.name,
1569 "vim_message": "VIM status reported FAILED",
1570 }
1571 expected_vdur_update = {
1572 "vdur.0.status": "FAILED",
1573 "vdur.0.name": server8.name,
1574 }
1575 self.monitor.check_vm_status_updates(
1576 vdur_vim_info_update, vdur_update, server8, vdur_path
1577 )
1578 self.assertDictEqual(vdur_vim_info_update, expected_vdur_vim_info_update)
1579 self.assertDictEqual(vdur_update, expected_vdur_update)
1580 mock_server_info.assert_called_once_with(server8)
1582 @patch("osm_ng_ro.monitor.MonitorVms._get_server_info")
1583 def test_check_vm_status_updates_get_server_info_raises(self, mock_server_info):
1584 server8 = create_server("server8-id", "server8", status="FAILED")
1585 mock_server_info.side_effect = yaml.YAMLError("Cannot represent an object.")
1586 vdur_vim_info_update = {}
1587 vdur_update = {}
1588 expected_vdur_vim_info_update = {
1589 "vim_status": "FAILED",
1590 "vim_message": "VIM status reported FAILED",
1591 }
1592 expected_vdur_update = {
1593 "vdur.0.status": "FAILED",
1594 }
1595 with self.assertRaises(yaml.YAMLError) as err:
1596 self.monitor.check_vm_status_updates(
1597 vdur_vim_info_update, vdur_update, server8, vdur_path
1598 )
1599 self.assertEqual(str(err.exception.args[0]), "Cannot represent an object.")
1600 self.assertDictEqual(vdur_vim_info_update, expected_vdur_vim_info_update)
1601 self.assertDictEqual(vdur_update, expected_vdur_update)
1602 mock_server_info.assert_called_once_with(server8)
1604 def test_get_interface_info(self):
1605 interface = {"vim_interface_id": "4d081f50-e13a-4306-a67e-1edb28d76013"}
1606 ports = {"ports": [port1, port2]}
1607 result = self.monitor.get_interface_info(ports, interface, server1)
1608 self.assertEqual(result, port1)
1610 def test_get_interface_info_port_id_mismatch(self):
1611 interface = {"vim_interface_id": "4d081f50-e13a-4306-a67e-1edb28d76013"}
1612 ports = {"ports": [port2]}
1613 result = self.monitor.get_interface_info(ports, interface, server1)
1614 self.assertEqual(result, None)
1616 def test_get_interface_info_device_id_mismatch(self):
1617 interface = {"vim_interface_id": "4d081f50-e13a-4306-a67e-1edb28d76013"}
1618 ports = {"ports": [port1, port2]}
1619 result = self.monitor.get_interface_info(ports, interface, server2)
1620 self.assertEqual(result, None)
1622 def test_get_interface_info_empty_ports(self):
1623 interface = {"vim_interface_id": "4d081f50-e13a-4306-a67e-1edb28d76013"}
1624 ports = {"ports": []}
1625 result = self.monitor.get_interface_info(ports, interface, server2)
1626 self.assertEqual(result, None)
1628 def test_check_vlan_pci_update(self):
1629 interface_info = interface_with_binding
1630 index = 1
1631 vdur_vim_info_update = {"interfaces": [{}, {}]}
1632 expected_vdur_vim_info_update = {
1633 "interfaces": [{}, {"pci": "0000:86:17.4", "vlan": 400}]
1634 }
1635 self.monitor.check_vlan_pci_updates(interface_info, index, vdur_vim_info_update)
1636 self.assertDictEqual(vdur_vim_info_update, expected_vdur_vim_info_update)
1638 def test_check_vlan_pci_update_empty_interface_info(self):
1639 interface_info = {}
1640 index = 1
1641 vdur_vim_info_update = {"interfaces": [{}, {}]}
1642 expected_vdur_vim_info_update = {"interfaces": [{}, {}]}
1643 self.monitor.check_vlan_pci_updates(interface_info, index, vdur_vim_info_update)
1644 self.assertDictEqual(vdur_vim_info_update, expected_vdur_vim_info_update)
1646 def test_check_vlan_pci_update_index_out_of_range(self):
1647 interface_info = interface_with_binding
1648 index = 3
1649 vdur_vim_info_update = {"interfaces": [{}]}
1650 expected_vdur_vim_info_update = {"interfaces": [{}]}
1651 with self.assertRaises(IndexError) as err:
1652 self.monitor.check_vlan_pci_updates(
1653 interface_info, index, vdur_vim_info_update
1654 )
1655 self.assertEqual(str(err.exception.args[0]), "list index out of range")
1656 self.assertEqual(vdur_vim_info_update, expected_vdur_vim_info_update)
1658 def test_check_vlan_pci_update_empty_vdur_vim_info_update(self):
1659 interface_info = interface_with_binding
1660 index = 0
1661 vdur_vim_info_update = {}
1662 expected_vdur_vim_info_update = {}
1663 with self.assertRaises(KeyError) as err:
1664 self.monitor.check_vlan_pci_updates(
1665 interface_info, index, vdur_vim_info_update
1666 )
1667 self.assertEqual(str(err.exception.args[0]), "interfaces")
1668 self.assertEqual(vdur_vim_info_update, expected_vdur_vim_info_update)
1670 @patch("osm_ng_ro.monitor.MonitorVms._get_current_ip_address")
1671 def test_check_vdur_interface_updates(self, mock_get_current_ip_address):
1672 vdur_update, vnfr_update = {}, {}
1673 index = 0
1674 interface_info = {
1675 "fixed_ips": [{"ip_address": ip1_addr}],
1676 "mac_address": mac1_addr,
1677 }
1678 mock_get_current_ip_address.return_value = ip1_addr
1679 expected_vdur_update = {
1680 "vdur.0.interfaces.0.ip-address": ip1_addr,
1681 "vdur.0.ip-address": ip1_addr,
1682 "vdur.0.interfaces.0.mac-address": mac1_addr,
1683 }
1684 expected_vnfr_update = {
1685 "35c034cc-8c5b-48c4-bfa2-17a71577ef19.ip-address": ip1_addr
1686 }
1687 self.monitor.check_vdur_interface_updates(
1688 vdur_update,
1689 vdur_path,
1690 index,
1691 interface_info,
1692 old_interface2,
1693 vnfr_update,
1694 vnfr_id,
1695 )
1696 self.assertEqual(vnfr_update, expected_vnfr_update)
1697 self.assertEqual(vdur_update, expected_vdur_update)
1698 mock_get_current_ip_address.assert_called_once_with(interface_info)
1700 @patch("osm_ng_ro.monitor.MonitorVms._get_current_ip_address")
1701 def test_check_vdur_interface_updates_not_mgmt_interface(
1702 self, mock_get_current_ip_address
1703 ):
1704 vdur_update, vnfr_update = {}, {}
1705 index = 0
1706 interface_info = {
1707 "fixed_ips": [{"ip_address": ip1_addr}],
1708 "mac_address": mac1_addr,
1709 }
1710 mock_get_current_ip_address.return_value = ip1_addr
1711 old_interface = {}
1712 expected_vdur_update = {
1713 "vdur.0.interfaces.0.ip-address": ip1_addr,
1714 "vdur.0.interfaces.0.mac-address": mac1_addr,
1715 }
1716 self.monitor.check_vdur_interface_updates(
1717 vdur_update,
1718 vdur_path,
1719 index,
1720 interface_info,
1721 old_interface,
1722 vnfr_update,
1723 vnfr_id,
1724 )
1725 self.assertEqual(vnfr_update, {})
1726 self.assertEqual(vdur_update, expected_vdur_update)
1727 mock_get_current_ip_address.assert_called_once_with(interface_info)
1729 @patch("osm_ng_ro.monitor.MonitorVms._get_current_ip_address")
1730 def test_check_vdur_interface_updates_without_mac_address(
1731 self, mock_get_current_ip_address
1732 ):
1733 vdur_update, vnfr_update = {}, {}
1734 index = 0
1735 interface_info = {"fixed_ips": [{"ip_address": ip1_addr}]}
1736 mock_get_current_ip_address.return_value = ip1_addr
1737 expected_vdur_update = {
1738 "vdur.0.interfaces.0.ip-address": ip1_addr,
1739 "vdur.0.ip-address": ip1_addr,
1740 "vdur.0.interfaces.0.mac-address": None,
1741 }
1742 expected_vnfr_update = {
1743 "35c034cc-8c5b-48c4-bfa2-17a71577ef19.ip-address": ip1_addr
1744 }
1745 self.monitor.check_vdur_interface_updates(
1746 vdur_update,
1747 vdur_path,
1748 index,
1749 interface_info,
1750 old_interface2,
1751 vnfr_update,
1752 vnfr_id,
1753 )
1754 self.assertEqual(vnfr_update, expected_vnfr_update)
1755 self.assertEqual(vdur_update, expected_vdur_update)
1756 mock_get_current_ip_address.assert_called_once_with(interface_info)
1758 @patch("osm_ng_ro.monitor.MonitorVms._get_current_ip_address")
1759 def test_check_vdur_interface_updates_without_ip_address(
1760 self, mock_get_current_ip_address
1761 ):
1762 vdur_update, vnfr_update = {}, {}
1763 index = 0
1764 interface_info = {"fixed_ips": [], "mac_address": mac1_addr}
1765 mock_get_current_ip_address.return_value = None
1766 expected_vdur_update = {
1767 "vdur.0.interfaces.0.mac-address": mac1_addr,
1768 }
1769 expected_vnfr_update = {}
1770 self.monitor.check_vdur_interface_updates(
1771 vdur_update,
1772 vdur_path,
1773 index,
1774 interface_info,
1775 old_interface2,
1776 vnfr_update,
1777 vnfr_id,
1778 )
1779 self.assertEqual(vnfr_update, expected_vnfr_update)
1780 self.assertEqual(vdur_update, expected_vdur_update)
1781 mock_get_current_ip_address.assert_called_once_with(interface_info)
1783 @patch("osm_ng_ro.monitor.MonitorVms._get_current_ip_address")
1784 def test_check_vdur_interface_updates_wrong_interface_info_format(
1785 self, mock_get_current_ip_address
1786 ):
1787 vdur_update, vnfr_update = {}, {}
1788 index = 0
1789 interface_info = {"fixed_ips": ip1_addr, "mac_address": mac1_addr}
1790 mock_get_current_ip_address.side_effect = TypeError(
1791 "str is not list like object."
1792 )
1793 old_interface = {}
1794 with self.assertRaises(TypeError) as err:
1795 self.monitor.check_vdur_interface_updates(
1796 vdur_update,
1797 vdur_path,
1798 index,
1799 interface_info,
1800 old_interface,
1801 vnfr_update,
1802 vnfr_id,
1803 )
1804 self.assertEqual(str(err.exception), "str is not list like object.")
1805 self.assertEqual(vnfr_update, {})
1806 self.assertEqual(vdur_update, {})
1807 mock_get_current_ip_address.assert_called_once_with(interface_info)
1809 def test_get_current_ip_address(self):
1810 interface_info = {
1811 "fixed_ips": [{"ip_address": ip1_addr}],
1812 "mac_address": mac1_addr,
1813 }
1814 result = self.monitor._get_current_ip_address(interface_info)
1815 self.assertEqual(result, ip1_addr)
1817 def test_get_current_ip_address_no_ip(self):
1818 interface_info = {"fixed_ips": [{}], "mac_address": mac1_addr}
1819 result = self.monitor._get_current_ip_address(interface_info)
1820 self.assertEqual(result, None)
1822 def test_backup_vdu_interfaces_without_vim_message(self):
1823 vdur_vim_info_update = {
1824 "interfaces": {"mac_address": mac1_addr},
1825 }
1826 expected_vdur_vim_info_update = {
1827 "interfaces": {"mac_address": mac1_addr},
1828 "interfaces_backup": {"mac_address": mac1_addr},
1829 }
1830 self.monitor.backup_vdu_interfaces(vdur_vim_info_update)
1831 self.assertDictEqual(expected_vdur_vim_info_update, vdur_vim_info_update)
1833 def test_backup_vdu_interfaces_with_vim_message(self):
1834 vdur_vim_info_update = {
1835 "interfaces": {"mac_address": mac1_addr},
1836 "vim_message": "Deleted Externally",
1837 }
1838 expected_vdur_vim_info_update = {
1839 "interfaces": {"mac_address": mac1_addr},
1840 "vim_message": "Deleted Externally",
1841 }
1842 self.monitor.backup_vdu_interfaces(vdur_vim_info_update)
1843 self.assertDictEqual(expected_vdur_vim_info_update, vdur_vim_info_update)
1845 def test_backup_vdu_interfaces_with_empty_interfaces(self):
1846 vdur_vim_info_update = {
1847 "interfaces": {},
1848 }
1849 expected_vdur_vim_info_update = {
1850 "interfaces": {},
1851 }
1852 self.monitor.backup_vdu_interfaces(vdur_vim_info_update)
1853 self.assertDictEqual(expected_vdur_vim_info_update, vdur_vim_info_update)
1855 @patch("osm_ng_ro.monitor.MonitorVms.serialize")
1856 def test_update_vdur_vim_info_interfaces(self, mock_serialize):
1857 index = 1
1858 vdur_vim_info_update = {
1859 "interfaces": [{}, {"mac_address": mac1_addr, "compute_node": "host1"}]
1860 }
1861 all_server_info = deepcopy(server_other_info)
1862 host_data = {"OS-EXT-SRV-ATTR:host": "nova"}
1863 mock_serialize.return_value = serialized_interface_info
1864 all_server_info.update(host_data)
1865 server7 = create_server(vm1_id, "server7", info=all_server_info)
1866 expected_vdur_vim_info_update = {
1867 "interfaces": [
1868 {},
1869 {
1870 "mac_address": mac2_addr,
1871 "compute_node": "nova",
1872 "vim_info": serialized_interface_info,
1873 "vim_net_id": net1_id,
1874 "ip_address": ip1_addr,
1875 },
1876 ]
1877 }
1878 self.monitor.update_vdur_vim_info_interfaces(
1879 vdur_vim_info_update, index, interface_info2, server7
1880 )
1881 self.assertDictEqual(vdur_vim_info_update, expected_vdur_vim_info_update)
1882 mock_serialize.assert_called_once_with(interface_info2)
1884 @patch("osm_ng_ro.monitor.MonitorVms.serialize")
1885 def test_update_vdur_vim_info_interfaces_serialize_raises(self, mock_serialize):
1886 index = 1
1887 vdur_vim_info_update = {
1888 "interfaces": [{}, {"mac_address": mac1_addr, "compute_node": "host1"}]
1889 }
1890 all_server_info = deepcopy(server_other_info)
1891 host_data = {"OS-EXT-SRV-ATTR:host": "nova"}
1892 mock_serialize.side_effect = yaml.YAMLError("Cannot represent an object.")
1893 all_server_info.update(host_data)
1894 server7 = create_server(vm1_id, "server7", info=all_server_info)
1895 expected_vdur_vim_info = deepcopy(vdur_vim_info_update)
1896 with self.assertRaises(yaml.YAMLError) as err:
1897 self.monitor.update_vdur_vim_info_interfaces(
1898 vdur_vim_info_update, index, interface_info2, server7
1899 )
1900 self.assertEqual(str(err.exception), "Cannot represent an object.")
1901 self.assertDictEqual(vdur_vim_info_update, expected_vdur_vim_info)
1902 mock_serialize.assert_called_once_with(interface_info2)
1904 @patch("osm_ng_ro.monitor.MonitorVms.serialize")
1905 def test_update_vdur_vim_info_interfaces_empty_interface_info(self, mock_serialize):
1906 index = 1
1907 vdur_vim_info_update = {
1908 "interfaces": [{}, {"mac_address": mac1_addr, "compute_node": "host1"}]
1909 }
1910 interface_info = {}
1911 all_server_info = deepcopy(server_other_info)
1912 host_data = {"OS-EXT-SRV-ATTR:host": "nova"}
1913 all_server_info.update(host_data)
1914 server7 = create_server(vm1_id, "server7", info=all_server_info)
1915 expected_vdur_vim_info = deepcopy(vdur_vim_info_update)
1916 with self.assertRaises(KeyError) as err:
1917 self.monitor.update_vdur_vim_info_interfaces(
1918 vdur_vim_info_update, index, interface_info, server7
1919 )
1920 self.assertEqual(str(err.exception.args[0]), "mac_address")
1921 self.assertDictEqual(vdur_vim_info_update, expected_vdur_vim_info)
1922 mock_serialize.assert_not_called()
1924 @patch("osm_ng_ro.monitor.MonitorVms.serialize")
1925 def test_update_vdur_vim_info_interfaces_invalid_vdur_vim_info(
1926 self, mock_serialize
1927 ):
1928 index = 1
1929 vdur_vim_info_update = {
1930 "interfaces": [{"mac_address": mac1_addr, "compute_node": "host1"}, {}]
1931 }
1932 expected_vdur_vim_info = deepcopy(vdur_vim_info_update)
1933 with self.assertRaises(MonitorVmsException) as err:
1934 self.monitor.update_vdur_vim_info_interfaces(
1935 vdur_vim_info_update, index, interface_info2, server7
1936 )
1937 self.assertEqual(
1938 str(err.exception.args[0]), "Existing interfaces info could not found."
1939 )
1940 self.assertDictEqual(vdur_vim_info_update, expected_vdur_vim_info)
1941 mock_serialize.assert_not_called()
1943 @patch("osm_ng_ro.monitor.MonitorVms.update_vdur_vim_info_interfaces")
1944 @patch("osm_ng_ro.monitor.MonitorVms.check_vlan_pci_updates")
1945 @patch("osm_ng_ro.monitor.MonitorVms.check_vdur_interface_updates")
1946 def test_prepare_interface_updates(
1947 self,
1948 mock_check_vdur_interface_updates,
1949 mock_check_vlan_pci_updates,
1950 mock_update_vdur_vim_info_interfaces,
1951 ):
1952 vdur_vim_info_update = {
1953 "interfaces": [{"mac_address": mac1_addr, "compute_node": "host1"}]
1954 }
1955 interface_info = {
1956 "fixed_ips": [{"ip_address": ip1_addr}],
1957 "mac_address": mac2_addr,
1958 "network_id": net1_id,
1959 }
1960 old_interface = {
1961 "mgmt_vdu_interface": True,
1962 "mgmt_vnf_interface": True,
1963 }
1964 index = 0
1965 vnfr_update, vdur_update = {}, {}
1966 self.monitor.prepare_interface_updates(
1967 vdur_vim_info_update,
1968 index,
1969 interface_info,
1970 server7,
1971 vdur_path,
1972 vnfr_update,
1973 old_interface2,
1974 vdur_update,
1975 vnfr_id,
1976 )
1977 mock_update_vdur_vim_info_interfaces.assert_called_once_with(
1978 vdur_vim_info_update, index, interface_info, server7
1979 )
1980 mock_check_vlan_pci_updates.assert_called_once_with(
1981 interface_info, index, vdur_vim_info_update
1982 )
1983 mock_check_vdur_interface_updates.assert_called_once_with(
1984 vdur_update,
1985 vdur_path,
1986 index,
1987 interface_info,
1988 old_interface,
1989 vnfr_update,
1990 vnfr_id,
1991 )
1993 @patch("osm_ng_ro.monitor.MonitorVms.update_vdur_vim_info_interfaces")
1994 @patch("osm_ng_ro.monitor.MonitorVms.check_vlan_pci_updates")
1995 @patch("osm_ng_ro.monitor.MonitorVms.check_vdur_interface_updates")
1996 def test_prepare_interface_updates_update_vdur_vim_info_interfaces_raises(
1997 self,
1998 mock_check_vdur_interface_updates,
1999 mock_check_vlan_pci_updates,
2000 mock_update_vdur_vim_info_interfaces,
2001 ):
2002 vdur_vim_info_update = {
2003 "interfaces": [{"mac_address": mac1_addr, "compute_node": "host1"}]
2004 }
2005 index = 0
2006 vnfr_update, vdur_update = {}, {}
2007 mock_update_vdur_vim_info_interfaces.side_effect = MonitorVmsException(
2008 "Existing interfaces info could not found."
2009 )
2010 with self.assertRaises(MonitorVmsException) as err:
2011 self.monitor.prepare_interface_updates(
2012 vdur_vim_info_update,
2013 index,
2014 interface_info2,
2015 server7,
2016 vdur_path,
2017 vnfr_update,
2018 old_interface2,
2019 vdur_update,
2020 vnfr_id,
2021 )
2022 self.assertEqual(
2023 str(err.exception.args[0]), "Existing interfaces info could not found."
2024 )
2025 mock_update_vdur_vim_info_interfaces.assert_called_once_with(
2026 vdur_vim_info_update, index, interface_info2, server7
2027 )
2028 check_if_assert_not_called(
2029 [mock_check_vlan_pci_updates, mock_check_vdur_interface_updates]
2030 )
2032 @patch("osm_ng_ro.monitor.MonitorVms.update_vdur_vim_info_interfaces")
2033 @patch("osm_ng_ro.monitor.MonitorVms.check_vlan_pci_updates")
2034 @patch("osm_ng_ro.monitor.MonitorVms.check_vdur_interface_updates")
2035 def test_prepare_interface_updates_check_vlan_pci_updates_raises(
2036 self,
2037 mock_check_vdur_interface_updates,
2038 mock_check_vlan_pci_updates,
2039 mock_update_vdur_vim_info_interfaces,
2040 ):
2041 vdur_vim_info_update = {
2042 "interfaces": [{"mac_address": mac1_addr, "compute_node": "host1"}]
2043 }
2044 index = 0
2045 vnfr_update, vdur_update = {}, {}
2046 mock_check_vlan_pci_updates.side_effect = KeyError("vlan is not found.")
2047 with self.assertRaises(KeyError) as err:
2048 self.monitor.prepare_interface_updates(
2049 vdur_vim_info_update,
2050 index,
2051 interface_info2,
2052 server7,
2053 vdur_path,
2054 vnfr_update,
2055 old_interface2,
2056 vdur_update,
2057 vnfr_id,
2058 )
2059 self.assertEqual(str(err.exception.args[0]), "vlan is not found.")
2060 mock_update_vdur_vim_info_interfaces.assert_called_once_with(
2061 vdur_vim_info_update, index, interface_info2, server7
2062 )
2063 mock_check_vlan_pci_updates.assert_called_once_with(
2064 interface_info2, index, vdur_vim_info_update
2065 )
2066 mock_check_vdur_interface_updates.assert_not_called()
2068 @patch("osm_ng_ro.monitor.MonitorVms.get_interface_info")
2069 @patch("osm_ng_ro.monitor.MonitorVms.prepare_interface_updates")
2070 def test_check_vm_interface_updates(
2071 self, mock_prepare_interface_updates, mock_get_interface_info
2072 ):
2073 vdur_vim_info_update = {
2074 "interfaces": [{"mac_address": mac1_addr, "compute_node": "host1"}]
2075 }
2076 index = 0
2077 interface_info = {
2078 "fixed_ips": [{"ip_address": ip1_addr}],
2079 "mac_address": mac2_addr,
2080 "network_id": net1_id,
2081 "status": "ACTIVE",
2082 }
2083 vnfr_update, vdur_update = {}, {}
2084 ports = {"ports": [port1, port2]}
2085 existing_vim_info = sample_vim_info
2086 mock_get_interface_info.return_value = interface_info
2087 self.monitor.check_vm_interface_updates(
2088 server7,
2089 existing_vim_info,
2090 ports,
2091 vdur_vim_info_update,
2092 vdur_update,
2093 vdur_path,
2094 vnfr_update,
2095 vnfr_id,
2096 )
2097 mock_get_interface_info.assert_called_once_with(ports, old_interface, server7)
2098 mock_prepare_interface_updates.assert_called_once_with(
2099 vdur_vim_info_update,
2100 index,
2101 interface_info,
2102 server7,
2103 vdur_path,
2104 vnfr_update,
2105 old_interface,
2106 vdur_update,
2107 vnfr_id,
2108 )
2109 self.assertNotIn("vim_message", vdur_vim_info_update)
2111 @patch("osm_ng_ro.monitor.MonitorVms.get_interface_info")
2112 @patch("osm_ng_ro.monitor.MonitorVms.prepare_interface_updates")
2113 def test_check_vm_interface_updates_interface_new_status_is_nok(
2114 self, mock_prepare_interface_updates, mock_get_interface_info
2115 ):
2116 vdur_vim_info_update = {
2117 "interfaces": [{"mac_address": mac1_addr, "compute_node": "host1"}]
2118 }
2119 interface_info = {
2120 "fixed_ips": [{"ip_address": ip1_addr}],
2121 "mac_address": mac2_addr,
2122 "network_id": net1_id,
2123 "status": "DOWN",
2124 }
2125 vnfr_update, vdur_update = {}, {}
2126 ports = {"ports": [port1, port2]}
2127 existing_vim_info = sample_vim_info
2128 mock_get_interface_info.return_value = interface_info
2129 self.monitor.check_vm_interface_updates(
2130 server7,
2131 existing_vim_info,
2132 ports,
2133 vdur_vim_info_update,
2134 vdur_update,
2135 vdur_path,
2136 vnfr_update,
2137 vnfr_id,
2138 )
2139 mock_get_interface_info.assert_called_once_with(ports, old_interface, server7)
2140 mock_prepare_interface_updates.assert_not_called()
2141 self.assertEqual(
2142 vdur_vim_info_update["vim_message"],
2143 "Interface 4d081f50-e13a-4306-a67e-1edb28d76013 status: DOWN",
2144 )
2146 @patch("osm_ng_ro.monitor.MonitorVms.get_interface_info")
2147 @patch("osm_ng_ro.monitor.MonitorVms.prepare_interface_updates")
2148 def test_check_vm_interface_updates_no_new_interface_info(
2149 self, mock_prepare_interface_updates, mock_get_interface_info
2150 ):
2151 vdur_vim_info_update = {
2152 "interfaces": [{"mac_address": mac1_addr, "compute_node": "host1"}]
2153 }
2154 vnfr_update, vdur_update = {}, {}
2155 ports = {"ports": [port1, port2]}
2156 existing_vim_info = sample_vim_info
2157 mock_get_interface_info.return_value = None
2158 self.monitor.check_vm_interface_updates(
2159 server7,
2160 existing_vim_info,
2161 ports,
2162 vdur_vim_info_update,
2163 vdur_update,
2164 vdur_path,
2165 vnfr_update,
2166 vnfr_id,
2167 )
2168 mock_get_interface_info.assert_called_once_with(ports, old_interface, server7)
2169 mock_prepare_interface_updates.assert_not_called()
2170 self.assertEqual(
2171 vdur_vim_info_update["vim_message"],
2172 "Interface 4d081f50-e13a-4306-a67e-1edb28d76013 deleted externally.",
2173 )
2175 @patch("osm_ng_ro.monitor.MonitorVms.get_interface_info")
2176 @patch("osm_ng_ro.monitor.MonitorVms.prepare_interface_updates")
2177 def test_check_vm_interface_updates_no_existing_interface(
2178 self, mock_prepare_interface_updates, mock_get_interface_info
2179 ):
2180 vdur_vim_info_update = {
2181 "interfaces": [{"mac_address": mac1_addr, "compute_node": "host1"}]
2182 }
2183 interface_info = {
2184 "fixed_ips": [{"ip_address": ip1_addr}],
2185 "mac_address": mac2_addr,
2186 "network_id": net1_id,
2187 "status": "ACTIVE",
2188 }
2189 vnfr_update, vdur_update = {}, {}
2190 ports = {"ports": [port1, port2]}
2191 updated_sample_vim_info = deepcopy(sample_vim_info)
2192 updated_sample_vim_info["interfaces"] = []
2193 existing_vim_info = updated_sample_vim_info
2194 mock_get_interface_info.return_value = interface_info
2195 self.monitor.check_vm_interface_updates(
2196 server7,
2197 existing_vim_info,
2198 ports,
2199 vdur_vim_info_update,
2200 vdur_update,
2201 vdur_path,
2202 vnfr_update,
2203 vnfr_id,
2204 )
2205 check_if_assert_not_called(
2206 [mock_get_interface_info, mock_prepare_interface_updates]
2207 )
2208 self.assertNotIn("vim_message", vdur_vim_info_update)
2210 def test_update_in_database(self):
2211 all_updates = [{"some-key": "some-value"}, {"other-key": "other-value"}]
2212 self.monitor.update_in_database(all_updates, vnfr_id)
2213 self.assertEqual(self.monitor.db.set_list.call_count, 2)
2214 _call_mock_set_list = self.monitor.db.set_list.call_args_list
2215 self.assertEqual(
2216 _call_mock_set_list[0][0],
2217 ("vnfrs",),
2218 )
2219 self.assertEqual(
2220 _call_mock_set_list[0][1],
2221 (
2222 {
2223 "q_filter": {"_id": vnfr_id},
2224 "update_dict": {"some-key": "some-value"},
2225 }
2226 ),
2227 )
2228 self.assertEqual(
2229 _call_mock_set_list[1][0],
2230 ("vnfrs",),
2231 )
2232 self.assertEqual(
2233 _call_mock_set_list[1][1],
2234 (
2235 {
2236 "q_filter": {"_id": vnfr_id},
2237 "update_dict": {"other-key": "other-value"},
2238 }
2239 ),
2240 )
2242 def test_update_in_database_set_list_raises(self):
2243 all_updates = [{"some-key": "some-value"}, {"other-key": "other-value"}]
2244 self.monitor.db.set_list.side_effect = DbException("Connection failed.")
2245 with self.assertRaises(MonitorDbException) as err:
2246 self.monitor.update_in_database(all_updates, vnfr_id)
2247 self.assertEqual(
2248 str(err.exception.args[0]),
2249 "Error while updating differences in VNFR database exception Connection failed.",
2250 )
2251 self.assertEqual(self.monitor.db.set_list.call_count, 1)
2252 _call_mock_set_list = self.monitor.db.set_list.call_args_list
2253 self.assertEqual(
2254 _call_mock_set_list[0][0],
2255 ("vnfrs",),
2256 )
2257 self.assertEqual(
2258 _call_mock_set_list[0][1],
2259 (
2260 {
2261 "q_filter": {"_id": vnfr_id},
2262 "update_dict": {"some-key": "some-value"},
2263 }
2264 ),
2265 )
2267 def test_update_in_database_empty_all_updates(self):
2268 all_updates = []
2269 self.monitor.update_in_database(all_updates, vnfr_id)
2270 self.monitor.db.set_list.assert_not_called()
2272 @patch("osm_ng_ro.monitor.MonitorVms._get_vm_data_from_db")
2273 @patch("osm_ng_ro.monitor.MonitorVms.check_vm_status_updates")
2274 @patch("osm_ng_ro.monitor.MonitorVms.check_vm_interface_updates")
2275 @patch("osm_ng_ro.monitor.MonitorVms.backup_vdu_interfaces")
2276 @patch("osm_ng_ro.monitor.MonitorVms.update_in_database")
2277 def test_report_vdur_updates_no_change_in_vdur(
2278 self,
2279 mock_update_in_database,
2280 mock_backup_vdu_interfaces,
2281 mock_check_vm_interface_updates,
2282 mock_check_vm_status_updates,
2283 mock_get_vm_data_from_db,
2284 ):
2285 existing_vim_info = sample_vim_info
2286 vdur_vim_info_update = deepcopy(existing_vim_info)
2287 mock_get_vm_data_from_db.return_value = (
2288 vdur_path,
2289 vdur_vim_info_update,
2290 None,
2291 existing_vim_info,
2292 vnfr_id,
2293 vim_info_path,
2294 )
2295 ports = {"ports": [port1, port2]}
2296 self.monitor.report_vdur_updates(server7, sample_vm, ports)
2297 check_if_assert_not_called(
2298 [mock_update_in_database, mock_backup_vdu_interfaces]
2299 )
2300 mock_get_vm_data_from_db.assert_called_with(sample_vm)
2301 self.assertEqual(mock_get_vm_data_from_db.call_count, 1)
2302 mock_check_vm_status_updates.assert_called_once_with(
2303 vdur_vim_info_update, {}, server7, vdur_path
2304 )
2305 mock_check_vm_interface_updates.assert_called_once_with(
2306 server7,
2307 existing_vim_info,
2308 ports,
2309 vdur_vim_info_update,
2310 {},
2311 vdur_path,
2312 {},
2313 vnfr_id,
2314 )
2316 @patch("osm_ng_ro.monitor.MonitorVms._get_vm_data_from_db")
2317 @patch("osm_ng_ro.monitor.MonitorVms.check_vm_status_updates")
2318 @patch("osm_ng_ro.monitor.MonitorVms.check_vm_interface_updates")
2319 @patch("osm_ng_ro.monitor.MonitorVms.backup_vdu_interfaces")
2320 @patch("osm_ng_ro.monitor.MonitorVms.update_in_database")
2321 def test_report_vdur_updates_vdur_changed(
2322 self,
2323 mock_update_in_database,
2324 mock_backup_vdu_interfaces,
2325 mock_check_vm_interface_updates,
2326 mock_check_vm_status_updates,
2327 mock_get_vm_data_from_db,
2328 ):
2329 existing_vim_info = sample_vim_info
2330 vdur_vim_info_update = {
2331 "interfaces": [{"mac_address": mac1_addr, "compute_node": "host1"}]
2332 }
2333 mock_get_vm_data_from_db.return_value = (
2334 vdur_path,
2335 vdur_vim_info_update,
2336 None,
2337 existing_vim_info,
2338 vnfr_id,
2339 vim_info_path,
2340 )
2341 all_updates = [{}, {vim_info_path: vdur_vim_info_update}, {}]
2342 ports = {"ports": [port1, port2]}
2343 self.monitor.report_vdur_updates(server7, sample_vm, ports)
2344 mock_get_vm_data_from_db.assert_called_with(sample_vm)
2345 self.assertEqual(mock_get_vm_data_from_db.call_count, 1)
2346 mock_check_vm_status_updates.assert_called_once_with(
2347 vdur_vim_info_update, {}, server7, vdur_path
2348 )
2349 mock_check_vm_interface_updates.assert_called_once_with(
2350 server7,
2351 existing_vim_info,
2352 ports,
2353 vdur_vim_info_update,
2354 {},
2355 vdur_path,
2356 {},
2357 vnfr_id,
2358 )
2359 mock_backup_vdu_interfaces.assert_called_once_with(vdur_vim_info_update)
2360 mock_update_in_database.assert_called_once_with(all_updates, vnfr_id)
2362 @patch("osm_ng_ro.monitor.MonitorVms._get_vm_data_from_db")
2363 @patch("osm_ng_ro.monitor.MonitorVms.check_vm_status_updates")
2364 @patch("osm_ng_ro.monitor.MonitorVms.check_vm_interface_updates")
2365 @patch("osm_ng_ro.monitor.MonitorVms.backup_vdu_interfaces")
2366 @patch("osm_ng_ro.monitor.MonitorVms.update_in_database")
2367 def test_report_vdur_updates_check_vm_status_updates_raises(
2368 self,
2369 mock_update_in_database,
2370 mock_backup_vdu_interfaces,
2371 mock_check_vm_interface_updates,
2372 mock_check_vm_status_updates,
2373 mock_get_vm_data_from_db,
2374 ):
2375 existing_vim_info = sample_vim_info
2376 vdur_vim_info_update = {
2377 "interfaces": [{"mac_address": mac1_addr, "compute_node": "host1"}]
2378 }
2379 mock_get_vm_data_from_db.return_value = (
2380 vdur_path,
2381 vdur_vim_info_update,
2382 None,
2383 existing_vim_info,
2384 vnfr_id,
2385 vim_info_path,
2386 )
2387 ports = {"ports": [port1, port2]}
2388 mock_check_vm_status_updates.side_effect = yaml.YAMLError(
2389 "Cannot represent an object."
2390 )
2391 with self.assertRaises(yaml.YAMLError) as err:
2392 self.monitor.report_vdur_updates(server7, sample_vm, ports)
2393 self.assertEqual(str(err.exception), "Cannot represent an object.")
2394 mock_get_vm_data_from_db.assert_called_with(sample_vm)
2395 self.assertEqual(mock_get_vm_data_from_db.call_count, 1)
2396 mock_check_vm_status_updates.assert_called_once_with(
2397 vdur_vim_info_update, {}, server7, vdur_path
2398 )
2399 check_if_assert_not_called(
2400 [
2401 mock_check_vm_interface_updates,
2402 mock_backup_vdu_interfaces,
2403 mock_update_in_database,
2404 ]
2405 )
2407 @patch("osm_ng_ro.monitor.MonitorVms._get_vm_data_from_db")
2408 @patch("osm_ng_ro.monitor.MonitorVms.check_vm_status_updates")
2409 @patch("osm_ng_ro.monitor.MonitorVms.check_vm_interface_updates")
2410 @patch("osm_ng_ro.monitor.MonitorVms.backup_vdu_interfaces")
2411 @patch("osm_ng_ro.monitor.MonitorVms.update_in_database")
2412 def test_report_vdur_updates_database_update_raises(
2413 self,
2414 mock_update_in_database,
2415 mock_backup_vdu_interfaces,
2416 mock_check_vm_interface_updates,
2417 mock_check_vm_status_updates,
2418 mock_get_vm_data_from_db,
2419 ):
2420 existing_vim_info = sample_vim_info
2421 vdur_vim_info_update = {
2422 "interfaces": [{"mac_address": mac1_addr, "compute_node": "host1"}]
2423 }
2424 mock_get_vm_data_from_db.return_value = (
2425 vdur_path,
2426 vdur_vim_info_update,
2427 None,
2428 existing_vim_info,
2429 vnfr_id,
2430 vim_info_path,
2431 )
2432 all_updates = [{}, {vim_info_path: vdur_vim_info_update}, {}]
2433 ports = {"ports": [port1, port2]}
2434 mock_update_in_database.side_effect = MonitorDbException(
2435 f"Error while updating differences in VNFR {vnfr_id}."
2436 )
2437 with self.assertRaises(MonitorDbException) as err:
2438 self.monitor.report_vdur_updates(server7, sample_vm, ports)
2439 self.assertEqual(
2440 str(err.exception), f"Error while updating differences in VNFR {vnfr_id}."
2441 )
2442 mock_get_vm_data_from_db.assert_called_with(sample_vm)
2443 self.assertEqual(mock_get_vm_data_from_db.call_count, 1)
2444 mock_check_vm_status_updates.assert_called_once_with(
2445 vdur_vim_info_update, {}, server7, vdur_path
2446 )
2447 mock_check_vm_interface_updates.assert_called_once_with(
2448 server7,
2449 existing_vim_info,
2450 ports,
2451 vdur_vim_info_update,
2452 {},
2453 vdur_path,
2454 {},
2455 vnfr_id,
2456 )
2457 mock_backup_vdu_interfaces.assert_called_once_with(vdur_vim_info_update)
2458 mock_update_in_database.assert_called_once_with(all_updates, vnfr_id)
2460 @patch("osm_ng_ro.monitor.MonitorVms._get_vm_data_from_db")
2461 @patch("osm_ng_ro.monitor.MonitorVms.check_vm_status_updates")
2462 @patch("osm_ng_ro.monitor.MonitorVms.check_vm_interface_updates")
2463 @patch("osm_ng_ro.monitor.MonitorVms.backup_vdu_interfaces")
2464 @patch("osm_ng_ro.monitor.MonitorVms.update_in_database")
2465 def test_report_vdur_updates_no_vm_data(
2466 self,
2467 mock_update_in_database,
2468 mock_backup_vdu_interfaces,
2469 mock_check_vm_interface_updates,
2470 mock_check_vm_status_updates,
2471 mock_get_vm_data_from_db,
2472 ):
2473 mock_get_vm_data_from_db.return_value = None
2474 ports = {"ports": [port1, port2]}
2475 self.monitor.report_vdur_updates(server7, sample_vm, ports)
2476 check_if_assert_not_called(
2477 [
2478 mock_update_in_database,
2479 mock_backup_vdu_interfaces,
2480 mock_check_vm_interface_updates,
2481 mock_check_vm_status_updates,
2482 ]
2483 )
2484 mock_get_vm_data_from_db.assert_called_once_with(sample_vm)
2486 @patch("osm_ng_ro.monitor.MonitorVms.find_ro_tasks_to_monitor")
2487 @patch("osm_ng_ro.monitor.MonitorVms.prepare_vims_to_monitor")
2488 @patch("osm_ng_ro.monitor.MonitorVms.update_vnfrs")
2489 def test_run_no_db_vims(
2490 self,
2491 mock_update_vnfrs,
2492 mock_prepare_vims_to_monitor,
2493 mock_find_ro_tasks_to_monitor,
2494 ):
2495 self.monitor.db_vims = None
2496 self.monitor.run()
2497 check_if_assert_not_called(
2498 [
2499 mock_prepare_vims_to_monitor,
2500 mock_find_ro_tasks_to_monitor,
2501 mock_update_vnfrs,
2502 ]
2503 )
2505 @patch("osm_ng_ro.monitor.MonitorVms.find_ro_tasks_to_monitor")
2506 @patch("osm_ng_ro.monitor.MonitorVms.prepare_vims_to_monitor")
2507 @patch("osm_ng_ro.monitor.MonitorVms.update_vnfrs")
2508 def test_run_refresh_disabled(
2509 self,
2510 mock_update_vnfrs,
2511 mock_prepare_vims_to_monitor,
2512 mock_find_ro_tasks_to_monitor,
2513 ):
2514 self.monitor.db_vims = vims
2515 self.monitor.refresh_config.active = -1
2516 self.monitor.run()
2517 check_if_assert_not_called(
2518 [
2519 mock_prepare_vims_to_monitor,
2520 mock_find_ro_tasks_to_monitor,
2521 mock_update_vnfrs,
2522 ]
2523 )
2525 @patch("osm_ng_ro.monitor.MonitorVms.find_ro_tasks_to_monitor")
2526 @patch("osm_ng_ro.monitor.MonitorVms.prepare_vims_to_monitor")
2527 @patch("osm_ng_ro.monitor.MonitorVms.update_vnfrs")
2528 def test_run_no_proper_ro_task(
2529 self,
2530 mock_update_vnfrs,
2531 mock_prepare_vims_to_monitor,
2532 mock_find_ro_tasks_to_monitor,
2533 ):
2534 self.monitor.db_vims = vims
2535 self.monitor.refresh_config.active = 60
2536 mock_find_ro_tasks_to_monitor.return_value = []
2537 self.monitor.run()
2538 check_if_assert_not_called([mock_prepare_vims_to_monitor, mock_update_vnfrs])
2539 mock_find_ro_tasks_to_monitor.assert_called_once()
2541 @patch("osm_ng_ro.monitor.MonitorVms.find_ro_tasks_to_monitor")
2542 @patch("osm_ng_ro.monitor.MonitorVms.update_vnfrs")
2543 def test_run_with_proper_ro_task(
2544 self, mock_update_vnfrs, mock_find_ro_tasks_to_monitor
2545 ):
2546 self.monitor.db_vims = vims
2547 all_servers = [server1, server2]
2548 vim1_vms = [
2549 VmToMonitor(
2550 vm_id=vm1_id,
2551 target_record=target_record,
2552 )
2553 ]
2554 vim2_vms = [
2555 VmToMonitor(
2556 vm_id=vm2_id,
2557 target_record=target_record2,
2558 )
2559 ]
2560 all_ports = {"ports": [port1, port2]}
2561 mock_vim_connector = MagicMock()
2562 mock_vim_connector.get_monitoring_data.return_value = all_servers, all_ports
2563 self.monitor.my_vims = {
2564 vim1_id: mock_vim_connector,
2565 vim2_id: mock_vim_connector,
2566 vim3_id: mock_vim_connector,
2567 }
2568 self.monitor.refresh_config.active = 60
2569 mock_find_ro_tasks_to_monitor.return_value = [ro_task1, ro_task2]
2570 self.monitor.run()
2571 mock_find_ro_tasks_to_monitor.assert_called_once()
2572 _call_mock_update_vnfrs = mock_update_vnfrs.call_args_list
2573 self.assertEqual(mock_update_vnfrs.call_count, 2)
2574 self.assertEqual(
2575 _call_mock_update_vnfrs[0][0],
2576 (all_servers, all_ports, vim1_vms),
2577 )
2578 self.assertEqual(
2579 _call_mock_update_vnfrs[1][0],
2580 (all_servers, all_ports, vim2_vms),
2581 )
2582 self.assertEqual(mock_vim_connector.get_monitoring_data.call_count, 2)
2584 @patch("osm_ng_ro.monitor.MonitorVms.find_ro_tasks_to_monitor")
2585 @patch("osm_ng_ro.monitor.MonitorVms.update_vnfrs")
2586 def test_run_update_vnfrs_raises(
2587 self, mock_update_vnfrs, mock_find_ro_tasks_to_monitor
2588 ):
2589 self.monitor.db_vims = vims
2590 all_servers = [server1, server2]
2591 vim1_vms = [
2592 VmToMonitor(
2593 vm_id=vm1_id,
2594 target_record=target_record,
2595 )
2596 ]
2597 all_ports = {"ports": [port1, port2]}
2598 mock_vim_connector = MagicMock()
2599 mock_vim_connector.get_monitoring_data.return_value = all_servers, all_ports
2600 self.monitor.my_vims = {
2601 vim1_id: mock_vim_connector,
2602 vim2_id: mock_vim_connector,
2603 vim3_id: mock_vim_connector,
2604 }
2605 self.monitor.refresh_config.active = 60
2606 mock_find_ro_tasks_to_monitor.return_value = [ro_task1, ro_task2]
2607 mock_update_vnfrs.side_effect = DbException("DB is not active state.")
2608 with self.assertRaises(MonitorVmsException) as err:
2609 self.monitor.run()
2610 self.assertEqual(
2611 str(err.exception),
2612 "Exception while monitoring Openstack VMs: database exception DB is not active state.",
2613 )
2614 mock_find_ro_tasks_to_monitor.assert_called_once()
2615 _call_mock_update_vnfrs = mock_update_vnfrs.call_args_list
2616 self.assertEqual(mock_update_vnfrs.call_count, 1)
2617 self.assertEqual(
2618 _call_mock_update_vnfrs[0][0],
2619 (all_servers, all_ports, vim1_vms),
2620 )
2621 self.assertEqual(mock_vim_connector.get_monitoring_data.call_count, 1)
2623 @patch("osm_ng_ro.monitor.MonitorVms.prepare_vims_to_monitor")
2624 @patch("osm_ng_ro.monitor.MonitorVms.find_ro_tasks_to_monitor")
2625 @patch("osm_ng_ro.monitor.MonitorVms.update_vnfrs")
2626 def test_run_prepare_vims_to_monitor_raises(
2627 self,
2628 mock_update_vnfrs,
2629 mock_find_ro_tasks_to_monitor,
2630 mock_prepare_vims_to_monitor,
2631 ):
2632 self.monitor.db_vims = vims
2633 mock_vim_connector = MagicMock()
2634 self.monitor.my_vims = {
2635 vim1_id: mock_vim_connector,
2636 vim2_id: mock_vim_connector,
2637 vim3_id: mock_vim_connector,
2638 }
2639 self.monitor.refresh_config.active = 60
2640 mock_find_ro_tasks_to_monitor.return_value = [ro_task1, ro_task2]
2641 mock_prepare_vims_to_monitor.side_effect = KeyError("vim_id")
2642 with self.assertRaises(MonitorVmsException) as err:
2643 self.monitor.run()
2644 self.assertEqual(
2645 str(err.exception), "Exception while monitoring Openstack VMs: 'vim_id'"
2646 )
2647 mock_find_ro_tasks_to_monitor.assert_called_once()
2648 check_if_assert_not_called(
2649 [mock_update_vnfrs, mock_vim_connector.get_monitoring_data]
2650 )
2652 @patch("osm_ng_ro.monitor.monitoring_task")
2653 @patch("osm_ng_ro.monitor.threading.Timer")
2654 @patch("osm_ng_ro.monitor.MonitorVms")
2655 def test_start_monitoring(
2656 self, mock_monitor_vms, mock_threading_timer, mock_monitoring_task
2657 ):
2658 mock_monitor_vms.return_value.refresh_config.active = 20
2659 mock_threading_timer.return_value = mock_monitoring_task
2660 start_monitoring(config)
2661 mock_threading_timer.assert_called_once_with(
2662 20, start_monitoring, args=(config,)
2663 )
2664 mock_threading_timer.return_value = CopyingMock(threading.Timer)
2665 self.assertEqual(mock_threading_timer.call_count, 1)
2666 mock_monitor_vms.return_value.run.assert_called_once()
2667 mock_monitor_vms.assert_called_once_with(config)
2668 mock_monitoring_task.start.assert_called_once()
2670 @patch("osm_ng_ro.monitor.monitoring_task")
2671 @patch("osm_ng_ro.monitor.threading.Timer")
2672 @patch("osm_ng_ro.monitor.MonitorVms")
2673 def test_start_monitoring_empty_config(
2674 self, mock_monitor_vms, mock_threading_timer, mock_monitoring_task
2675 ):
2676 with self.assertRaises(MonitorVmsException) as err:
2677 start_monitoring(config={})
2678 self.assertEqual(
2679 str(err.exception),
2680 "Wrong configuration format is provided.",
2681 )
2682 check_if_assert_not_called(
2683 [mock_threading_timer, mock_monitor_vms, mock_monitoring_task]
2684 )
2686 @patch("osm_ng_ro.monitor.monitoring_task")
2687 @patch("osm_ng_ro.monitor.threading.Timer")
2688 @patch("osm_ng_ro.monitor.MonitorVms")
2689 def test_start_monitoring_monitor_vms_raises(
2690 self, mock_monitor_vms, mock_threading_timer, mock_monitoring_task
2691 ):
2692 mock_monitor_vms.side_effect = MonitorDbException("Can not connect to DB.")
2693 with self.assertRaises(MonitorDbException) as err:
2694 start_monitoring(config)
2695 self.assertEqual(str(err.exception), "Can not connect to DB.")
2696 mock_monitor_vms.assert_called_once_with(config)
2697 check_if_assert_not_called([mock_threading_timer, mock_monitoring_task])
2699 @patch("osm_ng_ro.monitor.monitoring_task")
2700 @patch("osm_ng_ro.monitor.threading.Timer")
2701 @patch("osm_ng_ro.monitor.MonitorVms")
2702 def test_start_monitoring_timer_thread_raises(
2703 self, mock_monitor_vms, mock_threading_timer, mock_monitoring_task
2704 ):
2705 mock_threading_timer.side_effect = RuntimeError(
2706 "cannot release un-acquired lock"
2707 )
2708 mock_monitor_vms.return_value.refresh_config.active = 2
2709 with self.assertRaises(RuntimeError) as err:
2710 start_monitoring(config)
2711 self.assertEqual(str(err.exception), "cannot release un-acquired lock")
2712 mock_monitor_vms.assert_called_once_with(config)
2713 mock_monitor_vms.return_value.run.assert_called_once()
2714 mock_threading_timer.assert_called_once_with(
2715 2, start_monitoring, args=(config,)
2716 )
2717 mock_monitoring_task.start.assert_not_called()
2719 @patch("osm_ng_ro.monitor.monitoring_task")
2720 def test_stop_monitoring(self, mock_monitoring_task):
2721 mock_monitoring_task.return_value = CopyingMock(threading.Timer)
2722 stop_monitoring()
2723 self.assertIsNotNone(mock_monitoring_task)
2724 mock_monitoring_task.cancel.assert_called_once()
2726 @patch("osm_ng_ro.monitor.monitoring_task")
2727 def test_stop_monitoring_no_task(self, mock_monitoring_task):
2728 mock_monitoring_task = CopyingMock(threading.Timer, return_value=None)
2729 stop_monitoring()
2730 mock_monitoring_task.cancel.assert_not_called()
2733if __name__ == "__main__":
2734 unittest.main()