Changeset View
Changeset View
Standalone View
Standalone View
swh/lister/npm/tests/test_tasks.py
# Copyright (C) 2019-2020 The Software Heritage developers | |||||
# See the AUTHORS file at the top-level directory of this distribution | |||||
# License: GNU General Public License version 3, or any later version | |||||
# See top-level LICENSE file for more information | |||||
from contextlib import contextmanager | from contextlib import contextmanager | ||||
from unittest.mock import patch | from unittest.mock import patch | ||||
@contextmanager | @contextmanager | ||||
def mock_save(lister): | def mock_save(lister): | ||||
yield | yield | ||||
def test_ping(swh_app, celery_session_worker): | def test_ping(swh_scheduler_celery_app, swh_scheduler_celery_worker): | ||||
res = swh_app.send_task("swh.lister.npm.tasks.ping") | res = swh_scheduler_celery_app.send_task("swh.lister.npm.tasks.ping") | ||||
assert res | assert res | ||||
res.wait() | res.wait() | ||||
assert res.successful() | assert res.successful() | ||||
assert res.result == "OK" | assert res.result == "OK" | ||||
@patch("swh.lister.npm.tasks.save_registry_state") | @patch("swh.lister.npm.tasks.save_registry_state") | ||||
@patch("swh.lister.npm.tasks.NpmLister") | @patch("swh.lister.npm.tasks.NpmLister") | ||||
def test_lister(lister, save, swh_app, celery_session_worker): | def test_lister(lister, save, swh_scheduler_celery_app, swh_scheduler_celery_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 | ||||
save.side_effect = mock_save | save.side_effect = mock_save | ||||
res = swh_app.send_task("swh.lister.npm.tasks.NpmListerTask") | res = swh_scheduler_celery_app.send_task("swh.lister.npm.tasks.NpmListerTask") | ||||
assert res | assert res | ||||
res.wait() | res.wait() | ||||
assert res.successful() | assert res.successful() | ||||
lister.assert_called_once_with() | lister.assert_called_once_with() | ||||
lister.run.assert_called_once_with() | lister.run.assert_called_once_with() | ||||
@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_scheduler_celery_app, swh_scheduler_celery_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 | ||||
seq.return_value = 42 | seq.return_value = 42 | ||||
save.side_effect = mock_save | save.side_effect = mock_save | ||||
res = swh_app.send_task("swh.lister.npm.tasks.NpmIncrementalListerTask") | res = swh_scheduler_celery_app.send_task( | ||||
"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) |