diff --git a/swh/web/admin/origin_save.py b/swh/web/admin/origin_save.py index 1b3271b9..6245daaa 100644 --- a/swh/web/admin/origin_save.py +++ b/swh/web/admin/origin_save.py @@ -1,210 +1,214 @@ # Copyright (C) 2018-2019 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 +import json + from django.conf import settings from django.contrib.admin.views.decorators import staff_member_required from django.core.exceptions import ObjectDoesNotExist from django.core.paginator import Paginator from django.http import HttpResponse, JsonResponse from django.shortcuts import render from django.views.decorators.http import require_POST from swh.web.admin.adminurls import admin_route from swh.web.common.models import ( SaveAuthorizedOrigin, SaveOriginRequest, SaveUnauthorizedOrigin, ) from swh.web.common.origin_save import ( SAVE_REQUEST_PENDING, SAVE_REQUEST_REJECTED, create_save_origin_request, ) @admin_route(r"origin/save/", view_name="admin-origin-save") @staff_member_required(view_func=None, login_url=settings.LOGIN_URL) def _admin_origin_save(request): return render(request, "admin/origin-save.html") def _datatables_origin_urls_response(request, urls_query_set): search_value = request.GET["search[value]"] if search_value: urls_query_set = urls_query_set.filter(url__icontains=search_value) column_order = request.GET["order[0][column]"] field_order = request.GET["columns[%s][name]" % column_order] order_dir = request.GET["order[0][dir]"] if order_dir == "desc": field_order = "-" + field_order urls_query_set = urls_query_set.order_by(field_order) table_data = {} table_data["draw"] = int(request.GET["draw"]) table_data["recordsTotal"] = urls_query_set.count() table_data["recordsFiltered"] = urls_query_set.count() length = int(request.GET["length"]) page = int(request.GET["start"]) / length + 1 paginator = Paginator(urls_query_set, length) urls_query_set = paginator.page(page).object_list table_data["data"] = [{"url": u.url} for u in urls_query_set] return JsonResponse(table_data) @admin_route( r"origin/save/authorized_urls/list/", view_name="admin-origin-save-authorized-urls-list", ) @staff_member_required(view_func=None, login_url=settings.LOGIN_URL) def _admin_origin_save_authorized_urls_list(request): authorized_urls = SaveAuthorizedOrigin.objects.all() return _datatables_origin_urls_response(request, authorized_urls) @admin_route( r"origin/save/authorized_urls/add/(?P.+)/", view_name="admin-origin-save-add-authorized-url", ) @require_POST @staff_member_required(view_func=None, login_url=settings.LOGIN_URL) def _admin_origin_save_add_authorized_url(request, origin_url): try: SaveAuthorizedOrigin.objects.get(url=origin_url) except ObjectDoesNotExist: # add the new authorized url SaveAuthorizedOrigin.objects.create(url=origin_url) # check if pending save requests with that url prefix exist pending_save_requests = SaveOriginRequest.objects.filter( origin_url__startswith=origin_url, status=SAVE_REQUEST_PENDING ) # create origin save tasks for previously pending requests for psr in pending_save_requests: create_save_origin_request(psr.visit_type, psr.origin_url) status_code = 200 else: status_code = 400 return HttpResponse(status=status_code) @admin_route( r"origin/save/authorized_urls/remove/(?P.+)/", view_name="admin-origin-save-remove-authorized-url", ) @require_POST @staff_member_required(view_func=None, login_url=settings.LOGIN_URL) def _admin_origin_save_remove_authorized_url(request, origin_url): try: entry = SaveAuthorizedOrigin.objects.get(url=origin_url) except ObjectDoesNotExist: status_code = 404 else: entry.delete() status_code = 200 return HttpResponse(status=status_code) @admin_route( r"origin/save/unauthorized_urls/list/", view_name="admin-origin-save-unauthorized-urls-list", ) @staff_member_required(view_func=None, login_url=settings.LOGIN_URL) def _admin_origin_save_unauthorized_urls_list(request): unauthorized_urls = SaveUnauthorizedOrigin.objects.all() return _datatables_origin_urls_response(request, unauthorized_urls) @admin_route( r"origin/save/unauthorized_urls/add/(?P.+)/", view_name="admin-origin-save-add-unauthorized-url", ) @require_POST @staff_member_required(view_func=None, login_url=settings.LOGIN_URL) def _admin_origin_save_add_unauthorized_url(request, origin_url): try: SaveUnauthorizedOrigin.objects.get(url=origin_url) except ObjectDoesNotExist: SaveUnauthorizedOrigin.objects.create(url=origin_url) # check if pending save requests with that url prefix exist pending_save_requests = SaveOriginRequest.objects.filter( origin_url__startswith=origin_url, status=SAVE_REQUEST_PENDING ) # mark pending requests as rejected for psr in pending_save_requests: psr.status = SAVE_REQUEST_REJECTED psr.save() status_code = 200 else: status_code = 400 return HttpResponse(status=status_code) @admin_route( r"origin/save/unauthorized_urls/remove/(?P.+)/", view_name="admin-origin-save-remove-unauthorized-url", ) @require_POST @staff_member_required(view_func=None, login_url=settings.LOGIN_URL) def _admin_origin_save_remove_unauthorized_url(request, origin_url): try: entry = SaveUnauthorizedOrigin.objects.get(url=origin_url) except ObjectDoesNotExist: status_code = 404 else: entry.delete() status_code = 200 return HttpResponse(status=status_code) @admin_route( r"origin/save/request/accept/(?P.+)/url/(?P.+)/", view_name="admin-origin-save-request-accept", ) @require_POST @staff_member_required(view_func=None, login_url=settings.LOGIN_URL) def _admin_origin_save_request_accept(request, visit_type, origin_url): try: SaveAuthorizedOrigin.objects.get(url=origin_url) except ObjectDoesNotExist: SaveAuthorizedOrigin.objects.create(url=origin_url) create_save_origin_request(visit_type, origin_url) return HttpResponse(status=200) @admin_route( r"origin/save/request/reject/(?P.+)/url/(?P.+)/", view_name="admin-origin-save-request-reject", ) @require_POST @staff_member_required(view_func=None, login_url=settings.LOGIN_URL) def _admin_origin_save_request_reject(request, visit_type, origin_url): try: SaveUnauthorizedOrigin.objects.get(url=origin_url) except ObjectDoesNotExist: SaveUnauthorizedOrigin.objects.create(url=origin_url) sor = SaveOriginRequest.objects.get( visit_type=visit_type, origin_url=origin_url, status=SAVE_REQUEST_PENDING ) + sor.status = SAVE_REQUEST_REJECTED + sor.note = json.loads(request.body).get("note") sor.save() return HttpResponse(status=200) @admin_route( r"origin/save/request/remove/(?P.+)/", view_name="admin-origin-save-request-remove", ) @require_POST @staff_member_required(view_func=None, login_url=settings.LOGIN_URL) def _admin_origin_save_request_remove(request, sor_id): try: entry = SaveOriginRequest.objects.get(id=sor_id) except ObjectDoesNotExist: status_code = 404 else: entry.delete() status_code = 200 return HttpResponse(status=status_code) diff --git a/swh/web/tests/admin/test_origin_save.py b/swh/web/tests/admin/test_origin_save.py index 63babc3c..e55c11fc 100644 --- a/swh/web/tests/admin/test_origin_save.py +++ b/swh/web/tests/admin/test_origin_save.py @@ -1,182 +1,213 @@ # Copyright (C) 2015-2021 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 import pytest from swh.web.common.models import ( SAVE_REQUEST_ACCEPTED, SAVE_REQUEST_PENDING, SAVE_REQUEST_REJECTED, SAVE_TASK_NOT_YET_SCHEDULED, SaveAuthorizedOrigin, SaveOriginRequest, SaveUnauthorizedOrigin, ) from swh.web.common.origin_save import can_save_origin from swh.web.common.utils import reverse from swh.web.tests.utils import check_http_get_response, check_http_post_response _authorized_origin_url = "https://scm.ourproject.org/anonscm/" _unauthorized_origin_url = "https://www.softwareheritage.org/" pytestmark = pytest.mark.django_db @pytest.fixture(autouse=True) def populated_db(): SaveAuthorizedOrigin.objects.create(url=_authorized_origin_url) SaveUnauthorizedOrigin.objects.create(url=_unauthorized_origin_url) def check_not_login(client, url): login_url = reverse("login", query_params={"next": url}) resp = check_http_post_response(client, url, status_code=302) assert unquote(resp.url) == login_url def test_add_authorized_origin_url(client, staff_user): authorized_url = "https://scm.adullact.net/anonscm/" assert can_save_origin(authorized_url) == SAVE_REQUEST_PENDING url = reverse( "admin-origin-save-add-authorized-url", url_args={"origin_url": authorized_url} ) check_not_login(client, url) assert can_save_origin(authorized_url) == SAVE_REQUEST_PENDING client.force_login(staff_user) check_http_post_response(client, url, status_code=200) assert can_save_origin(authorized_url) == SAVE_REQUEST_ACCEPTED def test_remove_authorized_origin_url(client, staff_user): assert can_save_origin(_authorized_origin_url) == SAVE_REQUEST_ACCEPTED url = reverse( "admin-origin-save-remove-authorized-url", url_args={"origin_url": _authorized_origin_url}, ) check_not_login(client, url) assert can_save_origin(_authorized_origin_url) == SAVE_REQUEST_ACCEPTED client.force_login(staff_user) check_http_post_response(client, url, status_code=200) assert can_save_origin(_authorized_origin_url) == SAVE_REQUEST_PENDING def test_add_unauthorized_origin_url(client, staff_user): unauthorized_url = "https://www.yahoo./" assert can_save_origin(unauthorized_url) == SAVE_REQUEST_PENDING url = reverse( "admin-origin-save-add-unauthorized-url", url_args={"origin_url": unauthorized_url}, ) check_not_login(client, url) assert can_save_origin(unauthorized_url) == SAVE_REQUEST_PENDING client.force_login(staff_user) check_http_post_response(client, url, status_code=200) assert can_save_origin(unauthorized_url) == SAVE_REQUEST_REJECTED def test_remove_unauthorized_origin_url(client, staff_user): assert can_save_origin(_unauthorized_origin_url) == SAVE_REQUEST_REJECTED url = reverse( "admin-origin-save-remove-unauthorized-url", url_args={"origin_url": _unauthorized_origin_url}, ) check_not_login(client, url) assert can_save_origin(_unauthorized_origin_url) == SAVE_REQUEST_REJECTED client.force_login(staff_user) check_http_post_response(client, url, status_code=200) assert can_save_origin(_unauthorized_origin_url) == SAVE_REQUEST_PENDING def test_accept_pending_save_request(client, staff_user, swh_scheduler): visit_type = "git" origin_url = "https://v2.pikacode.com/bthate/botlib.git" save_request_url = reverse( "api-1-save-origin", url_args={"visit_type": visit_type, "origin_url": origin_url}, ) response = check_http_post_response(client, save_request_url, status_code=200) assert response.data["save_request_status"] == SAVE_REQUEST_PENDING accept_request_url = reverse( "admin-origin-save-request-accept", url_args={"visit_type": visit_type, "origin_url": origin_url}, ) check_not_login(client, accept_request_url) client.force_login(staff_user) response = check_http_post_response(client, accept_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_task_status"] == SAVE_TASK_NOT_YET_SCHEDULED def test_reject_pending_save_request(client, staff_user, swh_scheduler): visit_type = "git" origin_url = "https://wikipedia.com" save_request_url = reverse( "api-1-save-origin", url_args={"visit_type": visit_type, "origin_url": origin_url}, ) response = check_http_post_response(client, save_request_url, status_code=200) assert response.data["save_request_status"] == SAVE_REQUEST_PENDING reject_request_url = reverse( "admin-origin-save-request-reject", url_args={"visit_type": visit_type, "origin_url": origin_url}, ) check_not_login(client, reject_request_url) client.force_login(staff_user) response = check_http_post_response(client, reject_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]["note"] is None + + +def test_reject_pending_save_request_with_note(client, staff_user, swh_scheduler): + + visit_type = "git" + origin_url = "https://wikipedia.com" + + save_request_url = reverse( + "api-1-save-origin", + url_args={"visit_type": visit_type, "origin_url": origin_url}, + ) + + response = check_http_post_response(client, save_request_url, status_code=200) + assert response.data["save_request_status"] == SAVE_REQUEST_PENDING + + reject_request_url = reverse( + "admin-origin-save-request-reject", + url_args={"visit_type": visit_type, "origin_url": origin_url}, + ) + + data = {"note": "The URL does not target a git repository"} + + client.force_login(staff_user) + response = check_http_post_response( + client, reject_request_url, status_code=200, data=data + ) + + 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]["note"] == data["note"] def test_remove_save_request(client, staff_user): sor = SaveOriginRequest.objects.create( visit_type="git", origin_url="https://wikipedia.com", status=SAVE_REQUEST_PENDING, ) assert SaveOriginRequest.objects.count() == 1 remove_request_url = reverse( "admin-origin-save-request-remove", url_args={"sor_id": sor.id} ) check_not_login(client, remove_request_url) client.force_login(staff_user) check_http_post_response(client, remove_request_url, status_code=200) assert SaveOriginRequest.objects.count() == 0