Changeset View
Changeset View
Standalone View
Standalone View
swh/deposit/tests/conftest.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 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 copy import deepcopy | |||||
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 TYPE_CHECKING, Dict, Mapping | ||||
from django.test.utils import setup_databases # type: ignore | from django.test.utils import setup_databases # type: ignore | ||||
from django.urls import reverse_lazy as reverse | from django.urls import reverse_lazy as 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.auth.pytest_plugin import keycloak_mock_factory | |||||
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 DEPOSIT_PERMISSION | |||||
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, | ||||
SE_IRI, | SE_IRI, | ||||
setup_django_for, | setup_django_for, | ||||
) | ) | ||||
from swh.deposit.parsers import parse_xml | from swh.deposit.parsers import parse_xml | ||||
from swh.deposit.tests.common import ( | from swh.deposit.tests.common import ( | ||||
create_arborescence_archive, | create_arborescence_archive, | ||||
post_archive, | post_archive, | ||||
post_atom, | post_atom, | ||||
) | ) | ||||
from swh.model.hashutil import hash_to_bytes | from swh.model.hashutil import hash_to_bytes | ||||
from swh.model.identifiers import CoreSWHID, ObjectType, QualifiedSWHID | from swh.model.identifiers import CoreSWHID, ObjectType, QualifiedSWHID | ||||
from swh.scheduler import get_scheduler | from swh.scheduler import get_scheduler | ||||
if TYPE_CHECKING: | |||||
from swh.deposit.models import Deposit, DepositClient, DepositCollection | |||||
# mypy is asked to ignore the import statement above because setup_databases | # mypy is asked to ignore the import statement above because setup_databases | ||||
# is not part of the d.t.utils.__all__ variable. | # is not part of the d.t.utils.__all__ variable. | ||||
USERNAME = "test" | |||||
EMAIL = "test@example.org" | |||||
COLLECTION = "test" | |||||
TEST_USER = { | TEST_USER = { | ||||
"username": "test", | "username": USERNAME, | ||||
"password": "password", | "password": "", | ||||
"email": "test@example.org", | "email": EMAIL, | ||||
"provider_url": "https://hal-test.archives-ouvertes.fr/", | "provider_url": "https://hal-test.archives-ouvertes.fr/", | ||||
"domain": "archives-ouvertes.fr/", | "domain": "archives-ouvertes.fr/", | ||||
"collection": {"name": "test"}, | "collection": {"name": COLLECTION}, | ||||
} | } | ||||
USER_INFO = { | |||||
"name": USERNAME, | |||||
"email": EMAIL, | |||||
"email_verified": False, | |||||
"family_name": "", | |||||
"given_name": "", | |||||
"groups": [], | |||||
"preferred_username": USERNAME, | |||||
"sub": "ffffffff-bbbb-4444-aaaa-14f61e6b7200", | |||||
} | |||||
ANOTHER_TEST_USER = { | USERNAME2 = "test2" | ||||
"username": "test2", | EMAIL2 = "test@example.org" | ||||
"password": "password2", | COLLECTION2 = "another-collection" | ||||
"email": "test@example2.org", | |||||
TEST_USER2 = { | |||||
"username": USERNAME2, | |||||
"password": "", | |||||
"email": EMAIL2, | |||||
"provider_url": "https://hal-test.archives-ouvertes.example/", | "provider_url": "https://hal-test.archives-ouvertes.example/", | ||||
"domain": "archives-ouvertes.example/", | "domain": "archives-ouvertes.example/", | ||||
"collection": {"name": "another-collection"}, | "collection": {"name": COLLECTION2}, | ||||
} | } | ||||
KEYCLOAK_SERVER_URL = "https://auth.swh.org/SWHTest" | |||||
KEYCLOAK_REALM_NAME = "SWHTest" | |||||
CLIENT_ID = "swh-deposit" | |||||
keycloak_mock_auth_success = keycloak_mock_factory( | |||||
server_url=KEYCLOAK_SERVER_URL, | |||||
realm_name=KEYCLOAK_REALM_NAME, | |||||
client_id=CLIENT_ID, | |||||
auth_success=True, | |||||
user_info=USER_INFO, | |||||
user_permissions=[DEPOSIT_PERMISSION], | |||||
) | |||||
keycloak_mock_auth_failure = keycloak_mock_factory( | |||||
server_url=KEYCLOAK_SERVER_URL, | |||||
realm_name=KEYCLOAK_REALM_NAME, | |||||
client_id=CLIENT_ID, | |||||
auth_success=False, | |||||
) | |||||
def pytest_configure(): | def pytest_configure(): | ||||
setup_django_for("testing") | setup_django_for("testing") | ||||
@pytest.fixture | @pytest.fixture | ||||
def requests_mock_datadir(datadir, requests_mock_datadir): | def requests_mock_datadir(datadir, requests_mock_datadir): | ||||
"""Override default behavior to deal with put/post methods | """Override default behavior to deal with put/post methods | ||||
""" | """ | ||||
cb = partial(get_response_cb, datadir=datadir) | cb = partial(get_response_cb, datadir=datadir) | ||||
requests_mock_datadir.put(re.compile("https://"), body=cb) | requests_mock_datadir.put(re.compile("https://"), body=cb) | ||||
requests_mock_datadir.post(re.compile("https://"), body=cb) | requests_mock_datadir.post(re.compile("https://"), body=cb) | ||||
return requests_mock_datadir | return requests_mock_datadir | ||||
@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": KEYCLOAK_SERVER_URL, | |||||
"realm_name": KEYCLOAK_REALM_NAME, | |||||
}, | |||||
} | } | ||||
@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 73 Lines • ▼ Show 20 Lines | def create_deposit_collection(collection_name: str): | ||||
try: | try: | ||||
collection = DepositCollection._default_manager.get(name=collection_name) | collection = DepositCollection._default_manager.get(name=collection_name) | ||||
except DepositCollection.DoesNotExist: | except DepositCollection.DoesNotExist: | ||||
collection = DepositCollection(name=collection_name) | collection = DepositCollection(name=collection_name) | ||||
collection.save() | collection.save() | ||||
return collection | return collection | ||||
def deposit_collection_factory(collection_name=TEST_USER["collection"]["name"]): | def deposit_collection_factory(collection_name): | ||||
@pytest.fixture | @pytest.fixture | ||||
def _deposit_collection(db, collection_name=collection_name): | def _deposit_collection(db, collection_name=collection_name): | ||||
return create_deposit_collection(collection_name) | return create_deposit_collection(collection_name) | ||||
return _deposit_collection | return _deposit_collection | ||||
deposit_collection = deposit_collection_factory() | deposit_collection = deposit_collection_factory(COLLECTION) | ||||
deposit_another_collection = deposit_collection_factory("another-collection") | deposit_another_collection = deposit_collection_factory(COLLECTION2) | ||||
def _create_deposit_user(db, collection, user_data): | def _create_deposit_user( | ||||
collection: "DepositCollection", user_data: Dict | |||||
) -> "DepositClient": | |||||
"""Create/Return the test_user "test" | """Create/Return the test_user "test" | ||||
""" | """ | ||||
from swh.deposit.models import DepositClient | from swh.deposit.models import DepositClient | ||||
try: | user_data_d = deepcopy(user_data) | ||||
user = DepositClient._default_manager.get(username=user_data["username"]) | user_data_d.pop("collection", None) | ||||
except DepositClient.DoesNotExist: | user, _ = DepositClient.objects.get_or_create( # type: ignore | ||||
user = DepositClient._default_manager.create_user( | username=user_data_d["username"], | ||||
username=user_data["username"], | defaults={**user_data_d, "collections": [collection.id]}, | ||||
email=user_data["email"], | |||||
password=user_data["password"], | |||||
provider_url=user_data["provider_url"], | |||||
domain=user_data["domain"], | |||||
) | ) | ||||
anlambert: Use `DepositClient.objects.get_or_create` instead. | |||||
Done Inline Actionsgood idea. ardumont: good idea. | |||||
user.collections = [collection.id] | |||||
user.save() | |||||
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(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(deposit_another_collection, TEST_USER2) | ||||
@pytest.fixture | @pytest.fixture | ||||
def client(): | def anonymous_client(): | ||||
"""Override pytest-django one which does not work for djangorestframework. | """Create an anonymous client (no credentials during queries to the deposit) | ||||
""" | """ | ||||
return APIClient() # <- drf's client | return APIClient() # <- drf's client | ||||
def _create_authenticated_client(client, user, user_data): | def mock_keycloakopenidconnect(mocker, keycloak_mock): | ||||
"""Returned a logged client | """Mock swh.deposit.auth.KeycloakOpenIDConnect to return the keycloak_mock | ||||
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… | |||||
""" | |||||
mock = mocker.patch("swh.deposit.auth.KeycloakOpenIDConnect") | |||||
mock.from_configfile.return_value = keycloak_mock | |||||
return mock | |||||
@pytest.fixture | |||||
def mock_keycloakopenidconnect_ok(mocker, keycloak_mock_auth_success): | |||||
"""Mock keycloak so it always accepts connection for user with the right | |||||
permissions | |||||
""" | |||||
return mock_keycloakopenidconnect(mocker, keycloak_mock_auth_success) | |||||
@pytest.fixture | |||||
def mock_keycloakopenidconnect_ko(mocker, keycloak_mock_auth_failure): | |||||
"""Mock keycloak so it always refuses connections.""" | |||||
return mock_keycloakopenidconnect(mocker, keycloak_mock_auth_failure) | |||||
Done Inline Actionss/Returned/Return/ anlambert: s/Returned/Return/ | |||||
@pytest.fixture | |||||
def unauthorized_client(anonymous_client, mock_keycloakopenidconnect_ko): | |||||
"""Create an unauthorized client (will see their authentication fail) | |||||
""" | |||||
return anonymous_client | |||||
def _create_authenticated_client(client, user): | |||||
"""Return a client whose credentials will be proposed to the deposit server. | |||||
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 | ||||
deposit_user. | deposit_user. | ||||
""" | """ | ||||
_token = "%s:%s" % (user.username, user_data["password"]) | _token = "%s:%s" % (user.username, "irrelevant-in-test-context") | ||||
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 authenticated_client(mock_keycloakopenidconnect_ok, anonymous_client, deposit_user): | ||||
yield from _create_authenticated_client(client, deposit_user, TEST_USER) | yield from _create_authenticated_client(anonymous_client, deposit_user) | ||||
@pytest.fixture | @pytest.fixture | ||||
def another_authenticated_client(deposit_another_user): | def insufficient_perm_client( | ||||
client = APIClient() | mocker, keycloak_mock_auth_success, anonymous_client, deposit_user | ||||
yield from _create_authenticated_client( | ): | ||||
client, deposit_another_user, ANOTHER_TEST_USER | """keycloak accepts connection but client returned has no deposit permission, so access | ||||
) | is not allowed. | ||||
""" | |||||
keycloak_mock_auth_success.user_permissions = [] | |||||
mock_keycloakopenidconnect(mocker, keycloak_mock_auth_success) | |||||
yield from _create_authenticated_client(anonymous_client, deposit_user) | |||||
@pytest.fixture | @pytest.fixture | ||||
def sample_archive(tmp_path): | def sample_archive(tmp_path): | ||||
"""Returns a sample archive | """Returns a sample archive | ||||
""" | """ | ||||
tmp_path = str(tmp_path) # pytest version limitation in previous version | tmp_path = str(tmp_path) # pytest version limitation in previous version | ||||
Show All 21 Lines | for filename in os.listdir(atom_path): | ||||
# Keep the filename without extension | # Keep the filename without extension | ||||
atom_name = filename.split(".")[0] | atom_name = filename.split(".")[0] | ||||
data[atom_name] = raw_content | data[atom_name] = raw_content | ||||
return data | return data | ||||
def internal_create_deposit( | |||||
client: "DepositClient", | |||||
collection: "DepositCollection", | |||||
external_id: str, | |||||
status: str, | |||||
) -> "Deposit": | |||||
"""Create a deposit for a given collection with internal tool | |||||
""" | |||||
from swh.deposit.models import Deposit | |||||
deposit = Deposit( | |||||
client=client, external_id=external_id, status=status, collection=collection | |||||
) | |||||
deposit.save() | |||||
return deposit | |||||
def create_deposit( | def create_deposit( | ||||
authenticated_client, | client, | ||||
collection_name: str, | collection_name: str, | ||||
sample_archive, | sample_archive, | ||||
external_id: str, | external_id: str, | ||||
deposit_status=DEPOSIT_STATUS_DEPOSITED, | deposit_status=DEPOSIT_STATUS_DEPOSITED, | ||||
in_progress=False, | in_progress=False, | ||||
): | ): | ||||
"""Create a skeleton shell deposit | """Create a skeleton shell deposit | ||||
""" | """ | ||||
url = reverse(COL_IRI, args=[collection_name]) | url = reverse(COL_IRI, args=[collection_name]) | ||||
# when | # when | ||||
response = post_archive( | response = post_archive( | ||||
authenticated_client, | client, | ||||
url, | url, | ||||
sample_archive, | sample_archive, | ||||
HTTP_SLUG=external_id, | HTTP_SLUG=external_id, | ||||
HTTP_IN_PROGRESS=str(in_progress).lower(), | HTTP_IN_PROGRESS=str(in_progress).lower(), | ||||
) | ) | ||||
# then | # then | ||||
assert response.status_code == status.HTTP_201_CREATED, response.content.decode() | assert response.status_code == status.HTTP_201_CREATED, response.content.decode() | ||||
▲ Show 20 Lines • Show All 162 Lines • Show Last 20 Lines |
Use DepositClient.objects.get_or_create instead.