diff --git a/test/test_celery_worker_global.py b/test/test_celery_worker_global.py index da36007024692deaed25f5f2853442e4dec64ce8..4baa87aac1b86525d33c5bc7a34408a6026a2179 100644 --- a/test/test_celery_worker_global.py +++ b/test/test_celery_worker_global.py @@ -40,6 +40,51 @@ def test_update_locations(mocker, mocked_worker_module, mocked_redis): assert len(list(_cached_locations())) > 0 +def test_update_locations(mocker, mocked_worker_module, mocked_redis): + + mocker.patch( + 'inventory_provider.db.opsdb.lookup_pop_info', + lambda c, h: [{'C': c, 'H': h}]) + + def _cached_locations(): + db = backend_db() + for k in db.keys(): + if k.startswith('opsdb:location:'): + yield k + + db = backend_db() + for k in list(_cached_locations()): + del db[k] + + assert len(list(_cached_locations())) == 0 # sanity + worker.update_equipment_locations() + assert len(list(_cached_locations())) > 0 + + +def test_access_services(mocker, mocked_worker_module, mocked_redis): + + opsdb_get_access_services = mocker.patch( + 'inventory_provider.db.opsdb.get_access_services') + opsdb_get_access_services.return_value = [ + {'name': 'some service name', 'a': 1, 'b': 2}, + {'name': 'another service name', 'c': 2, 'd': 3} + ] + + def _cached_locations(): + db = backend_db() + for k in db.keys(): + if k.startswith('opsdb:access_services:'): + yield k + + db = backend_db() + for k in list(_cached_locations()): + del db[k] + + assert len(list(_cached_locations())) == 0 # sanity + worker.update_access_services() + assert len(list(_cached_locations())) > 0 + + def test_InventoryTask_obj(data_config_filename): os.environ['INVENTORY_PROVIDER_CONFIG_FILENAME'] = data_config_filename task = worker.InventoryTask()