Differential D6170 Diff 22357 swh/web/tests/admin/test_origin_save.py

Changeset View

Standalone View

swh/web/tests/admin/test_origin_save.py

# Copyright (C) 2015-2019 The Software Heritage developers		# Copyright (C) 2015-2021 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		import pytest

▲ Show 20 Lines • Show All 105 Lines • ▼ Show 20 Lines	def test_remove_unauthorized_origin_url(client):

assert can_save_origin(_unauthorized_origin_url) == SAVE_REQUEST_REJECTED		assert can_save_origin(_unauthorized_origin_url) == SAVE_REQUEST_REJECTED

client.login(username=_user_name, password=_user_password)		client.login(username=_user_name, password=_user_password)
check_http_post_response(client, url, status_code=200)		check_http_post_response(client, url, status_code=200)
assert can_save_origin(_unauthorized_origin_url) == SAVE_REQUEST_PENDING		assert can_save_origin(_unauthorized_origin_url) == SAVE_REQUEST_PENDING


def test_accept_pending_save_request(client, mocker):		def test_accept_pending_save_request(client, swh_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(		save_request_url = reverse(
"api-1-save-origin",		"api-1-save-origin",
url_args={"visit_type": visit_type, "origin_url": origin_url},		url_args={"visit_type": visit_type, "origin_url": origin_url},
)		)
response = check_http_post_response(client, save_request_url, status_code=200)		response = check_http_post_response(client, save_request_url, status_code=200)
assert response.data["save_request_status"] == SAVE_REQUEST_PENDING		assert response.data["save_request_status"] == SAVE_REQUEST_PENDING

accept_request_url = reverse(		accept_request_url = reverse(
"admin-origin-save-request-accept",		"admin-origin-save-request-accept",
url_args={"visit_type": visit_type, "origin_url": origin_url},		url_args={"visit_type": visit_type, "origin_url": origin_url},
)		)

check_not_login(client, accept_request_url)		check_not_login(client, accept_request_url)

tasks_data = [
{
"priority": "high",
"policy": "oneshot",
"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

client.login(username=_user_name, password=_user_password)		client.login(username=_user_name, password=_user_password)
response = check_http_post_response(client, accept_request_url, status_code=200)		response = check_http_post_response(client, accept_request_url, status_code=200)

response = check_http_get_response(client, save_request_url, status_code=200)		response = check_http_get_response(client, save_request_url, status_code=200)
assert response.data[0]["save_request_status"] == SAVE_REQUEST_ACCEPTED		assert response.data[0]["save_request_status"] == SAVE_REQUEST_ACCEPTED
assert response.data[0]["save_task_status"] == SAVE_TASK_NOT_YET_SCHEDULED		assert response.data[0]["save_task_status"] == SAVE_TASK_NOT_YET_SCHEDULED


def test_reject_pending_save_request(client, mocker):		def test_reject_pending_save_request(client, swh_scheduler):
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(		save_request_url = reverse(
"api-1-save-origin",		"api-1-save-origin",
url_args={"visit_type": visit_type, "origin_url": origin_url},		url_args={"visit_type": visit_type, "origin_url": origin_url},
)		)

response = check_http_post_response(client, save_request_url, status_code=200)		response = check_http_post_response(client, save_request_url, status_code=200)
assert response.data["save_request_status"] == SAVE_REQUEST_PENDING		assert response.data["save_request_status"] == SAVE_REQUEST_PENDING

reject_request_url = reverse(		reject_request_url = reverse(
"admin-origin-save-request-reject",		"admin-origin-save-request-reject",
url_args={"visit_type": visit_type, "origin_url": origin_url},		url_args={"visit_type": visit_type, "origin_url": origin_url},
)		)

check_not_login(client, reject_request_url)		check_not_login(client, reject_request_url)

client.login(username=_user_name, password=_user_password)		client.login(username=_user_name, password=_user_password)
response = check_http_post_response(client, reject_request_url, status_code=200)		response = check_http_post_response(client, reject_request_url, status_code=200)

tasks_data = [
{
"priority": "high",
"policy": "oneshot",
"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 = check_http_get_response(client, save_request_url, status_code=200)		response = check_http_get_response(client, save_request_url, status_code=200)
assert response.data[0]["save_request_status"] == SAVE_REQUEST_REJECTED		assert response.data[0]["save_request_status"] == SAVE_REQUEST_REJECTED


def test_remove_save_request(client):		def test_remove_save_request(client):
sor = SaveOriginRequest.objects.create(		sor = SaveOriginRequest.objects.create(
visit_type="git",		visit_type="git",
origin_url="https://wikipedia.com",		origin_url="https://wikipedia.com",
Show All 13 Lines