reonokiy/swh-lister
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

gogs, gitea: Fix task execution to pass along extra kwargs

Browse source 
Related to https://gitlab.softwareheritage.org/infra/sysadm-environment/-/issues/4684
This commit is contained in:
Antoine R. Dumont (@ardumont) 2022-12-14 16:09:56 +01:00
parent e785e67315
commit b3b5639e9a
No known key found for this signature in database
GPG key ID: 52E2E9840D10C3B8
4 changed files with 9 additions and 27 deletions
Download patch file Download diff file Expand all files Collapse all files

15
swh/lister/gitea/tasks.py
View file

@ -1,8 +1,8 @@
# 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
@ -10,16 +10,9 @@ 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()

4
swh/lister/gitea/tests/test_tasks.py
View file

@ -30,9 +30,7 @@ def test_full_listing(lister, swh_scheduler_celery_app, swh_scheduler_celery_wor
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()

13
swh/lister/gogs/tasks.py
View file

@ -2,7 +2,7 @@
# 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
@ -10,16 +10,9 @@ 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()

4
swh/lister/gogs/tests/test_tasks.py
View file

@ -30,9 +30,7 @@ def test_full_listing(lister, swh_scheduler_celery_app, swh_scheduler_celery_wor
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()