""" just checks that the worker methods call the right functions and some data ends up in the right place ... otherwise not very detailed """ from inventory_provider.tasks import worker from inventory_provider.tasks.common import _get_redis def backend_db(): return _get_redis({ 'redis': { 'hostname': None, 'port': None }, 'redis-databases': [0, 7] }).db def test_netconf_refresh_config(mocked_worker_module, router): del backend_db()['netconf:' + router] worker.netconf_refresh_config(router) assert backend_db()['netconf:' + router] def test_snmp_refresh_interfaces(mocked_worker_module, router): del backend_db()['snmp-interfaces:' + router] worker.snmp_refresh_interfaces(router, 'fake-community') assert backend_db()['snmp-interfaces:' + router] def test_reload_router_config(mocked_worker_module, router, mocker): saved_data = {} for key in ('netconf:' + router, 'snmp-interfaces:' + router): saved_data[key] = backend_db().pop(key) assert 'netconf:' + router not in backend_db() assert 'snmp-interfaces:' + router not in backend_db() def _mocked_netconf_refresh_config_apply(args): key = 'netconf:' + args[0] backend_db()[key] = saved_data[key] mocker.patch( 'inventory_provider.tasks.worker.netconf_refresh_config.apply', _mocked_netconf_refresh_config_apply) def _mocked_snmp_refresh_interfaces_apply(args): assert len(args) == 2 key = 'snmp-interfaces:' + args[0] backend_db()[key] = saved_data[key] mocker.patch( 'inventory_provider.tasks.worker.snmp_refresh_interfaces.apply', _mocked_snmp_refresh_interfaces_apply) def _mocked_update_status(self, **kwargs): pass mocker.patch( 'inventory_provider.tasks.worker.InventoryTask.update_state', _mocked_update_status) worker.reload_router_config(router) assert 'netconf:' + router in backend_db() assert 'snmp-interfaces:' + router in backend_db()