Changeset View
Changeset View
Standalone View
Standalone View
swh/deposit/tests/conftest.py
# Copyright (C) 2019-2020 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 General Public License version 3, or any later version | # License: GNU 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 | ||||
import base64 | import base64 | ||||
from functools import partial | from functools import partial | ||||
from io import BytesIO | from io import BytesIO | ||||
import os | import os | ||||
import re | import re | ||||
from typing import Mapping | from typing import Mapping | ||||
from django.test.utils import setup_databases # type: ignore | from django.test.utils import setup_databases # type: ignore | ||||
from django.urls import reverse | from django.urls import reverse | ||||
import psycopg2 | import psycopg2 | ||||
from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT | from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT | ||||
import pytest | import pytest | ||||
from rest_framework import status | from rest_framework import status | ||||
from rest_framework.test import APIClient | from rest_framework.test import APIClient | ||||
import yaml | import yaml | ||||
from swh.core.config import read | from swh.core.config import read | ||||
from swh.core.pytest_plugin import get_response_cb | from swh.core.pytest_plugin import get_response_cb | ||||
from swh.deposit.auth import KeycloakOpenIDConnectDirectGrant | |||||
from swh.deposit.config import ( | from swh.deposit.config import ( | ||||
COL_IRI, | COL_IRI, | ||||
DEPOSIT_STATUS_DEPOSITED, | DEPOSIT_STATUS_DEPOSITED, | ||||
DEPOSIT_STATUS_LOAD_FAILURE, | DEPOSIT_STATUS_LOAD_FAILURE, | ||||
DEPOSIT_STATUS_LOAD_SUCCESS, | DEPOSIT_STATUS_LOAD_SUCCESS, | ||||
DEPOSIT_STATUS_PARTIAL, | DEPOSIT_STATUS_PARTIAL, | ||||
DEPOSIT_STATUS_REJECTED, | DEPOSIT_STATUS_REJECTED, | ||||
DEPOSIT_STATUS_VERIFIED, | DEPOSIT_STATUS_VERIFIED, | ||||
▲ Show 20 Lines • Show All 51 Lines • ▼ Show 20 Lines | |||||
@pytest.fixture() | @pytest.fixture() | ||||
def deposit_config(swh_scheduler_config, swh_storage_backend_config): | def deposit_config(swh_scheduler_config, swh_storage_backend_config): | ||||
return { | return { | ||||
"max_upload_size": 500, | "max_upload_size": 500, | ||||
"extraction_dir": "/tmp/swh-deposit/test/extraction-dir", | "extraction_dir": "/tmp/swh-deposit/test/extraction-dir", | ||||
"checks": False, | "checks": False, | ||||
"scheduler": {"cls": "local", **swh_scheduler_config,}, | "scheduler": {"cls": "local", **swh_scheduler_config,}, | ||||
"storage_metadata": swh_storage_backend_config, | "storage_metadata": swh_storage_backend_config, | ||||
"keycloak": { | |||||
"server_url": "https://auth.swh.org/SWHTest", | |||||
"realm_name": "SWHTest", | |||||
}, | |||||
} | } | ||||
@pytest.fixture() | @pytest.fixture() | ||||
def deposit_config_path(tmp_path, monkeypatch, deposit_config): | def deposit_config_path(tmp_path, monkeypatch, deposit_config): | ||||
conf_path = os.path.join(tmp_path, "deposit.yml") | conf_path = os.path.join(tmp_path, "deposit.yml") | ||||
with open(conf_path, "w") as f: | with open(conf_path, "w") as f: | ||||
f.write(yaml.dump(deposit_config)) | f.write(yaml.dump(deposit_config)) | ||||
▲ Show 20 Lines • Show All 97 Lines • ▼ Show 20 Lines | def _create_deposit_user(db, collection, user_data): | ||||
from swh.deposit.models import DepositClient | from swh.deposit.models import DepositClient | ||||
try: | try: | ||||
user = DepositClient._default_manager.get(username=user_data["username"]) | user = DepositClient._default_manager.get(username=user_data["username"]) | ||||
except DepositClient.DoesNotExist: | except DepositClient.DoesNotExist: | ||||
user = DepositClient._default_manager.create_user( | user = DepositClient._default_manager.create_user( | ||||
username=user_data["username"], | username=user_data["username"], | ||||
email=user_data["email"], | email=user_data["email"], | ||||
password=user_data["password"], | # password=user_data["password"], | ||||
provider_url=user_data["provider_url"], | provider_url=user_data["provider_url"], | ||||
domain=user_data["domain"], | domain=user_data["domain"], | ||||
) | ) | ||||
user.collections = [collection.id] | user.collections = [collection.id] | ||||
user.save() | user.save() | ||||
anlambert: Use `DepositClient.objects.get_or_create` instead. | |||||
Done Inline Actionsgood idea. ardumont: good idea. | |||||
return user | return user | ||||
@pytest.fixture | @pytest.fixture | ||||
def deposit_user(db, deposit_collection): | def deposit_user(db, deposit_collection): | ||||
return _create_deposit_user(db, deposit_collection, TEST_USER) | return _create_deposit_user(db, deposit_collection, TEST_USER) | ||||
@pytest.fixture | @pytest.fixture | ||||
def deposit_another_user(db, deposit_another_collection): | def deposit_another_user(db, deposit_another_collection): | ||||
return _create_deposit_user(db, deposit_another_collection, ANOTHER_TEST_USER) | return _create_deposit_user(db, deposit_another_collection, ANOTHER_TEST_USER) | ||||
@pytest.fixture | @pytest.fixture | ||||
def client(): | def client(): | ||||
"""Override pytest-django one which does not work for djangorestframework. | """Override pytest-django one which does not work for djangorestframework. | ||||
""" | """ | ||||
return APIClient() # <- drf's client | return APIClient() # <- drf's client | ||||
def _create_authenticated_client(client, user, user_data): | def _create_authenticated_client(client, user, user_data): | ||||
"""Returned a logged client | """Returned a logged client | ||||
This also patched the client instance to keep a reference on the associated | This also patched the client instance to keep a reference on the associated | ||||
Not Done Inline ActionsThis setup could be moved to swh-auth later as swh-web will also need to do such kind of mocking. anlambert: This setup could be moved to `swh-auth` later as `swh-web` will also need to do such kind of… | |||||
deposit_user. | deposit_user. | ||||
""" | """ | ||||
_token = "%s:%s" % (user.username, user_data["password"]) | _token = "%s:%s" % (user.username, user_data["password"]) | ||||
token = base64.b64encode(_token.encode("utf-8")) | token = base64.b64encode(_token.encode("utf-8")) | ||||
authorization = "Basic %s" % token.decode("utf-8") | authorization = "Basic %s" % token.decode("utf-8") | ||||
client.credentials(HTTP_AUTHORIZATION=authorization) | client.credentials(HTTP_AUTHORIZATION=authorization) | ||||
client.deposit_client = user | client.deposit_client = user | ||||
yield client | yield client | ||||
client.logout() | client.logout() | ||||
@pytest.fixture | @pytest.fixture | ||||
def authenticated_client(client, deposit_user): | def keycloak_access(): | ||||
return KeycloakOpenIDConnectDirectGrant( | |||||
"https://auth.swh.org/SWHTest", "SWHTest", "test-swh-deposit" | |||||
) | |||||
@pytest.fixture | |||||
Done Inline Actionss/Returned/Return/ anlambert: s/Returned/Return/ | |||||
def keycloak_authorize_access(keycloak_access, requests_mock): | |||||
response = { | |||||
"access_token": "2YotnFZFEjr1zCsicMWpAA", | |||||
"token_type": "example", | |||||
"expires_in": 3600, | |||||
"refresh_token": "tGzv3JOkF0XG5Qx2TlKWIA", | |||||
"example_parameter": "example_value", | |||||
} | |||||
requests_mock.post(keycloak_access.token_url(), status_code=200, json=response) | |||||
@pytest.fixture | |||||
def authenticated_client(client, deposit_user, keycloak_authorize_access): | |||||
yield from _create_authenticated_client(client, deposit_user, TEST_USER) | yield from _create_authenticated_client(client, deposit_user, TEST_USER) | ||||
@pytest.fixture | @pytest.fixture | ||||
def another_authenticated_client(deposit_another_user): | def another_authenticated_client(deposit_another_user, keycloak_authorize_access): | ||||
client = APIClient() | client = APIClient() | ||||
yield from _create_authenticated_client( | yield from _create_authenticated_client( | ||||
client, deposit_another_user, ANOTHER_TEST_USER | client, deposit_another_user, ANOTHER_TEST_USER | ||||
) | ) | ||||
@pytest.fixture | @pytest.fixture | ||||
def sample_archive(tmp_path): | def sample_archive(tmp_path): | ||||
▲ Show 20 Lines • Show All 218 Lines • Show Last 20 Lines |
Use DepositClient.objects.get_or_create instead.