diff --git a/swh/web/tests/admin/test_origin_save.py b/swh/web/tests/admin/test_origin_save.py index 0338c3c0..b0e20d04 100644 --- a/swh/web/tests/admin/test_origin_save.py +++ b/swh/web/tests/admin/test_origin_save.py @@ -1,232 +1,232 @@ # Copyright (C) 2015-2018 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU Affero General Public License version 3, or any later version # See top-level LICENSE file for more information from urllib.parse import unquote from django.contrib.auth import get_user_model from unittest.mock import patch from swh.web.common.models import ( SaveAuthorizedOrigin, SaveUnauthorizedOrigin, SaveOriginRequest ) from swh.web.common.origin_save import can_save_origin from swh.web.common.models import ( SAVE_REQUEST_PENDING, SAVE_REQUEST_ACCEPTED, SAVE_REQUEST_REJECTED, SAVE_TASK_NOT_YET_SCHEDULED ) from swh.web.common.utils import reverse from swh.web.tests.testcase import WebTestCase _user_name = 'swh-web-admin' _user_mail = 'admin@swh-web.org' _user_password = '..34~pounds~BEAUTY~march~63..' _authorized_origin_url = 'https://scm.ourproject.org/anonscm/' _unauthorized_origin_url = 'https://www.softwareheritage.org/' class OriginSaveAdminTestCase(WebTestCase): @classmethod def setUpTestData(cls): # noqa: N802 User = get_user_model() # noqa: N806 user = User.objects.create_user(_user_name, _user_mail, _user_password) user.is_staff = True user.save() SaveAuthorizedOrigin.objects.create(url=_authorized_origin_url) SaveUnauthorizedOrigin.objects.create(url=_unauthorized_origin_url) def check_not_login(self, url): login_url = reverse('login', query_params={'next': url}) response = self.client.post(url) self.assertEqual(response.status_code, 302) self.assertEqual(unquote(response.url), login_url) def test_add_authorized_origin_url(self): authorized_url = 'https://scm.adullact.net/anonscm/' self.assertEqual(can_save_origin(authorized_url), SAVE_REQUEST_PENDING) url = reverse('admin-origin-save-add-authorized-url', url_args={'origin_url': authorized_url}) self.check_not_login(url) self.assertEqual(can_save_origin(authorized_url), SAVE_REQUEST_PENDING) self.client.login(username=_user_name, password=_user_password) response = self.client.post(url) self.assertEqual(response.status_code, 200) self.assertEqual(can_save_origin(authorized_url), SAVE_REQUEST_ACCEPTED) def test_remove_authorized_origin_url(self): self.assertEqual(can_save_origin(_authorized_origin_url), SAVE_REQUEST_ACCEPTED) url = reverse('admin-origin-save-remove-authorized-url', url_args={'origin_url': _authorized_origin_url}) self.check_not_login(url) self.assertEqual(can_save_origin(_authorized_origin_url), SAVE_REQUEST_ACCEPTED) self.client.login(username=_user_name, password=_user_password) response = self.client.post(url) self.assertEqual(response.status_code, 200) self.assertEqual(can_save_origin(_authorized_origin_url), SAVE_REQUEST_PENDING) def test_add_unauthorized_origin_url(self): unauthorized_url = 'https://www.yahoo./' self.assertEqual(can_save_origin(unauthorized_url), SAVE_REQUEST_PENDING) url = reverse('admin-origin-save-add-unauthorized-url', url_args={'origin_url': unauthorized_url}) self.check_not_login(url) self.assertEqual(can_save_origin(unauthorized_url), SAVE_REQUEST_PENDING) self.client.login(username=_user_name, password=_user_password) response = self.client.post(url) self.assertEqual(response.status_code, 200) self.assertEqual(can_save_origin(unauthorized_url), SAVE_REQUEST_REJECTED) def test_remove_unauthorized_origin_url(self): self.assertEqual(can_save_origin(_unauthorized_origin_url), SAVE_REQUEST_REJECTED) url = reverse('admin-origin-save-remove-unauthorized-url', url_args={'origin_url': _unauthorized_origin_url}) self.check_not_login(url) self.assertEqual(can_save_origin(_unauthorized_origin_url), SAVE_REQUEST_REJECTED) self.client.login(username=_user_name, password=_user_password) response = self.client.post(url) self.assertEqual(response.status_code, 200) self.assertEqual(can_save_origin(_unauthorized_origin_url), SAVE_REQUEST_PENDING) @patch('swh.web.common.origin_save.scheduler') def test_accept_pending_save_request(self, mock_scheduler): origin_type = 'git' origin_url = 'https://v2.pikacode.com/bthate/botlib.git' save_request_url = reverse('api-save-origin', url_args={'origin_type': origin_type, 'origin_url': origin_url}) response = self.client.post(save_request_url, data={}, content_type='application/x-www-form-urlencoded') # noqa self.assertEqual(response.status_code, 200) self.assertEqual(response.data['save_request_status'], SAVE_REQUEST_PENDING) accept_request_url = reverse('admin-origin-save-request-accept', url_args={'origin_type': origin_type, 'origin_url': origin_url}) self.check_not_login(accept_request_url) tasks_data = [ { 'priority': 'high', 'policy': 'oneshot', - 'type': 'origin-update-git', + 'type': 'load-git', 'arguments': { 'kwargs': { 'repo_url': origin_url }, 'args': [] }, 'status': 'next_run_not_scheduled', 'id': 1, } ] mock_scheduler.create_tasks.return_value = tasks_data mock_scheduler.get_tasks.return_value = tasks_data self.client.login(username=_user_name, password=_user_password) response = self.client.post(accept_request_url) self.assertEqual(response.status_code, 200) response = self.client.get(save_request_url) self.assertEqual(response.status_code, 200) self.assertEqual(response.data[0]['save_request_status'], SAVE_REQUEST_ACCEPTED) self.assertEqual(response.data[0]['save_task_status'], SAVE_TASK_NOT_YET_SCHEDULED) @patch('swh.web.common.origin_save.scheduler') def test_reject_pending_save_request(self, mock_scheduler): origin_type = 'git' origin_url = 'https://wikipedia.com' save_request_url = reverse('api-save-origin', url_args={'origin_type': origin_type, 'origin_url': origin_url}) response = self.client.post(save_request_url, data={}, content_type='application/x-www-form-urlencoded') # noqa self.assertEqual(response.status_code, 200) self.assertEqual(response.data['save_request_status'], SAVE_REQUEST_PENDING) reject_request_url = reverse('admin-origin-save-request-reject', url_args={'origin_type': origin_type, 'origin_url': origin_url}) self.check_not_login(reject_request_url) self.client.login(username=_user_name, password=_user_password) response = self.client.post(reject_request_url) self.assertEqual(response.status_code, 200) tasks_data = [ { 'priority': 'high', 'policy': 'oneshot', - 'type': 'origin-update-git', + 'type': 'load-git', 'arguments': { 'kwargs': { 'repo_url': origin_url }, 'args': [] }, 'status': 'next_run_not_scheduled', 'id': 1, } ] mock_scheduler.create_tasks.return_value = tasks_data mock_scheduler.get_tasks.return_value = tasks_data response = self.client.get(save_request_url) self.assertEqual(response.status_code, 200) self.assertEqual(response.data[0]['save_request_status'], SAVE_REQUEST_REJECTED) def test_remove_save_request(self): sor = SaveOriginRequest.objects.create(origin_type='git', origin_url='https://wikipedia.com', # noqa status=SAVE_REQUEST_PENDING) self.assertEqual(SaveOriginRequest.objects.count(), 1) remove_request_url = reverse('admin-origin-save-request-remove', url_args={'sor_id': sor.id}) self.check_not_login(remove_request_url) self.client.login(username=_user_name, password=_user_password) response = self.client.post(remove_request_url) self.assertEqual(response.status_code, 200) self.assertEqual(SaveOriginRequest.objects.count(), 0) diff --git a/swh/web/tests/api/views/test_origin_save.py b/swh/web/tests/api/views/test_origin_save.py index fd9a306c..c210dbf5 100644 --- a/swh/web/tests/api/views/test_origin_save.py +++ b/swh/web/tests/api/views/test_origin_save.py @@ -1,261 +1,261 @@ # Copyright (C) 2018 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU Affero General Public License version 3, or any later version # See top-level LICENSE file for more information from datetime import datetime, timedelta from django.utils import timezone from rest_framework.test import APITestCase from unittest.mock import patch from swh.web.common.utils import reverse from swh.web.common.models import ( SaveUnauthorizedOrigin, SaveOriginRequest, SAVE_REQUEST_ACCEPTED, SAVE_REQUEST_REJECTED, SAVE_REQUEST_PENDING ) from swh.web.common.models import ( SAVE_TASK_NOT_CREATED, SAVE_TASK_NOT_YET_SCHEDULED, SAVE_TASK_SCHEDULED, SAVE_TASK_FAILED, SAVE_TASK_SUCCEED ) from swh.web.tests.testcase import WebTestCase class SaveApiTestCase(WebTestCase, APITestCase): @classmethod def setUpTestData(cls): # noqa: N802 SaveUnauthorizedOrigin.objects.create( url='https://github.com/user/illegal_repo') SaveUnauthorizedOrigin.objects.create( url='https://gitlab.com/user_to_exclude') def test_invalid_origin_type(self): url = reverse('api-save-origin', url_args={'origin_type': 'foo', 'origin_url': 'https://github.com/torvalds/linux'}) # noqa response = self.client.post(url) self.assertEqual(response.status_code, 400) def test_invalid_origin_url(self): url = reverse('api-save-origin', url_args={'origin_type': 'git', 'origin_url': 'bar'}) response = self.client.post(url) self.assertEqual(response.status_code, 400) def check_created_save_request_status(self, mock_scheduler, origin_url, scheduler_task_status, expected_request_status, expected_task_status=None, visit_date=None): if not scheduler_task_status: mock_scheduler.get_tasks.return_value = [] else: mock_scheduler.get_tasks.return_value = \ [{ 'priority': 'high', 'policy': 'oneshot', - 'type': 'origin-update-git', + 'type': 'load-git', 'arguments': { 'kwargs': { 'repo_url': origin_url }, 'args': [] }, 'status': scheduler_task_status, 'id': 1, }] mock_scheduler.create_tasks.return_value = \ [{ 'priority': 'high', 'policy': 'oneshot', - 'type': 'origin-update-git', + 'type': 'load-git', 'arguments': { 'kwargs': { 'repo_url': origin_url }, 'args': [] }, 'status': 'next_run_not_scheduled', 'id': 1, }] url = reverse('api-save-origin', url_args={'origin_type': 'git', 'origin_url': origin_url}) with patch('swh.web.common.origin_save._get_visit_info_for_save_request') as mock_visit_date: # noqa mock_visit_date.return_value = (visit_date, None) response = self.client.post(url) if expected_request_status != SAVE_REQUEST_REJECTED: self.assertEqual(response.status_code, 200) self.assertEqual(response.data['save_request_status'], expected_request_status) self.assertEqual(response.data['save_task_status'], expected_task_status) else: self.assertEqual(response.status_code, 403) def check_save_request_status(self, mock_scheduler, origin_url, expected_request_status, expected_task_status, scheduler_task_status='next_run_not_scheduled', # noqa visit_date=None): mock_scheduler.get_tasks.return_value = \ [{ 'priority': 'high', 'policy': 'oneshot', - 'type': 'origin-update-git', + 'type': 'load-git', 'arguments': { 'kwargs': { 'repo_url': origin_url }, 'args': [] }, 'status': scheduler_task_status, 'id': 1, }] url = reverse('api-save-origin', url_args={'origin_type': 'git', 'origin_url': origin_url}) with patch('swh.web.common.origin_save._get_visit_info_for_save_request') as mock_visit_date: # noqa mock_visit_date.return_value = (visit_date, None) response = self.client.get(url) self.assertEqual(response.status_code, 200) save_request_data = response.data[0] self.assertEqual(save_request_data['save_request_status'], expected_request_status) self.assertEqual(save_request_data['save_task_status'], expected_task_status) # Check that save task status is still available when # the scheduler task has been archived mock_scheduler.get_tasks.return_value = [] response = self.client.get(url) self.assertEqual(response.status_code, 200) save_request_data = response.data[0] self.assertEqual(save_request_data['save_task_status'], expected_task_status) @patch('swh.web.common.origin_save.scheduler') def test_save_request_rejected(self, mock_scheduler): origin_url = 'https://github.com/user/illegal_repo' self.check_created_save_request_status(mock_scheduler, origin_url, None, SAVE_REQUEST_REJECTED) self.check_save_request_status(mock_scheduler, origin_url, SAVE_REQUEST_REJECTED, SAVE_TASK_NOT_CREATED) @patch('swh.web.common.origin_save.scheduler') def test_save_request_pending(self, mock_scheduler): origin_url = 'https://unkwownforge.com/user/repo' self.check_created_save_request_status(mock_scheduler, origin_url, None, SAVE_REQUEST_PENDING, SAVE_TASK_NOT_CREATED) self.check_save_request_status(mock_scheduler, origin_url, SAVE_REQUEST_PENDING, SAVE_TASK_NOT_CREATED) @patch('swh.web.common.origin_save.scheduler') def test_save_request_succeed(self, mock_scheduler): origin_url = 'https://github.com/Kitware/CMake' self.check_created_save_request_status(mock_scheduler, origin_url, None, SAVE_REQUEST_ACCEPTED, SAVE_TASK_NOT_YET_SCHEDULED) self.check_save_request_status(mock_scheduler, origin_url, SAVE_REQUEST_ACCEPTED, SAVE_TASK_SCHEDULED, scheduler_task_status='next_run_scheduled') # noqa self.check_save_request_status(mock_scheduler, origin_url, SAVE_REQUEST_ACCEPTED, SAVE_TASK_SCHEDULED, scheduler_task_status='completed', visit_date=None) # noqa visit_date = datetime.now(tz=timezone.utc) + timedelta(hours=1) self.check_save_request_status(mock_scheduler, origin_url, SAVE_REQUEST_ACCEPTED, SAVE_TASK_SUCCEED, scheduler_task_status='completed', visit_date=visit_date) # noqa @patch('swh.web.common.origin_save.scheduler') def test_save_request_failed(self, mock_scheduler): origin_url = 'https://gitlab.com/inkscape/inkscape' self.check_created_save_request_status(mock_scheduler, origin_url, None, SAVE_REQUEST_ACCEPTED, SAVE_TASK_NOT_YET_SCHEDULED) self.check_save_request_status(mock_scheduler, origin_url, SAVE_REQUEST_ACCEPTED, SAVE_TASK_SCHEDULED, scheduler_task_status='next_run_scheduled') # noqa self.check_save_request_status(mock_scheduler, origin_url, SAVE_REQUEST_ACCEPTED, SAVE_TASK_FAILED, scheduler_task_status='disabled') # noqa @patch('swh.web.common.origin_save.scheduler') def test_create_save_request_only_when_needed(self, mock_scheduler): origin_url = 'https://github.com/webpack/webpack' SaveOriginRequest.objects.create(origin_type='git', origin_url=origin_url, status=SAVE_REQUEST_ACCEPTED, # noqa loading_task_id=56) self.check_created_save_request_status(mock_scheduler, origin_url, 'next_run_not_scheduled', SAVE_REQUEST_ACCEPTED, SAVE_TASK_NOT_YET_SCHEDULED) sors = list(SaveOriginRequest.objects.filter(origin_type='git', origin_url=origin_url)) self.assertEqual(len(sors), 1) self.check_created_save_request_status(mock_scheduler, origin_url, 'next_run_scheduled', SAVE_REQUEST_ACCEPTED, SAVE_TASK_SCHEDULED) sors = list(SaveOriginRequest.objects.filter(origin_type='git', origin_url=origin_url)) self.assertEqual(len(sors), 1) visit_date = datetime.now(tz=timezone.utc) + timedelta(hours=1) self.check_created_save_request_status(mock_scheduler, origin_url, 'completed', SAVE_REQUEST_ACCEPTED, SAVE_TASK_NOT_YET_SCHEDULED, visit_date=visit_date) sors = list(SaveOriginRequest.objects.filter(origin_type='git', origin_url=origin_url)) self.assertEqual(len(sors), 2) self.check_created_save_request_status(mock_scheduler, origin_url, 'disabled', SAVE_REQUEST_ACCEPTED, SAVE_TASK_NOT_YET_SCHEDULED) sors = list(SaveOriginRequest.objects.filter(origin_type='git', origin_url=origin_url)) self.assertEqual(len(sors), 3) def test_get_save_requests_unknown_origin(self): unknown_origin_url = 'https://gitlab.com/foo/bar' url = reverse('api-save-origin', url_args={'origin_type': 'git', 'origin_url': unknown_origin_url}) response = self.client.get(url) self.assertEqual(response.status_code, 404) self.assertEqual(response.data, { 'exception': 'NotFoundExc', 'reason': ('No save requests found for origin with type ' 'git and url %s.') % unknown_origin_url })