Differential D8889 Diff 32101 swh/web/save_origin_webhooks/tests/utils.py

Changeset View

Standalone View

swh/web/save_origin_webhooks/tests/utils.py

# Copyright (C) 2022 The Software Heritage developers		# Copyright (C) 2022 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution		# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version		# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information		# See top-level LICENSE file for more information

from typing import Any, Dict		from typing import Any, Dict

		from swh.web.save_code_now.models import SaveOriginRequest
from swh.web.tests.helpers import check_api_post_responses		from swh.web.tests.helpers import check_api_post_responses
from swh.web.utils import reverse		from swh.web.utils import reverse


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


Show All 19 Lines	):
assert resp.data["origin_url"] == expected_origin_url		assert resp.data["origin_url"] == expected_origin_url
assert resp.data["visit_type"] == expected_visit_type		assert resp.data["visit_type"] == expected_visit_type

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

		request = SaveOriginRequest.objects.get(
		origin_url=expected_origin_url, visit_type=expected_visit_type
		)
		assert request.from_webhook


def origin_save_webhook_receiver_invalid_request_test(		def origin_save_webhook_receiver_invalid_request_test(
forge_type: str,		forge_type: str,
http_headers: Dict[str, Any],		http_headers: Dict[str, Any],
payload: Dict[str, Any],		payload: Dict[str, Any],
api_client,		api_client,
):		):
url = reverse(f"api-1-origin-save-webhook-{forge_type.lower()}")		url = reverse(f"api-1-origin-save-webhook-{forge_type.lower()}")
▲ Show 20 Lines • Show All 119 Lines • Show Last 20 Lines