test_add_forge_now.py
No OneTemporary
Actions

Size

12 KB

Subscribers

None

test_add_forge_now.py
View Options

	# 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 urllib.parse import urlencode

	import iso8601
	import pytest

	from swh.web.add_forge_now.models import Request
	from swh.web.api.views.add_forge_now import MODERATOR_ROLE
	from swh.web.common.utils import reverse
	from swh.web.tests.utils import (
	check_api_get_responses,
	check_api_post_response,
	check_http_post_response,
	create_django_permission,
	)


	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 = {
	"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",
	}

	ADD_OTHER_FORGE_DATA = {
	"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",
	}


	@pytest.mark.django_db(transaction=True, reset_sequences=True)
	def test_add_forge_request_create_success(api_client, regular_user):
	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)

	assert resp.data == {
	**ADD_FORGE_DATA,
	"id": 1,
	"status": "PENDING",
	"submission_date": resp.data["submission_date"],
	"submitter_name": regular_user.username,
	"submitter_email": regular_user.email,
	}

	assert date_before < iso8601.parse_date(resp.data["submission_date"]) < date_after

	request = Request.objects.all()[0]

	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),
	status_code=201,
	)

	date_after = datetime.datetime.now(tz=datetime.timezone.utc)

	assert resp.data == {
	**ADD_FORGE_DATA,
	"id": 1,
	"status": "PENDING",
	"submission_date": resp.data["submission_date"],
	"submitter_name": regular_user.username,
	"submitter_email": regular_user.email,
	}

	assert date_before < iso8601.parse_date(resp.data["submission_date"]) < date_after

	request = Request.objects.all()[0]

	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_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, status_code=201,
	)
	check_api_post_response(
	api_client, url, data=ADD_FORGE_DATA, status_code=409,
	)

	requests = Request.objects.all()
	assert len(requests) == 1


	@pytest.fixture
	def moderator_user(regular_user2):
	regular_user2.user_permissions.add(create_django_permission(MODERATOR_ROLE))
	return regular_user2


	@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, moderator_user):
	api_client.force_login(moderator_user)
	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):
	api_client.force_login(regular_user)
	url = reverse("api-1-add-forge-request-create")
	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, moderator_user):
	_create_add_forge_request(api_client, regular_user)

	api_client.force_login(moderator_user)
	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, moderator_user
	):
	_create_add_forge_request(api_client, regular_user)

	api_client.force_login(moderator_user)
	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, moderator_user):
	_create_add_forge_request(api_client, regular_user)

	api_client.force_login(moderator_user)
	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, moderator_user
	):
	_create_add_forge_request(api_client, regular_user)

	api_client.force_login(moderator_user)
	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, moderator_user, mocker
	):

	_block_while_testing = mocker.patch(
	"swh.web.api.views.add_forge_now._block_while_testing"
	)
	_block_while_testing.side_effect = lambda: time.sleep(1)

	_create_add_forge_request(api_client, regular_user)

	api_client.force_login(moderator_user)
	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["forge_url"],
	"forge_type": ADD_FORGE_DATA["forge_type"],
	"status": "PENDING",
	"submission_date": resp.data[0]["submission_date"],
	}

	assert resp.data == [add_forge_request]

	_create_add_forge_request(api_client, regular_user, data=ADD_OTHER_FORGE_DATA)

	resp = check_api_get_responses(api_client, url, status_code=200)

	other_forge_request = {
	"forge_url": ADD_OTHER_FORGE_DATA["forge_url"],
	"forge_type": ADD_OTHER_FORGE_DATA["forge_type"],
	"status": "PENDING",
	"submission_date": resp.data[0]["submission_date"],
	}

	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, moderator_user):
	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_OTHER_FORGE_DATA)

	api_client.force_login(moderator_user)
	resp = check_api_get_responses(api_client, url, status_code=200)

	add_forge_request = {
	**ADD_FORGE_DATA,
	"status": "PENDING",
	"submission_date": resp.data[1]["submission_date"],
	"submitter_name": regular_user.username,
	"submitter_email": regular_user.email,
	"id": 1,
	}

	other_forge_request = {
	**ADD_OTHER_FORGE_DATA,
	"status": "PENDING",
	"submission_date": resp.data[0]["submission_date"],
	"submitter_name": regular_user.username,
	"submitter_email": regular_user.email,
	"id": 2,
	}

	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_OTHER_FORGE_DATA)

	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_OTHER_FORGE_DATA)

	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

File Metadata

Mime Type: text/x-python
Expires: Jul 4 2025, 6:07 PM (4 w, 6 d ago)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 3351244

test_add_forge_now.pyNo OneTemporaryActions

test_add_forge_now.pyView Options

File Metadata

Event Timeline

test_add_forge_now.py
No OneTemporary
Actions

test_add_forge_now.py
View Options