diff --git a/swh/lister/core/tests/conftest.py b/swh/lister/core/tests/conftest.py index 07ef181..f4a0e99 100644 --- a/swh/lister/core/tests/conftest.py +++ b/swh/lister/core/tests/conftest.py @@ -1,24 +1,25 @@ import pytest @pytest.fixture(scope='session') def celery_enable_logging(): return True @pytest.fixture(scope='session') def celery_includes(): return [ 'swh.lister.bitbucket.tasks', + 'swh.lister.debian.tasks', 'swh.lister.github.tasks', ] # override the celery_session_app fixture to monkeypatch the 'main' # swh.scheduler.celery_backend.config.app Celery application # with the test application. @pytest.fixture(scope='session') def swh_app(celery_session_app): import swh.scheduler.celery_backend.config swh.scheduler.celery_backend.config.app = celery_session_app yield celery_session_app diff --git a/swh/lister/debian/tests/__init__.py b/swh/lister/debian/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/swh/lister/debian/tests/conftest.py b/swh/lister/debian/tests/conftest.py new file mode 100644 index 0000000..507fef9 --- /dev/null +++ b/swh/lister/debian/tests/conftest.py @@ -0,0 +1 @@ +from swh.lister.core.tests.conftest import * # noqa diff --git a/swh/lister/debian/tests/test_tasks.py b/swh/lister/debian/tests/test_tasks.py new file mode 100644 index 0000000..bd1d0b4 --- /dev/null +++ b/swh/lister/debian/tests/test_tasks.py @@ -0,0 +1,29 @@ +from time import sleep +from celery.result import GroupResult + +from unittest.mock import patch + + +def test_ping(swh_app, celery_session_worker): + res = swh_app.send_task( + 'swh.lister.debian.tasks.ping') + assert res + res.wait() + assert res.successful() + assert res.result == 'OK' + + +@patch('swh.lister.debian.tasks.DebianLister') +def test_lister(lister, swh_app, celery_session_worker): + # setup the mocked DebianLister + lister.return_value = lister + lister.run.return_value = None + + res = swh_app.send_task( + 'swh.lister.debian.tasks.DebianListerTask', ('stretch',)) + assert res + res.wait() + assert res.successful() + + lister.assert_called_once_with() + lister.run.assert_called_once_with('stretch')