Differential D8800 Diff 31712 swh/web/tests/save_origin_webhooks/utils.py

Changeset View

Standalone View

swh/web/tests/save_origin_webhooks/utils.py

This file was added.

				# 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 typing import Any, Dict

				from swh.web.tests.helpers import check_api_post_responses
				from swh.web.utils import reverse


				def _django_http_headers(http_headers: Dict[str, Any]):
				return {f"HTTP_{k.upper().replace('-', '_')}": v for k, v in http_headers.items()}


				def origin_save_webhook_receiver_test(
				forge_type: str,
				http_headers: Dict[str, Any],
				payload: Dict[str, Any],
				expected_origin_url: str,
				expected_visit_type: str,
				api_client,
				swh_scheduler,
				):
				url = reverse(f"api-1-origin-save-webhook-{forge_type.lower()}")

				resp = check_api_post_responses(
				api_client,
				url,
				status_code=200,
				data=payload,
				**_django_http_headers(http_headers),
				)

				assert resp.data["origin_url"] == expected_origin_url
				assert resp.data["visit_type"] == expected_visit_type

				tasks = swh_scheduler.search_tasks(task_type=f"load-{expected_visit_type}")
				assert tasks
				task = dict(tasks[0].items())
				assert task["arguments"]["kwargs"]["url"] == expected_origin_url


				def origin_save_webhook_receiver_invalid_request_test(
				forge_type: str,
				http_headers: Dict[str, Any],
				payload: Dict[str, Any],
				api_client,
				):
				url = reverse(f"api-1-origin-save-webhook-{forge_type.lower()}")

				resp = check_api_post_responses(
				api_client,
				url,
				status_code=400,
				data=payload,
				**_django_http_headers(http_headers),
				)

				assert resp.data == {
				"exception": "BadInputExc",
				"reason": (
				f"POST request has not been sent by a {forge_type} webhook "
				"and has not been processed."
				),
				}


				def origin_save_webhook_receiver_invalid_event_test(
				forge_type: str,
				http_headers: Dict[str, Any],
				payload: Dict[str, Any],
				api_client,
				):
				url = reverse(f"api-1-origin-save-webhook-{forge_type.lower()}")

				resp = check_api_post_responses(
				api_client,
				url,
				status_code=400,
				data=payload,
				**_django_http_headers(http_headers),
				)

				assert resp.data == {
				"exception": "BadInputExc",
				"reason": (
				f"Event sent by {forge_type} webhook is not a push one, request has "
				"not been processed."
				),
				}


				def origin_save_webhook_receiver_invalid_content_type_test(
				forge_type: str,
				http_headers: Dict[str, Any],
				payload: Dict[str, Any],
				api_client,
				):
				url = reverse(f"api-1-origin-save-webhook-{forge_type.lower()}")

				bad_content_type = "application/x-www-form-urlencoded"
				http_headers["Content-Type"] = bad_content_type

				resp = check_api_post_responses(
				api_client,
				url,
				status_code=400,
				data=payload,
				**_django_http_headers(http_headers),
				)

				assert resp.data == {
				"exception": "BadInputExc",
				"reason": (
				f"Invalid content type '{bad_content_type}' for the POST request sent by "
				f"{forge_type} webhook, it should be 'application/json'."
				),
				}


				def origin_save_webhook_receiver_no_repo_url_test(
				forge_type: str,
				http_headers: Dict[str, Any],
				payload: Dict[str, Any],
				api_client,
				):
				url = reverse(f"api-1-origin-save-webhook-{forge_type.lower()}")

				resp = check_api_post_responses(
				api_client,
				url,
				status_code=400,
				data=payload,
				**_django_http_headers(http_headers),
				)

				assert resp.data == {
				"exception": "BadInputExc",
				"reason": (
				f"Repository URL could not be extracted from {forge_type} webhook payload."
				),
				}