Changeset View
Changeset View
Standalone View
Standalone View
swh/lister/npm/tests/test_tasks.py
| Show All 34 Lines | |||||
| @patch('swh.lister.npm.tasks.save_registry_state') | @patch('swh.lister.npm.tasks.save_registry_state') | ||||
| @patch('swh.lister.npm.tasks.get_last_update_seq') | @patch('swh.lister.npm.tasks.get_last_update_seq') | ||||
| @patch('swh.lister.npm.tasks.NpmIncrementalLister') | @patch('swh.lister.npm.tasks.NpmIncrementalLister') | ||||
| def test_incremental(lister, seq, save, swh_app, celery_session_worker): | def test_incremental(lister, seq, save, swh_app, celery_session_worker): | ||||
| # setup the mocked NpmLister | # setup the mocked NpmLister | ||||
| lister.return_value = lister | lister.return_value = lister | ||||
| lister.run.return_value = None | lister.run.return_value = None | ||||
| lister.request_headers.return_value = [] | |||||
| seq.return_value = 42 | seq.return_value = 42 | ||||
| save.side_effect = mock_save | save.side_effect = mock_save | ||||
| res = swh_app.send_task( | res = swh_app.send_task( | ||||
| 'swh.lister.npm.tasks.NpmIncrementalListerTask') | 'swh.lister.npm.tasks.NpmIncrementalListerTask') | ||||
| assert res | assert res | ||||
| res.wait() | res.wait() | ||||
| assert res.successful() | assert res.successful() | ||||
| lister.assert_called_once_with() | lister.assert_called_once_with() | ||||
| seq.assert_called_once_with(lister) | seq.assert_called_once_with(lister) | ||||
| lister.run.assert_called_once_with(min_bound=42) | lister.run.assert_called_once_with(min_bound=42) | ||||