diff --git a/swh/web/add_forge_now/models.py b/swh/web/add_forge_now/models.py index 8683c440..c5dbab5d 100644 --- a/swh/web/add_forge_now/models.py +++ b/swh/web/add_forge_now/models.py @@ -1,71 +1,98 @@ # Copyright (C) 2022 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 __future__ import annotations + import enum +from typing import List from django.db import models class RequestStatus(enum.Enum): """Request statuses. Values are used in the ui. """ PENDING = "Pending" WAITING_FOR_FEEDBACK = "Waiting for feedback" FEEDBACK_TO_HANDLE = "Feedback to handle" ACCEPTED = "Accepted" SCHEDULED = "Scheduled" FIRST_LISTING_DONE = "First listing done" FIRST_ORIGIN_LOADED = "First origin loaded" REJECTED = "Rejected" SUSPENDED = "Suspended" DENIED = "Denied" @classmethod def choices(cls): return tuple((variant.name, variant.value) for variant in cls) + def allowed_next_statuses(self) -> List[RequestStatus]: + next_statuses = { + self.PENDING: [self.WAITING_FOR_FEEDBACK, self.REJECTED, self.SUSPENDED], + self.WAITING_FOR_FEEDBACK: [self.FEEDBACK_TO_HANDLE], + self.FEEDBACK_TO_HANDLE: [ + self.WAITING_FOR_FEEDBACK, + self.ACCEPTED, + self.REJECTED, + self.SUSPENDED, + ], + self.ACCEPTED: [self.SCHEDULED], + self.SCHEDULED: [ + self.FIRST_LISTING_DONE, + # in case of race condition between lister and loader: + self.FIRST_ORIGIN_LOADED, + ], + self.FIRST_LISTING_DONE: [self.FIRST_ORIGIN_LOADED], + self.FIRST_ORIGIN_LOADED: [], + self.REJECTED: [], + self.SUSPENDED: [self.PENDING], + self.DENIED: [], + } + return next_statuses[self] # type: ignore + class RequestActorRole(enum.Enum): MODERATOR = "moderator" SUBMITTER = "submitter" @classmethod def choices(cls): return tuple((variant.name, variant.value) for variant in cls) class RequestHistory(models.Model): """Comment or status change. This is commented or changed by either submitter or moderator. """ request = models.ForeignKey("Request", models.DO_NOTHING) text = models.TextField() actor = models.TextField() actor_role = models.TextField(choices=RequestActorRole.choices()) date = models.DateTimeField(auto_now_add=True) new_status = models.TextField(choices=RequestStatus.choices(), null=True) class Request(models.Model): status = models.TextField( choices=RequestStatus.choices(), default=RequestStatus.PENDING.name, ) submission_date = models.DateTimeField(auto_now_add=True) submitter_name = models.TextField() submitter_email = models.TextField() # FIXME: shall we do create a user model inside the webapp instead? forge_type = models.TextField() forge_url = models.TextField() forge_contact_email = models.EmailField() forge_contact_name = models.TextField() forge_contact_comment = models.TextField( help_text="Where did you find this contact information (url, ...)", ) diff --git a/swh/web/api/views/add_forge_now.py b/swh/web/api/views/add_forge_now.py index 1a87302f..ebe08927 100644 --- a/swh/web/api/views/add_forge_now.py +++ b/swh/web/api/views/add_forge_now.py @@ -1,108 +1,226 @@ # Copyright (C) 2022 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 typing import Union from django.core.exceptions import ObjectDoesNotExist -from django.forms import ModelForm +from django.db import transaction +from django.forms import CharField, ModelForm +from django.http import HttpResponseBadRequest from django.http.request import HttpRequest from django.http.response import HttpResponse, HttpResponseForbidden from rest_framework import serializers from rest_framework.request import Request from rest_framework.response import Response from swh.web.add_forge_now.models import Request as AddForgeRequest +from swh.web.add_forge_now.models import RequestActorRole as AddForgeNowRequestActorRole +from swh.web.add_forge_now.models import RequestHistory as AddForgeNowRequestHistory +from swh.web.add_forge_now.models import RequestStatus as AddForgeNowRequestStatus from swh.web.api.apidoc import api_doc, format_docstring from swh.web.api.apiurls import api_route from swh.web.common.exc import BadInputExc +MODERATOR_ROLE = "swh.web.add_forge_now.moderator" + + +def _block_while_testing(): + """Replaced by tests to check concurrency behavior + """ + pass + class AddForgeNowRequestForm(ModelForm): class Meta: model = AddForgeRequest fields = ( "forge_type", "forge_url", "forge_contact_email", "forge_contact_name", "forge_contact_comment", ) +class AddForgeNowRequestHistoryForm(ModelForm): + new_status = CharField(max_length=200, required=False,) + + class Meta: + model = AddForgeNowRequestHistory + fields = ("text", "new_status") + + class AddForgeNowRequestSerializer(serializers.ModelSerializer): class Meta: model = AddForgeRequest fields = "__all__" +class AddForgeNowRequestPublicSerializer(serializers.ModelSerializer): + """Serializes AddForgeRequest without private fields. + """ + + class Meta: + model = AddForgeRequest + fields = ("forge_url", "forge_type", "status", "submission_date") + + @api_route( r"/add-forge/request/create", "api-1-add-forge-request-create", methods=["POST"], ) @api_doc("/add-forge/request/create") @format_docstring() def api_add_forge_request_create(request: Union[HttpRequest, Request]) -> HttpResponse: """ .. http:post:: /api/1/add-forge/request/create/ Create a new request to add a forge to the list of those crawled regularly by Software Heritage. .. warning:: That endpoint is not publicly available and requires authentication in order to be able to request it. {common_headers} :[0-9]+)/update/", + "api-1-add-forge-request-update", + methods=["POST"], +) +@api_doc("/add-forge/request/update", tags=["hidden"]) +@format_docstring() +@transaction.atomic +def api_add_forge_request_update( + request: Union[HttpRequest, Request], id: int +) -> HttpResponse: + """ + .. http:post:: /api/1/add-forge/request/update/ + + Update a request to add a forge to the list of those crawled regularly + by Software Heritage. + + .. warning:: + That endpoint is not publicly available and requires authentication + in order to be able to request it. + + {common_headers} + + : HttpResponse: if isinstance(response, Response): drf_response = cast(Response, response) error_context = ( drf_response.data.pop("traceback") if isinstance(drf_response.data, dict) and "traceback" in drf_response.data else drf_response.data ) elif isinstance(response, StreamingHttpResponse): error_context = getattr(response, "traceback", response.streaming_content) else: error_context = getattr(response, "traceback", response.content) assert response.status_code == status_code, error_context if content_type != "*/*": assert response["Content-Type"].startswith(content_type) return response def check_http_get_response( client: Client, url: str, status_code: int, content_type: str = "*/*", http_origin: Optional[str] = None, server_name: Optional[str] = None, ) -> HttpResponse: """Helper function to check HTTP response for a GET request. Args: client: Django test client url: URL to check response status_code: expected HTTP status code content_type: expected response content type http_origin: optional HTTP_ORIGIN header value Returns: The HTTP response """ return _assert_http_response( response=client.get( url, HTTP_ACCEPT=content_type, HTTP_ORIGIN=http_origin, SERVER_NAME=server_name if server_name else "testserver", ), status_code=status_code, content_type=content_type, ) def check_http_post_response( client: Client, url: str, status_code: int, content_type: str = "*/*", request_content_type="application/json", data: Optional[Dict[str, Any]] = None, http_origin: Optional[str] = None, ) -> HttpResponse: """Helper function to check HTTP response for a POST request. Args: client: Django test client url: URL to check response status_code: expected HTTP status code content_type: expected response content type request_content_type: content type of request body data: optional POST data Returns: The HTTP response """ return _assert_http_response( response=client.post( url, data=data, content_type=request_content_type, HTTP_ACCEPT=content_type, HTTP_ORIGIN=http_origin, ), status_code=status_code, content_type=content_type, ) def check_api_get_responses( api_client: APIClient, url: str, status_code: int ) -> Response: """Helper function to check Web API responses for GET requests for all accepted content types (JSON, YAML, HTML). Args: api_client: DRF test client url: Web API URL to check responses status_code: expected HTTP status code Returns: The Web API JSON response """ # check JSON response response_json = check_http_get_response( api_client, url, status_code, content_type="application/json" ) # check HTML response (API Web UI) check_http_get_response(api_client, url, status_code, content_type="text/html") # check YAML response check_http_get_response( api_client, url, status_code, content_type="application/yaml" ) return cast(Response, response_json) def check_api_post_response( api_client: APIClient, url: str, status_code: int, content_type: str = "*/*", data: Optional[Dict[str, Any]] = None, ) -> HttpResponse: """Helper function to check Web API response for a POST request for all accepted content types. Args: api_client: DRF test client url: Web API URL to check response status_code: expected HTTP status code Returns: The HTTP response """ return _assert_http_response( response=api_client.post( url, data=data, format="json", HTTP_ACCEPT=content_type, ), status_code=status_code, content_type=content_type, ) def check_api_post_responses( api_client: APIClient, url: str, status_code: int, data: Optional[Dict[str, Any]] = None, ) -> Response: """Helper function to check Web API responses for POST requests for all accepted content types (JSON, YAML). Args: api_client: DRF test client url: Web API URL to check responses status_code: expected HTTP status code Returns: The Web API JSON response """ # check JSON response response_json = check_api_post_response( api_client, url, status_code, content_type="application/json", data=data ) # check YAML response check_api_post_response( api_client, url, status_code, content_type="application/yaml", data=data ) return cast(Response, response_json) def check_html_get_response( client: Client, url: str, status_code: int, template_used: Optional[str] = None ) -> HttpResponse: """Helper function to check HTML responses for a GET request. Args: client: Django test client url: URL to check responses status_code: expected HTTP status code template_used: optional used Django template to check Returns: The HTML response """ response = check_http_get_response( client, url, status_code, content_type="text/html" ) if template_used is not None: assert_template_used(response, template_used) return response def create_django_permission(perm_name: str) -> Permission: """Create permission out of a permission name string Args: perm_name: Permission name (e.g. swh.web.api.throttling_exempted, swh.ambassador, ...) Returns: The persisted permission """ perm_splitted = perm_name.split(".") app_label = ".".join(perm_splitted[:-1]) perm_name = perm_splitted[-1] - content_type = ContentType.objects.create(app_label=app_label, model="dummy") + content_type = ContentType.objects.create( + id=1000, app_label=app_label, model="dummy" + ) return Permission.objects.create( - codename=perm_name, name=perm_name, content_type=content_type, + codename=perm_name, name=perm_name, content_type=content_type, id=1000 )