Changeset View
Changeset View
Standalone 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 |