Page MenuHomeSoftware Heritage

test_api_views.py
No OneTemporary

test_api_views.py

# 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 datetime
import threading
import time
from typing import Dict
from urllib.parse import urlencode, urlparse
import iso8601
import pytest
from swh.web.add_forge_now.models import Request, RequestHistory
from swh.web.config import get_config
from swh.web.inbound_email.utils import get_address_for_pk
from swh.web.tests.helpers import (
check_api_get_responses,
check_api_post_response,
check_http_get_response,
check_http_post_response,
)
from swh.web.utils import reverse
@pytest.mark.django_db
def test_add_forge_request_create_anonymous_user(api_client):
url = reverse("api-1-add-forge-request-create")
check_api_post_response(api_client, url, status_code=403)
@pytest.mark.django_db
def test_add_forge_request_create_empty(api_client, regular_user):
api_client.force_login(regular_user)
url = reverse("api-1-add-forge-request-create")
resp = check_api_post_response(api_client, url, status_code=400)
assert '"forge_type"' in resp.data["reason"]
ADD_FORGE_DATA_FORGE1: Dict = {
"forge_type": "gitlab",
"forge_url": "https://gitlab.example.org",
"forge_contact_email": "admin@gitlab.example.org",
"forge_contact_name": "gitlab.example.org admin",
"forge_contact_comment": "user marked as owner in forge members",
"submitter_forward_username": True,
}
ADD_FORGE_DATA_FORGE2: Dict = {
"forge_type": "gitea",
"forge_url": "https://gitea.example.org",
"forge_contact_email": "admin@gitea.example.org",
"forge_contact_name": "gitea.example.org admin",
"forge_contact_comment": "user marked as owner in forge members",
"submitter_forward_username": True,
}
ADD_FORGE_DATA_FORGE3: Dict = {
"forge_type": "heptapod",
"forge_url": "https://heptapod.host/",
"forge_contact_email": "admin@example.org",
"forge_contact_name": "heptapod admin",
"forge_contact_comment": "", # authorized empty or null comment
"submitter_forward_username": False,
}
ADD_FORGE_DATA_FORGE4: Dict = {
**ADD_FORGE_DATA_FORGE3,
"forge_url": "https://heptapod2.host/",
"submitter_forward_username": "on",
}
ADD_FORGE_DATA_FORGE5: Dict = {
**ADD_FORGE_DATA_FORGE3,
"forge_url": "https://heptapod3.host/",
"submitter_forward_username": "off",
}
def inbound_email_for_pk(pk: int) -> str:
"""Check that the inbound email matches the one expected for the given pk"""
base_address = get_config()["add_forge_now"]["email_address"]
return get_address_for_pk(
salt="swh_web_add_forge_now", base_address=base_address, pk=pk
)
@pytest.mark.django_db(transaction=True, reset_sequences=True)
@pytest.mark.parametrize(
"add_forge_data",
[
ADD_FORGE_DATA_FORGE1,
ADD_FORGE_DATA_FORGE2,
ADD_FORGE_DATA_FORGE3,
ADD_FORGE_DATA_FORGE4,
],
)
def test_add_forge_request_create_success_post(
api_client, regular_user, add_forge_data
):
api_client.force_login(regular_user)
url = reverse("api-1-add-forge-request-create")
date_before = datetime.datetime.now(tz=datetime.timezone.utc)
resp = check_api_post_response(
api_client,
url,
data=add_forge_data,
status_code=201,
)
date_after = datetime.datetime.now(tz=datetime.timezone.utc)
consent = add_forge_data["submitter_forward_username"]
# map the expected result with what's expectedly read from the db to ease comparison
expected_consent_bool = consent == "on" if isinstance(consent, str) else consent
assert resp.data == {
**add_forge_data,
"id": resp.data["id"],
"status": "PENDING",
"submission_date": resp.data["submission_date"],
"submitter_name": regular_user.username,
"submitter_email": regular_user.email,
"submitter_forward_username": expected_consent_bool,
"last_moderator": resp.data["last_moderator"],
"last_modified_date": resp.data["last_modified_date"],
"inbound_email_address": inbound_email_for_pk(resp.data["id"]),
"forge_domain": urlparse(add_forge_data["forge_url"]).netloc,
}
assert date_before < iso8601.parse_date(resp.data["submission_date"]) < date_after
request = Request.objects.all().last()
assert request.forge_url == add_forge_data["forge_url"]
assert request.submitter_name == regular_user.username
@pytest.mark.django_db(transaction=True, reset_sequences=True)
def test_add_forge_request_create_success_form_encoded(client, regular_user):
client.force_login(regular_user)
url = reverse("api-1-add-forge-request-create")
date_before = datetime.datetime.now(tz=datetime.timezone.utc)
resp = check_http_post_response(
client,
url,
request_content_type="application/x-www-form-urlencoded",
data=urlencode(ADD_FORGE_DATA_FORGE1),
status_code=201,
)
date_after = datetime.datetime.now(tz=datetime.timezone.utc)
assert resp.data == {
**ADD_FORGE_DATA_FORGE1,
"id": resp.data["id"],
"status": "PENDING",
"submission_date": resp.data["submission_date"],
"submitter_name": regular_user.username,
"submitter_email": regular_user.email,
"last_moderator": resp.data["last_moderator"],
"last_modified_date": resp.data["last_modified_date"],
"inbound_email_address": inbound_email_for_pk(1),
"forge_domain": urlparse(ADD_FORGE_DATA_FORGE1["forge_url"]).netloc,
}
assert date_before < iso8601.parse_date(resp.data["submission_date"]) < date_after
request = Request.objects.all()[0]
assert request.forge_url == ADD_FORGE_DATA_FORGE1["forge_url"]
assert request.submitter_name == regular_user.username
@pytest.mark.django_db(transaction=True, reset_sequences=True)
def test_add_forge_request_create_duplicate(api_client, regular_user):
api_client.force_login(regular_user)
url = reverse("api-1-add-forge-request-create")
check_api_post_response(
api_client,
url,
data=ADD_FORGE_DATA_FORGE1,
status_code=201,
)
check_api_post_response(
api_client,
url,
data=ADD_FORGE_DATA_FORGE1,
status_code=409,
)
requests = Request.objects.all()
assert len(requests) == 1
@pytest.mark.django_db(transaction=True, reset_sequences=True)
def test_add_forge_request_update_anonymous_user(api_client):
url = reverse("api-1-add-forge-request-update", url_args={"id": 1})
check_api_post_response(api_client, url, status_code=403)
@pytest.mark.django_db(transaction=True, reset_sequences=True)
def test_add_forge_request_update_regular_user(api_client, regular_user):
api_client.force_login(regular_user)
url = reverse("api-1-add-forge-request-update", url_args={"id": 1})
check_api_post_response(api_client, url, status_code=403)
@pytest.mark.django_db(transaction=True, reset_sequences=True)
def test_add_forge_request_update_non_existent(api_client, add_forge_moderator):
api_client.force_login(add_forge_moderator)
url = reverse("api-1-add-forge-request-update", url_args={"id": 1})
check_api_post_response(api_client, url, status_code=400)
def create_add_forge_request(api_client, regular_user, data=ADD_FORGE_DATA_FORGE1):
api_client.force_login(regular_user)
url = reverse("api-1-add-forge-request-create")
return check_api_post_response(
api_client,
url,
data=data,
status_code=201,
)
@pytest.mark.django_db(transaction=True, reset_sequences=True)
def test_add_forge_request_update_empty(api_client, regular_user, add_forge_moderator):
create_add_forge_request(api_client, regular_user)
api_client.force_login(add_forge_moderator)
url = reverse("api-1-add-forge-request-update", url_args={"id": 1})
check_api_post_response(api_client, url, status_code=400)
@pytest.mark.django_db(transaction=True, reset_sequences=True)
def test_add_forge_request_update_missing_field(
api_client, regular_user, add_forge_moderator
):
create_add_forge_request(api_client, regular_user)
api_client.force_login(add_forge_moderator)
url = reverse("api-1-add-forge-request-update", url_args={"id": 1})
check_api_post_response(api_client, url, data={}, status_code=400)
check_api_post_response(
api_client, url, data={"new_status": "REJECTED"}, status_code=400
)
@pytest.mark.django_db(transaction=True, reset_sequences=True)
def test_add_forge_request_update(api_client, regular_user, add_forge_moderator):
create_add_forge_request(api_client, regular_user)
api_client.force_login(add_forge_moderator)
url = reverse("api-1-add-forge-request-update", url_args={"id": 1})
check_api_post_response(
api_client, url, data={"text": "updating request"}, status_code=200
)
check_api_post_response(
api_client,
url,
data={"new_status": "REJECTED", "text": "request rejected"},
status_code=200,
)
@pytest.mark.django_db(transaction=True, reset_sequences=True)
def test_add_forge_request_update_invalid_new_status(
api_client, regular_user, add_forge_moderator
):
create_add_forge_request(api_client, regular_user)
api_client.force_login(add_forge_moderator)
url = reverse("api-1-add-forge-request-update", url_args={"id": 1})
check_api_post_response(
api_client,
url,
data={"new_status": "ACCEPTED", "text": "request accepted"},
status_code=400,
)
@pytest.mark.django_db(transaction=True, reset_sequences=True)
def test_add_forge_request_update_status_concurrent(
api_client, regular_user, add_forge_moderator, mocker
):
_block_while_testing = mocker.patch(
"swh.web.add_forge_now.api_views._block_while_testing"
)
_block_while_testing.side_effect = lambda: time.sleep(1)
create_add_forge_request(api_client, regular_user)
api_client.force_login(add_forge_moderator)
url = reverse("api-1-add-forge-request-update", url_args={"id": 1})
worker_ended = False
def worker():
nonlocal worker_ended
check_api_post_response(
api_client,
url,
data={"new_status": "WAITING_FOR_FEEDBACK", "text": "waiting for message"},
status_code=200,
)
worker_ended = True
# this thread will first modify the request status to WAITING_FOR_FEEDBACK
thread = threading.Thread(target=worker)
thread.start()
# the other thread (slower) will attempt to modify the request status to REJECTED
# but it will not be allowed as the first faster thread already modified it
# and REJECTED state can not be reached from WAITING_FOR_FEEDBACK one
time.sleep(0.5)
check_api_post_response(
api_client,
url,
data={"new_status": "REJECTED", "text": "request accepted"},
status_code=400,
)
thread.join()
assert worker_ended
@pytest.mark.django_db(transaction=True, reset_sequences=True)
def test_add_forge_request_list_anonymous(api_client, regular_user):
url = reverse("api-1-add-forge-request-list")
resp = check_api_get_responses(api_client, url, status_code=200)
assert resp.data == []
create_add_forge_request(api_client, regular_user)
resp = check_api_get_responses(api_client, url, status_code=200)
add_forge_request = {
"forge_url": ADD_FORGE_DATA_FORGE1["forge_url"],
"forge_type": ADD_FORGE_DATA_FORGE1["forge_type"],
"status": "PENDING",
"submission_date": resp.data[0]["submission_date"],
"id": resp.data[0]["id"],
}
assert resp.data == [add_forge_request]
create_add_forge_request(api_client, regular_user, data=ADD_FORGE_DATA_FORGE2)
resp = check_api_get_responses(api_client, url, status_code=200)
other_forge_request = {
"forge_url": ADD_FORGE_DATA_FORGE2["forge_url"],
"forge_type": ADD_FORGE_DATA_FORGE2["forge_type"],
"status": "PENDING",
"submission_date": resp.data[0]["submission_date"],
"id": resp.data[0]["id"],
}
assert resp.data == [other_forge_request, add_forge_request]
@pytest.mark.django_db(transaction=True, reset_sequences=True)
def test_add_forge_request_list_moderator(
api_client, regular_user, add_forge_moderator
):
url = reverse("api-1-add-forge-request-list")
create_add_forge_request(api_client, regular_user)
create_add_forge_request(api_client, regular_user, data=ADD_FORGE_DATA_FORGE2)
api_client.force_login(add_forge_moderator)
resp = check_api_get_responses(api_client, url, status_code=200)
add_forge_request = {
**ADD_FORGE_DATA_FORGE1,
"status": "PENDING",
"submission_date": resp.data[1]["submission_date"],
"submitter_name": regular_user.username,
"submitter_email": regular_user.email,
"last_moderator": resp.data[1]["last_moderator"],
"last_modified_date": resp.data[1]["last_modified_date"],
"id": resp.data[1]["id"],
"inbound_email_address": inbound_email_for_pk(resp.data[1]["id"]),
"forge_domain": urlparse(ADD_FORGE_DATA_FORGE1["forge_url"]).netloc,
}
other_forge_request = {
**ADD_FORGE_DATA_FORGE2,
"status": "PENDING",
"submission_date": resp.data[0]["submission_date"],
"submitter_name": regular_user.username,
"submitter_email": regular_user.email,
"last_moderator": resp.data[0]["last_moderator"],
"last_modified_date": resp.data[0]["last_modified_date"],
"id": resp.data[0]["id"],
"inbound_email_address": inbound_email_for_pk(resp.data[0]["id"]),
"forge_domain": urlparse(ADD_FORGE_DATA_FORGE2["forge_url"]).netloc,
}
assert resp.data == [other_forge_request, add_forge_request]
@pytest.mark.django_db(transaction=True, reset_sequences=True)
def test_add_forge_request_list_pagination(
api_client, regular_user, api_request_factory
):
create_add_forge_request(api_client, regular_user)
create_add_forge_request(api_client, regular_user, data=ADD_FORGE_DATA_FORGE2)
url = reverse("api-1-add-forge-request-list", query_params={"per_page": 1})
resp = check_api_get_responses(api_client, url, 200)
assert len(resp.data) == 1
request = api_request_factory.get(url)
next_url = reverse(
"api-1-add-forge-request-list",
query_params={"page": 2, "per_page": 1},
request=request,
)
assert resp["Link"] == f'<{next_url}>; rel="next"'
resp = check_api_get_responses(api_client, next_url, 200)
assert len(resp.data) == 1
prev_url = reverse(
"api-1-add-forge-request-list",
query_params={"page": 1, "per_page": 1},
request=request,
)
assert resp["Link"] == f'<{prev_url}>; rel="previous"'
@pytest.mark.django_db(transaction=True, reset_sequences=True)
def test_add_forge_request_list_submitter_filtering(
api_client, regular_user, regular_user2
):
create_add_forge_request(api_client, regular_user)
create_add_forge_request(api_client, regular_user2, data=ADD_FORGE_DATA_FORGE2)
api_client.force_login(regular_user)
url = reverse(
"api-1-add-forge-request-list", query_params={"user_requests_only": 1}
)
resp = check_api_get_responses(api_client, url, status_code=200)
assert len(resp.data) == 1
@pytest.mark.django_db(transaction=True, reset_sequences=True)
def test_add_forge_request_get(api_client, regular_user, add_forge_moderator):
resp = create_add_forge_request(api_client, regular_user)
submission_date = resp.data["submission_date"]
url = reverse("api-1-add-forge-request-update", url_args={"id": 1})
api_client.force_login(add_forge_moderator)
check_api_post_response(
api_client,
url,
data={"new_status": "WAITING_FOR_FEEDBACK", "text": "waiting for message"},
status_code=200,
)
api_client.logout()
url = reverse("api-1-add-forge-request-get", url_args={"id": 1})
resp = check_api_get_responses(api_client, url, status_code=200)
assert resp.data == {
"request": {
"forge_url": ADD_FORGE_DATA_FORGE1["forge_url"],
"forge_type": ADD_FORGE_DATA_FORGE1["forge_type"],
"id": 1,
"status": "WAITING_FOR_FEEDBACK",
"submission_date": submission_date,
},
"history": [
{
"id": 1,
"actor_role": "SUBMITTER",
"date": resp.data["history"][0]["date"],
"new_status": "PENDING",
},
{
"id": 2,
"actor_role": "MODERATOR",
"date": resp.data["history"][1]["date"],
"new_status": "WAITING_FOR_FEEDBACK",
},
],
}
@pytest.mark.django_db(transaction=True, reset_sequences=True)
def test_add_forge_request_get_moderator(api_client, regular_user, add_forge_moderator):
resp = create_add_forge_request(api_client, regular_user)
submission_date = resp.data["submission_date"]
url = reverse("api-1-add-forge-request-update", url_args={"id": 1})
api_client.force_login(add_forge_moderator)
check_api_post_response(
api_client,
url,
data={"new_status": "WAITING_FOR_FEEDBACK", "text": "waiting for message"},
status_code=200,
)
url = reverse("api-1-add-forge-request-get", url_args={"id": 1})
resp = check_api_get_responses(api_client, url, status_code=200)
resp.data["history"] = [dict(history_item) for history_item in resp.data["history"]]
assert resp.data == {
"request": {
**ADD_FORGE_DATA_FORGE1,
"id": 1,
"status": "WAITING_FOR_FEEDBACK",
"submission_date": submission_date,
"submitter_name": regular_user.username,
"submitter_email": regular_user.email,
"last_moderator": add_forge_moderator.username,
"last_modified_date": resp.data["history"][1]["date"],
"inbound_email_address": inbound_email_for_pk(1),
"forge_domain": urlparse(ADD_FORGE_DATA_FORGE1["forge_url"]).netloc,
},
"history": [
{
"id": 1,
"text": "",
"actor": regular_user.username,
"actor_role": "SUBMITTER",
"date": resp.data["history"][0]["date"],
"new_status": "PENDING",
"message_source_url": None,
},
{
"id": 2,
"text": "waiting for message",
"actor": add_forge_moderator.username,
"actor_role": "MODERATOR",
"date": resp.data["history"][1]["date"],
"new_status": "WAITING_FOR_FEEDBACK",
"message_source_url": None,
},
],
}
@pytest.mark.django_db(transaction=True, reset_sequences=True)
def test_add_forge_request_get_moderator_message_source(
api_client, regular_user, add_forge_moderator
):
resp = create_add_forge_request(api_client, regular_user)
rh = RequestHistory(
request=Request.objects.get(pk=resp.data["id"]),
new_status="WAITING_FOR_FEEDBACK",
text="waiting for message",
actor=add_forge_moderator.username,
actor_role="MODERATOR",
message_source=b"test with a message source",
)
rh.save()
api_client.force_login(add_forge_moderator)
url = reverse("api-1-add-forge-request-get", url_args={"id": resp.data["id"]})
resp = check_api_get_responses(api_client, url, status_code=200)
resp.data["history"] = [dict(history_item) for history_item in resp.data["history"]]
# Check that the authentified moderator can't urlhack non-existent message sources
assert resp.data["history"][0]["message_source_url"] is None
empty_message_url = reverse(
"forge-add-message-source", url_args={"id": resp.data["history"][0]["id"]}
)
check_http_get_response(api_client, empty_message_url, status_code=404)
# Check that the authentified moderator can't urlhack non-existent message sources
non_existent_message_url = reverse(
"forge-add-message-source", url_args={"id": 9001}
)
check_http_get_response(api_client, non_existent_message_url, status_code=404)
# Check that the authentified moderator can access the message source when the url is
# given
message_source_url = resp.data["history"][-1]["message_source_url"]
assert message_source_url is not None
message_source_resp = check_http_get_response(
api_client, message_source_url, status_code=200, content_type="text/email"
)
# Check that the message source shows up as an attachment
assert message_source_resp.content == rh.message_source
disposition = message_source_resp["Content-Disposition"]
assert disposition.startswith("attachment; filename=")
assert disposition.endswith('.eml"')
# Check that a regular user can't access message sources
api_client.force_login(regular_user)
check_http_get_response(api_client, message_source_url, status_code=302)
api_client.force_login(add_forge_moderator)
@pytest.mark.django_db(transaction=True, reset_sequences=True)
def test_add_forge_request_get_invalid(api_client):
url = reverse("api-1-add-forge-request-get", url_args={"id": 3})
check_api_get_responses(api_client, url, status_code=400)

File Metadata

Mime Type
text/x-python
Expires
Wed, Jun 4, 7:24 PM (4 d, 21 h ago)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
3393177

Event Timeline