Differential D5137 Diff 18771 swh/deposit/tests/conftest.py

Changeset 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

		user_data_d = deepcopy(user_data)

try:		try:
user = DepositClient._default_manager.get(username=user_data["username"])		user = DepositClient.objects.get( # type: ignore
except DepositClient.DoesNotExist:		username=user_data_d["username"]
user = DepositClient._default_manager.create_user(
username=user_data["username"],
email=user_data["email"],
password=user_data["password"],
provider_url=user_data["provider_url"],
domain=user_data["domain"],
)		)
user.collections = [collection.id]		except DepositClient.DoesNotExist:
		user_data_d.pop("collection", None)
		user = DepositClient(**user_data_d, collections=[collection.id])
user.save()		user.save()
		anlambertUnsubmitted Not Done Inline Actions Use `DepositClient.objects.get_or_create` instead. anlambert: Use `DepositClient.objects.get_or_create` instead.
		ardumontAuthorUnsubmitted Done Inline Actions good 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(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

		anlambertUnsubmitted Not Done Inline Actions This 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)

		anlambertUnsubmitted Done Inline Actions s/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