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.auth.django.utils import oidc_user_from_profile | ||||
from swh.web.auth.utils import SWH_AMBASSADOR_PERMISSION | |||||
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 | ||||
VISIT_TYPES = ("git", "svn", "hg") | |||||
PRIVILEGED_VISIT_TYPES = tuple(list(VISIT_TYPES) + ["bundle"]) | |||||
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 | ||||
def test_save_types_list_default(client): | |||||
"""Unprivileged listing should display default list of visit types. | |||||
""" | |||||
url = reverse("origin-save-types-list") | |||||
resp = check_http_get_response(client, url, status_code=200) | |||||
actual_response = resp.json() | |||||
assert set(actual_response) == set(VISIT_TYPES) | |||||
ardumont: so far... That won't see the user as an ambassador one...
```
> assert set… | |||||
@pytest.mark.django_db | |||||
def test_save_types_list_privileged(client, keycloak_oidc): | |||||
"""Privileged listing should display all visit types. | |||||
""" | |||||
keycloak_oidc.realm_permissions = [SWH_AMBASSADOR_PERMISSION] | |||||
client.login(code="", code_verifier="", redirect_uri="") | |||||
url = reverse("origin-save-types-list") | |||||
resp = check_http_get_response(client, url, status_code=200) | |||||
actual_response = resp.json() | |||||
assert set(actual_response) == set(PRIVILEGED_VISIT_TYPES) | |||||
@pytest.mark.django_db | @pytest.mark.django_db | ||||
Not Done Inline ActionsBecause "origin-save-types-list" is not a DRF view so the bearer token auth will not work here. You need to use the client fixture and login with client.login(code="", code_verifier="", redirect_uri="") anlambert: Because `"origin-save-types-list"` is not a DRF view so the bearer token auth will not work… | |||||
Done Inline Actionsah! ok Thanks. ardumont: ah! ok
I bypassed it so far using a mock but i don't like that much.
So i'll give your… | |||||
Done Inline Actionsit worked ;) ardumont: it worked ;) | |||||
def test_save_origin_requests_list(client, mocker, keycloak_oidc): | def test_save_origin_requests_list(client, mocker): | ||||
visit_types = ("git", "svn", "hg") | |||||
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, | ||||
visit_date=datetime.now(tz=timezone.utc) + timedelta(hours=1), | visit_date=datetime.now(tz=timezone.utc) + timedelta(hours=1), | ||||
loading_task_id=i, | loading_task_id=i, | ||||
loading_task_status=SAVE_TASK_SUCCEEDED, | loading_task_status=SAVE_TASK_SUCCEEDED, | ||||
) | ) | ||||
mock_scheduler = mocker.patch("swh.web.common.origin_save.scheduler") | mock_scheduler = mocker.patch("swh.web.common.origin_save.scheduler") | ||||
mock_scheduler.get_tasks.return_value = [] | mock_scheduler.get_tasks.return_value = [] | ||||
mock_scheduler.get_task_runs.return_value = [] | mock_scheduler.get_task_runs.return_value = [] | ||||
# retrieve all save requests in 3 pages, sorted in descending order | # retrieve all save requests in 3 pages, sorted in descending order | ||||
# of request creation | # of request creation | ||||
for i, visit_type in enumerate(reversed(visit_types)): | for i, visit_type in enumerate(reversed(VISIT_TYPES)): | ||||
url = reverse( | url = reverse( | ||||
"origin-save-requests-list", | "origin-save-requests-list", | ||||
url_args={"status": "all"}, | url_args={"status": "all"}, | ||||
query_params={ | query_params={ | ||||
"draw": i + 1, | "draw": i + 1, | ||||
"search[value]": "", | "search[value]": "", | ||||
"order[0][column]": "0", | "order[0][column]": "0", | ||||
"columns[0][name]": "request_date", | "columns[0][name]": "request_date", | ||||
"order[0][dir]": "desc", | "order[0][dir]": "desc", | ||||
"length": nb_origins_per_type, | "length": nb_origins_per_type, | ||||
"start": i * nb_origins_per_type, | "start": i * nb_origins_per_type, | ||||
}, | }, | ||||
) | ) | ||||
resp = check_http_get_response( | resp = check_http_get_response( | ||||
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"] == len(visit_types) * nb_origins_per_type | assert sors["recordsFiltered"] == len(VISIT_TYPES) * 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"]) | ||||
# retrieve save requests filtered by visit type in a single page | # retrieve save requests filtered by visit type in a single page | ||||
for i, visit_type in enumerate(reversed(visit_types)): | for i, visit_type in enumerate(reversed(VISIT_TYPES)): | ||||
url = reverse( | url = reverse( | ||||
"origin-save-requests-list", | "origin-save-requests-list", | ||||
url_args={"status": "all"}, | url_args={"status": "all"}, | ||||
query_params={ | query_params={ | ||||
"draw": i + 1, | "draw": i + 1, | ||||
"search[value]": visit_type, | "search[value]": visit_type, | ||||
"order[0][column]": "0", | "order[0][column]": "0", | ||||
"columns[0][name]": "request_date", | "columns[0][name]": "request_date", | ||||
"order[0][dir]": "desc", | "order[0][dir]": "desc", | ||||
"length": nb_origins_per_type, | "length": nb_origins_per_type, | ||||
"start": 0, | "start": 0, | ||||
}, | }, | ||||
) | ) | ||||
resp = check_http_get_response( | resp = check_http_get_response( | ||||
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 | @pytest.mark.django_db | ||||
def test_save_origin_requests_list_user_filter(client, mocker, keycloak_oidc): | def test_save_origin_requests_list_user_filter(client, mocker, keycloak_oidc): | ||||
# anonymous user created a save request | # anonymous user created a save request | ||||
▲ Show 20 Lines • Show All 48 Lines • Show Last 20 Lines |
so far... That won't see the user as an ambassador one...