Page MenuHomeSoftware Heritage
Differential D968

Rewrite celery tasks as a decorated function
ClosedPublic
Actions

Authored by douardda on Jan 17 2019, 11:40 AM.

Details

Reviewers
vlorentz
Group Reviewers
Reviewers
Maniphest Tasks
T1480: Refactor tasks code as decorated functions using swh-scheduler 0.0.39
Commits
rDLDGab5fa4bbd684: Bump dependency on swh-scheduler 0.0.39
rDLDGc7f2cb11cd9a: Rewrite celery tasks as a decorated function
Summary

as well as the tests of the task. Requires swh-scheduler >= 0.0.39

Bump dependency on swh-scheduler 0.0.39

Related to T1480

Diff Detail

Repository
rDLDG Git loader
Lint
Automatic diff as part of commit; lint not applicable.
Unit
Automatic diff as part of commit; unit tests not applicable.

Revision Contents
Changeset List

PathSize
swh/
loader/
git/
62 lines
tests/
10 lines
87 lines

Diff 3080

View Options

swh/loader/git/tasks.py

# Copyright (C) 2015-2018 The Software Heritage developers # Copyright (C) 2015-2018 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution # See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version # License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information # See top-level LICENSE file for more information
import dateutil.parser import dateutil.parser
from swh.scheduler.task import Task from celery import current_app as app
from .from_disk import GitLoaderFromDisk, GitLoaderFromArchive from swh.loader.git.from_disk import GitLoaderFromDisk, GitLoaderFromArchive
from .loader import GitLoader from swh.loader.git.loader import GitLoader
# TODO: rename to LoadRemoteGitRepository @app.task(name=__name__ + '.UpdateGitRepository')
class UpdateGitRepository(Task): def update_git_repository(repo_url, base_url=None):
"""Import a git repository from a remote location""" """Import a git repository from a remote location"""
task_queue = 'swh_loader_git'
def run_task(self, repo_url, base_url=None):
"""Import a git repository"""
loader = GitLoader() loader = GitLoader()
loader.log = self.log
return loader.load(repo_url, base_url=base_url) return loader.load(repo_url, base_url=base_url)
class LoadDiskGitRepository(Task): @app.task(name=__name__ + '.LoadDiskGitRepository')
"""Import a git repository from disk""" def load_disk_git_repository(origin_url, directory, date):
task_queue = 'swh_loader_git_express' """Import a git repository from disk
def run_task(self, origin_url, directory, date):
"""Import a git repository, cloned in `directory` from `origin_url` at
`date`."""
loader = GitLoaderFromDisk() Import a git repository, cloned in `directory` from `origin_url` at
loader.log = self.log `date`.
"""
loader = GitLoaderFromDisk()
return loader.load(origin_url, directory, dateutil.parser.parse(date)) return loader.load(origin_url, directory, dateutil.parser.parse(date))
class UncompressAndLoadDiskGitRepository(Task): @app.task(name=__name__ + '.UncompressAndLoadDiskGitRepository')
"""Import a git repository from a zip archive""" def run_task(origin_url, archive_path, date):
task_queue = 'swh_loader_git_archive' """Import a git repository from a zip archive
def run_task(self, origin_url, archive_path, date): 1. Uncompress an archive repository in a local and temporary folder
"""1. Uncompress an archive repository in a local and temporary folder
2. Load it through the git disk loader 2. Load it through the git disk loader
3. Clean up the temporary folder 3. Clean up the temporary folder
""" """
loader = GitLoaderFromArchive() loader = GitLoaderFromArchive()
loader.log = self.log
return loader.load( return loader.load(
origin_url, archive_path, dateutil.parser.parse(date)) origin_url, archive_path, dateutil.parser.parse(date))
View Options

swh/loader/git/tests/conftest.py

  • This file was added.
import pytest
from swh.scheduler.tests.conftest import * # noqa
@pytest.fixture(scope='session')
def celery_includes():
return [
'swh.loader.git.tasks',
]
View Options

swh/loader/git/tests/test_tasks.py

# Copyright (C) 2018 The Software Heritage developers # Copyright (C) 2018 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution # See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version # License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information # See top-level LICENSE file for more information
import datetime import datetime
import unittest
from unittest.mock import patch from unittest.mock import patch
from swh.loader.git.tasks import (
UpdateGitRepository, LoadDiskGitRepository,
UncompressAndLoadDiskGitRepository
)
class TestTasks(unittest.TestCase):
def test_check_task_name(self):
task = UpdateGitRepository()
self.assertEqual(task.task_queue, 'swh_loader_git')
@patch('swh.loader.git.loader.GitLoader.load') @patch('swh.loader.git.loader.GitLoader.load')
def test_task(self, mock_loader): def test_git_loader(mock_loader, swh_app, celery_session_worker):
mock_loader.return_value = {'status': 'eventful'} mock_loader.return_value = {'status': 'eventful'}
task = UpdateGitRepository()
# given res = swh_app.send_task(
actual_result = task.run_task('origin_url') 'swh.loader.git.tasks.UpdateGitRepository',
('origin_url',))
self.assertEqual(actual_result, {'status': 'eventful'}) assert res
res.wait()
assert res.successful()
assert res.result == {'status': 'eventful'}
mock_loader.assert_called_once_with('origin_url', base_url=None) mock_loader.assert_called_once_with('origin_url', base_url=None)
class TestTasks2(unittest.TestCase):
def test_check_task_name(self):
task = LoadDiskGitRepository()
self.assertEqual(task.task_queue, 'swh_loader_git_express')
@patch('swh.loader.git.from_disk.GitLoaderFromDisk.load') @patch('swh.loader.git.from_disk.GitLoaderFromDisk.load')
def test_task(self, mock_loader): def test_git_loader_from_disk(mock_loader, swh_app, celery_session_worker):
mock_loader.return_value = {'status': 'uneventful'} mock_loader.return_value = {'status': 'uneventful'}
task = LoadDiskGitRepository()
# given res = swh_app.send_task(
actual_result = task.run_task('origin_url2', '/some/repo', 'swh.loader.git.tasks.LoadDiskGitRepository',
'2018-12-10 00:00') ('origin_url2', '/some/repo', '2018-12-10 00:00'))
assert res
self.assertEqual(actual_result, {'status': 'uneventful'}) res.wait()
assert res.successful()
assert res.result == {'status': 'uneventful'}
mock_loader.assert_called_once_with( mock_loader.assert_called_once_with(
'origin_url2', '/some/repo', datetime.datetime(2018, 12, 10, 0, 0)) 'origin_url2', '/some/repo', datetime.datetime(2018, 12, 10, 0, 0))
class TestTasks3(unittest.TestCase):
def test_check_task_name(self):
task = UncompressAndLoadDiskGitRepository()
self.assertEqual(task.task_queue, 'swh_loader_git_archive')
@patch('swh.loader.git.from_disk.GitLoaderFromArchive.load') @patch('swh.loader.git.from_disk.GitLoaderFromArchive.load')
def test_task(self, mock_loader): def test_git_loader_from_archive(mock_loader, swh_app, celery_session_worker):
mock_loader.return_value = {'status': 'failed'} mock_loader.return_value = {'status': 'failed'}
task = UncompressAndLoadDiskGitRepository()
# given
actual_result = task.run_task('origin_url3', '/path/repo',
'2017-01-10 00:00')
self.assertEqual(actual_result, {'status': 'failed'}) res = swh_app.send_task(
'swh.loader.git.tasks.UncompressAndLoadDiskGitRepository',
('origin_url3', '/some/repo', '2017-01-10 00:00'))
assert res
res.wait()
assert res.successful()
assert res.result == {'status': 'failed'}
mock_loader.assert_called_once_with( mock_loader.assert_called_once_with(
'origin_url3', '/path/repo', datetime.datetime(2017, 1, 10, 0, 0)) 'origin_url3', '/some/repo', datetime.datetime(2017, 1, 10, 0, 0))
About · Developer information · Legal