Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/git/tests/test_tasks.py
| Show All 12 Lines | |||||
| ) | ) | ||||
| class TestTasks(unittest.TestCase): | class TestTasks(unittest.TestCase): | ||||
| def test_check_task_name(self): | def test_check_task_name(self): | ||||
| task = UpdateGitRepository() | task = UpdateGitRepository() | ||||
| self.assertEqual(task.task_queue, 'swh_loader_git') | self.assertEqual(task.task_queue, 'swh_loader_git') | ||||
| @patch('swh.loader.git.updater.BulkUpdater.load') | @patch('swh.loader.git.loader.GitLoader.load') | ||||
| def test_task(self, mock_loader): | def test_task(self, mock_loader): | ||||
| mock_loader.return_value = {'status': 'eventful'} | mock_loader.return_value = {'status': 'eventful'} | ||||
| task = UpdateGitRepository() | task = UpdateGitRepository() | ||||
| # given | # given | ||||
| actual_result = task.run_task('origin_url') | actual_result = task.run_task('origin_url') | ||||
| self.assertEqual(actual_result, {'status': 'eventful'}) | self.assertEqual(actual_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): | class TestTasks2(unittest.TestCase): | ||||
| def test_check_task_name(self): | def test_check_task_name(self): | ||||
| task = LoadDiskGitRepository() | task = LoadDiskGitRepository() | ||||
| self.assertEqual(task.task_queue, 'swh_loader_git_express') | self.assertEqual(task.task_queue, 'swh_loader_git_express') | ||||
| @patch('swh.loader.git.loader.GitLoader.load') | @patch('swh.loader.git.from_disk.GitLoaderFromDisk.load') | ||||
| def test_task(self, mock_loader): | def test_task(self, mock_loader): | ||||
| mock_loader.return_value = {'status': 'uneventful'} | mock_loader.return_value = {'status': 'uneventful'} | ||||
| task = LoadDiskGitRepository() | task = LoadDiskGitRepository() | ||||
| # given | # given | ||||
| actual_result = task.run_task('origin_url2', '/some/repo', | actual_result = task.run_task('origin_url2', '/some/repo', | ||||
| '2018-12-10 00:00') | '2018-12-10 00:00') | ||||
| self.assertEqual(actual_result, {'status': 'uneventful'}) | self.assertEqual(actual_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): | class TestTasks3(unittest.TestCase): | ||||
| def test_check_task_name(self): | def test_check_task_name(self): | ||||
| task = UncompressAndLoadDiskGitRepository() | task = UncompressAndLoadDiskGitRepository() | ||||
| self.assertEqual(task.task_queue, 'swh_loader_git_archive') | self.assertEqual(task.task_queue, 'swh_loader_git_archive') | ||||
| @patch('swh.loader.git.loader.GitLoaderFromArchive.load') | @patch('swh.loader.git.from_disk.GitLoaderFromArchive.load') | ||||
| def test_task(self, mock_loader): | def test_task(self, mock_loader): | ||||
| mock_loader.return_value = {'status': 'failed'} | mock_loader.return_value = {'status': 'failed'} | ||||
| task = UncompressAndLoadDiskGitRepository() | task = UncompressAndLoadDiskGitRepository() | ||||
| # given | # given | ||||
| actual_result = task.run_task('origin_url3', '/path/repo', | actual_result = task.run_task('origin_url3', '/path/repo', | ||||
| '2017-01-10 00:00') | '2017-01-10 00:00') | ||||
| self.assertEqual(actual_result, {'status': 'failed'}) | self.assertEqual(actual_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', '/path/repo', datetime.datetime(2017, 1, 10, 0, 0)) | ||||