Differential D4573 Diff 16265 swh/deposit/tests/conftest.py

Changeset View

Standalone View

swh/deposit/tests/conftest.py

# Copyright (C) 2019-2020 The Software Heritage developers		# Copyright (C) 2019-2020 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
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
Show All 29 Lines	TEST_USER = {
"password": "password",		"password": "password",
"email": "test@example.org",		"email": "test@example.org",
"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": "test"},
}		}


		ANOTHER_TEST_USER = {
		"username": "test2",
		"password": "password2",
		"email": "test@example2.org",
		"provider_url": "https://hal-test.archives-ouvertes.example/",
		"domain": "archives-ouvertes.example/",
		"collection": {"name": "another-collection"},
		}


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

▲ Show 20 Lines • Show All 108 Lines • ▼ Show 20 Lines	def deposit_collection_factory(collection_name=TEST_USER["collection"]["name"]):

return _deposit_collection		return _deposit_collection


deposit_collection = deposit_collection_factory()		deposit_collection = deposit_collection_factory()
deposit_another_collection = deposit_collection_factory("another-collection")		deposit_another_collection = deposit_collection_factory("another-collection")


@pytest.fixture		def _create_deposit_user(db, collection, user_data):
def deposit_user(db, deposit_collection):
"""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:		try:
user = DepositClient._default_manager.get(username=TEST_USER["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=TEST_USER["username"],		username=user_data["username"],
email=TEST_USER["email"],		email=user_data["email"],
password=TEST_USER["password"],		password=user_data["password"],
provider_url=TEST_USER["provider_url"],		provider_url=user_data["provider_url"],
domain=TEST_USER["domain"],		domain=user_data["domain"],
)		)
user.collections = [deposit_collection.id]		user.collections = [collection.id]
user.save()		user.save()
return user		return user


@pytest.fixture		@pytest.fixture
		def deposit_user(db, deposit_collection):
		return _create_deposit_user(db, deposit_collection, TEST_USER)


		@pytest.fixture
		def deposit_another_user(db, deposit_another_collection):
		return _create_deposit_user(db, deposit_another_collection, ANOTHER_TEST_USER)


		@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


@pytest.fixture		def _create_authenticated_client(client, user, user_data):
def authenticated_client(client, deposit_user):
"""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
deposit_user.		deposit_user.

"""		"""
_token = "%s:%s" % (deposit_user.username, TEST_USER["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 = deposit_user		client.deposit_client = user
yield client		yield client
client.logout()		client.logout()


@pytest.fixture		@pytest.fixture
		def authenticated_client(client, deposit_user):
		yield from _create_authenticated_client(client, deposit_user, TEST_USER)


		@pytest.fixture
		def another_authenticated_client(deposit_another_user):
		client = APIClient()
		yield from _create_authenticated_client(
		client, deposit_another_user, ANOTHER_TEST_USER
		)


		@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
archive = create_arborescence_archive(		archive = create_arborescence_archive(
tmp_path, "archive1", "file1", b"some content in file"		tmp_path, "archive1", "file1", b"some content in file"
)		)
▲ Show 20 Lines • Show All 44 Lines • ▼ Show 20 Lines	response = authenticated_client.post(
HTTP_SLUG=external_id,		HTTP_SLUG=external_id,
HTTP_CONTENT_MD5=sample_archive["md5sum"],		HTTP_CONTENT_MD5=sample_archive["md5sum"],
HTTP_PACKAGING="http://purl.org/net/sword/package/SimpleZip",		HTTP_PACKAGING="http://purl.org/net/sword/package/SimpleZip",
HTTP_IN_PROGRESS="false",		HTTP_IN_PROGRESS="false",
HTTP_CONTENT_DISPOSITION="attachment; filename=%s" % (sample_archive["name"]),		HTTP_CONTENT_DISPOSITION="attachment; filename=%s" % (sample_archive["name"]),
)		)

# then		# then
assert response.status_code == status.HTTP_201_CREATED		assert response.status_code == status.HTTP_201_CREATED, response.content.decode()
from swh.deposit.models import Deposit		from swh.deposit.models import Deposit

deposit = Deposit._default_manager.get(external_id=external_id)		response_content = parse_xml(BytesIO(response.content))
		deposit_id = response_content["swh:deposit_id"]
		deposit = Deposit._default_manager.get(id=deposit_id)

if deposit.status != deposit_status:		if deposit.status != deposit_status:
deposit.status = deposit_status		deposit.status = deposit_status
deposit.save()		deposit.save()
assert deposit.status == deposit_status		assert deposit.status == deposit_status
return deposit		return deposit


▲ Show 20 Lines • Show All 149 Lines • Show Last 20 Lines