Changeset View
Changeset View
Standalone View
Standalone View
swh/web/tests/misc/test_origin_save.py
# Copyright (C) 2019-2021 The Software Heritage developers | # Copyright (C) 2019-2021 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 datetime import datetime, timedelta, timezone | from datetime import datetime, timedelta, timezone | ||||
import json | import json | ||||
import pytest | import pytest | ||||
from swh.auth.django.utils import oidc_user_from_profile | |||||
from swh.web.common.models import SaveOriginRequest | from swh.web.common.models import SaveOriginRequest | ||||
from swh.web.common.origin_save import SAVE_REQUEST_ACCEPTED, SAVE_TASK_SUCCEEDED | from swh.web.common.origin_save import SAVE_REQUEST_ACCEPTED, SAVE_TASK_SUCCEEDED | ||||
from swh.web.common.utils import reverse | from swh.web.common.utils import reverse | ||||
from swh.web.tests.utils import check_http_get_response | from swh.web.tests.utils import check_http_get_response | ||||
def test_old_save_url_redirection(client): | def test_old_save_url_redirection(client): | ||||
url = reverse("browse-origin-save") | url = reverse("browse-origin-save") | ||||
redirect_url = reverse("origin-save") | redirect_url = reverse("origin-save") | ||||
resp = check_http_get_response(client, url, status_code=302) | resp = check_http_get_response(client, url, status_code=302) | ||||
assert resp["location"] == redirect_url | assert resp["location"] == redirect_url | ||||
@pytest.mark.django_db | @pytest.mark.django_db | ||||
def test_save_origin_requests_list(client, mocker): | def test_save_origin_requests_list(client, mocker, keycloak_oidc): | ||||
visit_types = ("git", "svn", "hg") | visit_types = ("git", "svn", "hg") | ||||
ardumont: I did not immediately get what you added at the end (connecting the
user and filtering only on… | |||||
Done Inline ActionsIn fact I was just lazy to write a new test function but I should have. anlambert: In fact I was just lazy to write a new test function but I should have. | |||||
nb_origins_per_type = 10 | nb_origins_per_type = 10 | ||||
for visit_type in visit_types: | for visit_type in visit_types: | ||||
for i in range(nb_origins_per_type): | for i in range(nb_origins_per_type): | ||||
SaveOriginRequest.objects.create( | SaveOriginRequest.objects.create( | ||||
request_date=datetime.now(tz=timezone.utc), | request_date=datetime.now(tz=timezone.utc), | ||||
visit_type=visit_type, | visit_type=visit_type, | ||||
origin_url=f"https://{visit_type}.example.org/project{i}", | origin_url=f"https://{visit_type}.example.org/project{i}", | ||||
status=SAVE_REQUEST_ACCEPTED, | status=SAVE_REQUEST_ACCEPTED, | ||||
▲ Show 20 Lines • Show All 53 Lines • ▼ Show 20 Lines | for i, visit_type in enumerate(reversed(visit_types)): | ||||
client, url, status_code=200, content_type="application/json" | client, url, status_code=200, content_type="application/json" | ||||
) | ) | ||||
sors = json.loads(resp.content.decode("utf-8")) | sors = json.loads(resp.content.decode("utf-8")) | ||||
assert sors["draw"] == i + 1 | assert sors["draw"] == i + 1 | ||||
assert sors["recordsFiltered"] == nb_origins_per_type | assert sors["recordsFiltered"] == nb_origins_per_type | ||||
assert sors["recordsTotal"] == len(visit_types) * nb_origins_per_type | assert sors["recordsTotal"] == len(visit_types) * nb_origins_per_type | ||||
assert len(sors["data"]) == nb_origins_per_type | assert len(sors["data"]) == nb_origins_per_type | ||||
assert all(d["visit_type"] == visit_type for d in sors["data"]) | assert all(d["visit_type"] == visit_type for d in sors["data"]) | ||||
@pytest.mark.django_db | |||||
def test_save_origin_requests_list_user_filter(client, mocker, keycloak_oidc): | |||||
# anonymous user created a save request | |||||
sor = SaveOriginRequest.objects.create( | |||||
request_date=datetime.now(tz=timezone.utc), | |||||
visit_type="svn", | |||||
origin_url="https://svn.example.org/user/project", | |||||
status=SAVE_REQUEST_ACCEPTED, | |||||
visit_date=datetime.now(tz=timezone.utc) + timedelta(hours=1), | |||||
loading_task_id=1, | |||||
loading_task_status=SAVE_TASK_SUCCEEDED, | |||||
) | |||||
# authenticated user created a save request | |||||
user = oidc_user_from_profile(keycloak_oidc, keycloak_oidc.login()) | |||||
client.login(code="", code_verifier="", redirect_uri="") | |||||
sor = SaveOriginRequest.objects.create( | |||||
request_date=datetime.now(tz=timezone.utc), | |||||
visit_type="git", | |||||
origin_url="https://git.example.org/user/project", | |||||
status=SAVE_REQUEST_ACCEPTED, | |||||
visit_date=datetime.now(tz=timezone.utc) + timedelta(hours=1), | |||||
loading_task_id=2, | |||||
loading_task_status=SAVE_TASK_SUCCEEDED, | |||||
user_ids=f'"{user.id}"', | |||||
) | |||||
# filter save requests according to user id | |||||
url = reverse( | |||||
"origin-save-requests-list", | |||||
url_args={"status": "all"}, | |||||
query_params={ | |||||
"draw": 1, | |||||
"search[value]": "", | |||||
"order[0][column]": "0", | |||||
"columns[0][name]": "request_date", | |||||
"order[0][dir]": "desc", | |||||
"length": 10, | |||||
"start": "0", | |||||
"user_requests_only": "1", | |||||
}, | |||||
) | |||||
resp = check_http_get_response( | |||||
client, url, status_code=200, content_type="application/json" | |||||
) | |||||
sors = json.loads(resp.content.decode("utf-8")) | |||||
assert sors["recordsFiltered"] == 1 | |||||
assert sors["recordsTotal"] == 2 | |||||
assert sors["data"][0] == sor.to_dict() |
I did not immediately get what you added at the end (connecting the
user and filtering only on their input).
Might be describing a bit the test would be nice?