Page MenuHomeSoftware Heritage
Differential D2242 Diff 7865 swh/web/tests/admin/test_origin_save.py

Changeset View

Standalone View

View Options

swh/web/tests/admin/test_origin_save.py

# Copyright (C) 2015-2019 The Software Heritage developers # Copyright (C) 2015-2019 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 Affero General Public License version 3, or any later version # License: GNU Affero 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
from urllib.parse import unquote from urllib.parse import unquote
import pytest
from django.contrib.auth import get_user_model from django.contrib.auth import get_user_model
from unittest.mock import patch
from swh.web.common.models import ( from swh.web.common.models import (
SaveAuthorizedOrigin, SaveUnauthorizedOrigin, SaveOriginRequest SaveAuthorizedOrigin, SaveUnauthorizedOrigin, SaveOriginRequest
) )
from swh.web.common.origin_save import can_save_origin from swh.web.common.origin_save import can_save_origin
from swh.web.common.models import ( from swh.web.common.models import (
SAVE_REQUEST_PENDING, SAVE_REQUEST_ACCEPTED, SAVE_REQUEST_PENDING, SAVE_REQUEST_ACCEPTED,
SAVE_REQUEST_REJECTED, SAVE_TASK_NOT_YET_SCHEDULED SAVE_REQUEST_REJECTED, SAVE_TASK_NOT_YET_SCHEDULED
) )
from swh.web.common.utils import reverse from swh.web.common.utils import reverse
from swh.web.tests.testcase import WebTestCase
_user_name = 'swh-web-admin' _user_name = 'swh-web-admin'
_user_mail = 'admin@swh-web.org' _user_mail = 'admin@swh-web.org'
_user_password = '..34~pounds~BEAUTY~march~63..' _user_password = '..34~pounds~BEAUTY~march~63..'
_authorized_origin_url = 'https://scm.ourproject.org/anonscm/' _authorized_origin_url = 'https://scm.ourproject.org/anonscm/'
_unauthorized_origin_url = 'https://www.softwareheritage.org/' _unauthorized_origin_url = 'https://www.softwareheritage.org/'
class OriginSaveAdminTestCase(WebTestCase): pytestmark = pytest.mark.django_db
@classmethod
def setUpTestData(cls): # noqa: N802 @pytest.fixture(autouse=True)
vlorentzUnsubmitted
Not Done
Inline Actions

fixture names should be nouns, not verbs. eg. populated_db

vlorentz: fixture names should be nouns, not verbs. eg. `populated_db`
anlambertAuthorUnsubmitted
Done
Inline Actions

ack

anlambert: ack
User = get_user_model() # noqa: N806 def populated_db():
User = get_user_model()
user = User.objects.create_user(_user_name, _user_mail, _user_password) user = User.objects.create_user(_user_name, _user_mail, _user_password)
user.is_staff = True user.is_staff = True
user.save() user.save()
SaveAuthorizedOrigin.objects.create(url=_authorized_origin_url) SaveAuthorizedOrigin.objects.create(url=_authorized_origin_url)
SaveUnauthorizedOrigin.objects.create(url=_unauthorized_origin_url) SaveUnauthorizedOrigin.objects.create(url=_unauthorized_origin_url)
def check_not_login(self, url):
def check_not_login(client, url):
login_url = reverse('login', query_params={'next': url}) login_url = reverse('login', query_params={'next': url})
response = self.client.post(url) response = client.post(url)
self.assertEqual(response.status_code, 302) assert response.status_code == 302
self.assertEqual(unquote(response.url), login_url) assert unquote(response.url) == login_url
def test_add_authorized_origin_url(self): def test_add_authorized_origin_url(client):
authorized_url = 'https://scm.adullact.net/anonscm/' authorized_url = 'https://scm.adullact.net/anonscm/'
self.assertEqual(can_save_origin(authorized_url), assert can_save_origin(authorized_url) == SAVE_REQUEST_PENDING
SAVE_REQUEST_PENDING)
url = reverse('admin-origin-save-add-authorized-url', url = reverse('admin-origin-save-add-authorized-url',
url_args={'origin_url': authorized_url}) url_args={'origin_url': authorized_url})
self.check_not_login(url) check_not_login(client, url)
self.assertEqual(can_save_origin(authorized_url), assert can_save_origin(authorized_url) == SAVE_REQUEST_PENDING
SAVE_REQUEST_PENDING)
self.client.login(username=_user_name, password=_user_password) client.login(username=_user_name, password=_user_password)
response = self.client.post(url) response = client.post(url)
self.assertEqual(response.status_code, 200) assert response.status_code == 200
self.assertEqual(can_save_origin(authorized_url), assert can_save_origin(authorized_url) == SAVE_REQUEST_ACCEPTED
SAVE_REQUEST_ACCEPTED)
def test_remove_authorized_origin_url(self): def test_remove_authorized_origin_url(client):
self.assertEqual(can_save_origin(_authorized_origin_url), assert can_save_origin(_authorized_origin_url) == SAVE_REQUEST_ACCEPTED
SAVE_REQUEST_ACCEPTED)
url = reverse('admin-origin-save-remove-authorized-url', url = reverse('admin-origin-save-remove-authorized-url',
url_args={'origin_url': _authorized_origin_url}) url_args={'origin_url': _authorized_origin_url})
self.check_not_login(url) check_not_login(client, url)
assert can_save_origin(_authorized_origin_url) == SAVE_REQUEST_ACCEPTED
self.assertEqual(can_save_origin(_authorized_origin_url), client.login(username=_user_name, password=_user_password)
SAVE_REQUEST_ACCEPTED) response = client.post(url)
assert response.status_code == 200
assert can_save_origin(_authorized_origin_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_origin_url),
SAVE_REQUEST_PENDING)
def test_add_unauthorized_origin_url(self): def test_add_unauthorized_origin_url(client):
unauthorized_url = 'https://www.yahoo./' unauthorized_url = 'https://www.yahoo./'
self.assertEqual(can_save_origin(unauthorized_url), assert can_save_origin(unauthorized_url) == SAVE_REQUEST_PENDING
SAVE_REQUEST_PENDING)
url = reverse('admin-origin-save-add-unauthorized-url', url = reverse('admin-origin-save-add-unauthorized-url',
url_args={'origin_url': unauthorized_url}) url_args={'origin_url': unauthorized_url})
self.check_not_login(url) check_not_login(client, url)
self.assertEqual(can_save_origin(unauthorized_url), assert can_save_origin(unauthorized_url) == SAVE_REQUEST_PENDING
SAVE_REQUEST_PENDING)
self.client.login(username=_user_name, password=_user_password) client.login(username=_user_name, password=_user_password)
response = self.client.post(url) response = client.post(url)
self.assertEqual(response.status_code, 200) assert response.status_code == 200
self.assertEqual(can_save_origin(unauthorized_url), assert can_save_origin(unauthorized_url) == SAVE_REQUEST_REJECTED
SAVE_REQUEST_REJECTED)
def test_remove_unauthorized_origin_url(self): def test_remove_unauthorized_origin_url(client):
self.assertEqual(can_save_origin(_unauthorized_origin_url), assert can_save_origin(_unauthorized_origin_url) == SAVE_REQUEST_REJECTED
SAVE_REQUEST_REJECTED)
url = reverse('admin-origin-save-remove-unauthorized-url', url = reverse('admin-origin-save-remove-unauthorized-url',
url_args={'origin_url': _unauthorized_origin_url}) url_args={'origin_url': _unauthorized_origin_url})
self.check_not_login(url) check_not_login(client, url)
assert can_save_origin(_unauthorized_origin_url) == SAVE_REQUEST_REJECTED
self.assertEqual(can_save_origin(_unauthorized_origin_url), client.login(username=_user_name, password=_user_password)
SAVE_REQUEST_REJECTED) response = client.post(url)
assert response.status_code == 200
assert can_save_origin(_unauthorized_origin_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_origin_url),
SAVE_REQUEST_PENDING)
@patch('swh.web.common.origin_save.scheduler') def test_accept_pending_save_request(client, mocker):
def test_accept_pending_save_request(self, mock_scheduler): mock_scheduler = mocker.patch('swh.web.common.origin_save.scheduler')
visit_type = 'git' visit_type = 'git'
origin_url = 'https://v2.pikacode.com/bthate/botlib.git' origin_url = 'https://v2.pikacode.com/bthate/botlib.git'
save_request_url = reverse('api-1-save-origin', save_request_url = reverse('api-1-save-origin',
url_args={'visit_type': visit_type, url_args={'visit_type': visit_type,
'origin_url': origin_url}) 'origin_url': origin_url})
response = self.client.post(save_request_url, data={}, response = client.post(save_request_url, data={},
content_type='application/x-www-form-urlencoded') # noqa content_type='application/x-www-form-urlencoded')
self.assertEqual(response.status_code, 200) assert response.status_code == 200
self.assertEqual(response.data['save_request_status'], assert response.data['save_request_status'] == SAVE_REQUEST_PENDING
SAVE_REQUEST_PENDING)
accept_request_url = reverse('admin-origin-save-request-accept', accept_request_url = reverse('admin-origin-save-request-accept',
url_args={'visit_type': visit_type, url_args={'visit_type': visit_type,
'origin_url': origin_url}) 'origin_url': origin_url})
self.check_not_login(accept_request_url) check_not_login(client, accept_request_url)
tasks_data = [ tasks_data = [
{ {
'priority': 'high', 'priority': 'high',
'policy': 'oneshot', 'policy': 'oneshot',
'type': 'load-git', 'type': 'load-git',
'arguments': { 'arguments': {
'kwargs': { 'kwargs': {
'repo_url': origin_url 'repo_url': origin_url
}, },
'args': [] 'args': []
}, },
'status': 'next_run_not_scheduled', 'status': 'next_run_not_scheduled',
'id': 1, 'id': 1,
} }
] ]
mock_scheduler.create_tasks.return_value = tasks_data mock_scheduler.create_tasks.return_value = tasks_data
mock_scheduler.get_tasks.return_value = tasks_data mock_scheduler.get_tasks.return_value = tasks_data
self.client.login(username=_user_name, password=_user_password) client.login(username=_user_name, password=_user_password)
response = self.client.post(accept_request_url) response = client.post(accept_request_url)
self.assertEqual(response.status_code, 200) assert response.status_code == 200
response = self.client.get(save_request_url) response = client.get(save_request_url)
self.assertEqual(response.status_code, 200) assert response.status_code == 200
self.assertEqual(response.data[0]['save_request_status'], assert response.data[0]['save_request_status'] == SAVE_REQUEST_ACCEPTED
SAVE_REQUEST_ACCEPTED) assert response.data[0]['save_task_status'] == SAVE_TASK_NOT_YET_SCHEDULED
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): def test_reject_pending_save_request(client, mocker):
mock_scheduler = mocker.patch('swh.web.common.origin_save.scheduler')
visit_type = 'git' visit_type = 'git'
origin_url = 'https://wikipedia.com' origin_url = 'https://wikipedia.com'
save_request_url = reverse('api-1-save-origin', save_request_url = reverse('api-1-save-origin',
url_args={'visit_type': visit_type, url_args={'visit_type': visit_type,
'origin_url': origin_url}) 'origin_url': origin_url})
response = self.client.post(save_request_url, data={}, response = client.post(save_request_url, data={},
content_type='application/x-www-form-urlencoded') # noqa content_type='application/x-www-form-urlencoded')
self.assertEqual(response.status_code, 200) assert response.status_code == 200
self.assertEqual(response.data['save_request_status'], assert response.data['save_request_status'] == SAVE_REQUEST_PENDING
SAVE_REQUEST_PENDING)
reject_request_url = reverse('admin-origin-save-request-reject', reject_request_url = reverse('admin-origin-save-request-reject',
url_args={'visit_type': visit_type, url_args={'visit_type': visit_type,
'origin_url': origin_url}) 'origin_url': origin_url})
self.check_not_login(reject_request_url) check_not_login(client, reject_request_url)
self.client.login(username=_user_name, password=_user_password) client.login(username=_user_name, password=_user_password)
response = self.client.post(reject_request_url) response = client.post(reject_request_url)
self.assertEqual(response.status_code, 200) assert response.status_code == 200
tasks_data = [ tasks_data = [
{ {
'priority': 'high', 'priority': 'high',
'policy': 'oneshot', 'policy': 'oneshot',
'type': 'load-git', 'type': 'load-git',
'arguments': { 'arguments': {
'kwargs': { 'kwargs': {
'repo_url': origin_url 'repo_url': origin_url
}, },
'args': [] 'args': []
}, },
'status': 'next_run_not_scheduled', 'status': 'next_run_not_scheduled',
'id': 1, 'id': 1,
} }
] ]
mock_scheduler.create_tasks.return_value = tasks_data mock_scheduler.create_tasks.return_value = tasks_data
mock_scheduler.get_tasks.return_value = tasks_data mock_scheduler.get_tasks.return_value = tasks_data
response = self.client.get(save_request_url) response = client.get(save_request_url)
self.assertEqual(response.status_code, 200) assert response.status_code == 200
self.assertEqual(response.data[0]['save_request_status'], assert response.data[0]['save_request_status'] == SAVE_REQUEST_REJECTED
SAVE_REQUEST_REJECTED)
def test_remove_save_request(self): def test_remove_save_request(client):
sor = SaveOriginRequest.objects.create(visit_type='git', sor = SaveOriginRequest.objects.create(visit_type='git',
origin_url='https://wikipedia.com', # noqa origin_url='https://wikipedia.com', # noqa
status=SAVE_REQUEST_PENDING) status=SAVE_REQUEST_PENDING)
self.assertEqual(SaveOriginRequest.objects.count(), 1) assert SaveOriginRequest.objects.count() == 1
remove_request_url = reverse('admin-origin-save-request-remove', remove_request_url = reverse('admin-origin-save-request-remove',
url_args={'sor_id': sor.id}) url_args={'sor_id': sor.id})
self.check_not_login(remove_request_url) check_not_login(client, remove_request_url)
self.client.login(username=_user_name, password=_user_password) client.login(username=_user_name, password=_user_password)
response = self.client.post(remove_request_url) response = client.post(remove_request_url)
self.assertEqual(response.status_code, 200) assert response.status_code == 200
self.assertEqual(SaveOriginRequest.objects.count(), 0) assert SaveOriginRequest.objects.count() == 0
About · Developer information · Legal