diff --git a/swh/lister/gitea/tasks.py b/swh/lister/gitea/tasks.py index f6db2ed..4c70833 100644 --- a/swh/lister/gitea/tasks.py +++ b/swh/lister/gitea/tasks.py @@ -1,28 +1,21 @@ -# Copyright (C) 2020 the Software Heritage developers +# Copyright (C) 2020-2022 the Software Heritage developers # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information -from typing import Dict, Optional +from typing import Dict from celery import shared_task from .lister import GiteaLister @shared_task(name=__name__ + ".FullGiteaRelister") -def list_gitea_full( - url: str, - instance: Optional[str] = None, - api_token: Optional[str] = None, - page_size: Optional[int] = None, -) -> Dict[str, int]: +def list_gitea_full(**lister_args) -> Dict[str, int]: """Full update of a Gitea instance""" - lister = GiteaLister.from_configfile( - url=url, instance=instance, api_token=api_token, page_size=page_size - ) + lister = GiteaLister.from_configfile(**lister_args) return lister.run().dict() @shared_task(name=__name__ + ".ping") def _ping() -> str: return "OK" diff --git a/swh/lister/gitea/tests/test_tasks.py b/swh/lister/gitea/tests/test_tasks.py index a204cb1..f418e41 100644 --- a/swh/lister/gitea/tests/test_tasks.py +++ b/swh/lister/gitea/tests/test_tasks.py @@ -1,61 +1,59 @@ # Copyright (C) 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 unittest.mock import patch from swh.lister.pattern import ListerStats def test_ping(swh_scheduler_celery_app, swh_scheduler_celery_worker): res = swh_scheduler_celery_app.send_task("swh.lister.gitea.tasks.ping") assert res res.wait() assert res.successful() assert res.result == "OK" @patch("swh.lister.gitea.tasks.GiteaLister") def test_full_listing(lister, swh_scheduler_celery_app, swh_scheduler_celery_worker): lister.from_configfile.return_value = lister lister.run.return_value = ListerStats(pages=10, origins=500) kwargs = dict(url="https://try.gitea.io/api/v1") res = swh_scheduler_celery_app.send_task( "swh.lister.gitea.tasks.FullGiteaRelister", kwargs=kwargs, ) assert res res.wait() assert res.successful() - actual_kwargs = dict(**kwargs, instance=None, api_token=None, page_size=None) - - lister.from_configfile.assert_called_once_with(**actual_kwargs) + lister.from_configfile.assert_called_once_with(**kwargs) lister.run.assert_called_once_with() @patch("swh.lister.gitea.tasks.GiteaLister") def test_full_listing_params( lister, swh_scheduler_celery_app, swh_scheduler_celery_worker ): lister.from_configfile.return_value = lister lister.run.return_value = ListerStats(pages=10, origins=500) kwargs = dict( url="https://0xacab.org/api/v4", instance="0xacab", api_token="test", page_size=50, ) res = swh_scheduler_celery_app.send_task( "swh.lister.gitea.tasks.FullGiteaRelister", kwargs=kwargs, ) assert res res.wait() assert res.successful() lister.from_configfile.assert_called_once_with(**kwargs) lister.run.assert_called_once_with() diff --git a/swh/lister/gogs/tasks.py b/swh/lister/gogs/tasks.py index 81287fa..a3ad438 100644 --- a/swh/lister/gogs/tasks.py +++ b/swh/lister/gogs/tasks.py @@ -1,28 +1,21 @@ # Copyright (C) 2022 the Software Heritage developers # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information -from typing import Dict, Optional +from typing import Dict from celery import shared_task from .lister import GogsLister @shared_task(name=__name__ + ".FullGogsRelister") -def list_gogs_full( - url: str, - instance: Optional[str] = None, - api_token: Optional[str] = None, - page_size: Optional[int] = None, -) -> Dict[str, int]: +def list_gogs_full(**lister_args) -> Dict[str, int]: """Full update of a Gogs instance""" - lister = GogsLister.from_configfile( - url=url, instance=instance, api_token=api_token, page_size=page_size - ) + lister = GogsLister.from_configfile(**lister_args) return lister.run().dict() @shared_task(name=__name__ + ".ping") def _ping() -> str: return "OK" diff --git a/swh/lister/gogs/tests/test_tasks.py b/swh/lister/gogs/tests/test_tasks.py index 2f38341..9698fc3 100644 --- a/swh/lister/gogs/tests/test_tasks.py +++ b/swh/lister/gogs/tests/test_tasks.py @@ -1,61 +1,59 @@ # Copyright (C) 2022 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 unittest.mock import patch from swh.lister.pattern import ListerStats def test_ping(swh_scheduler_celery_app, swh_scheduler_celery_worker): res = swh_scheduler_celery_app.send_task("swh.lister.gogs.tasks.ping") assert res res.wait() assert res.successful() assert res.result == "OK" @patch("swh.lister.gogs.tasks.GogsLister") def test_full_listing(lister, swh_scheduler_celery_app, swh_scheduler_celery_worker): lister.from_configfile.return_value = lister lister.run.return_value = ListerStats(pages=10, origins=500) kwargs = dict(url="https://try.gogs.io/api/v1/") res = swh_scheduler_celery_app.send_task( "swh.lister.gogs.tasks.FullGogsRelister", kwargs=kwargs, ) assert res res.wait() assert res.successful() - actual_kwargs = dict(**kwargs, instance=None, api_token=None, page_size=None) - - lister.from_configfile.assert_called_once_with(**actual_kwargs) + lister.from_configfile.assert_called_once_with(**kwargs) lister.run.assert_called_once_with() @patch("swh.lister.gogs.tasks.GogsLister") def test_full_listing_params( lister, swh_scheduler_celery_app, swh_scheduler_celery_worker ): lister.from_configfile.return_value = lister lister.run.return_value = ListerStats(pages=10, origins=500) kwargs = dict( url="https://gogs-host.com/api/v1/", instance="foo", api_token="test", page_size=50, ) res = swh_scheduler_celery_app.send_task( "swh.lister.gogs.tasks.FullGogsRelister", kwargs=kwargs, ) assert res res.wait() assert res.successful() lister.from_configfile.assert_called_once_with(**kwargs) lister.run.assert_called_once_with()