Differential D4013 Diff 14193 swh/deposit/tests/conftest.py

Changeset View

Standalone View

swh/deposit/tests/conftest.py

	Show All 10 Lines
	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.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,
	EDIT_SE_IRI,			EDIT_SE_IRI,
				SWH_METADATA_AUTHORITY,
	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 create_arborescence_archive			from swh.deposit.tests.common import create_arborescence_archive
	from swh.model.identifiers import DIRECTORY, REVISION, SNAPSHOT, swhid			from swh.model.identifiers import DIRECTORY, REVISION, SNAPSHOT, swhid
	from swh.scheduler import get_scheduler			from swh.scheduler import get_scheduler
				from swh.storage import get_storage

	# 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.


	TEST_USER = {			TEST_USER = {
	"username": "test",			"username": "test",
	"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"},
	}			}


	def pytest_configure():			def pytest_configure():
	setup_django_for("testing")			setup_django_for("testing")


	@pytest.fixture()			@pytest.fixture()
	def deposit_config(swh_scheduler_config):			def deposit_config(swh_scheduler_config, swh_storage_backend_config):
				# storage_cfg = deepcopy(swh_storage_backend_config)
				# # actual side-track # otherwise: OSError: storage check config failed
				# storage_cfg["check_config"] = {"check_write": False}

	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,
	"provider": {			"provider": {
	"provider_name": "",			"provider_name": "",
	"provider_type": "deposit_client",			"provider_type": "deposit_client",
	"provider_url": "",			"provider_url": "",
	"metadata": {},			"metadata": {},
	},			},
	"scheduler": {"cls": "local", "args": swh_scheduler_config,},			"scheduler": {"cls": "local", "args": swh_scheduler_config,},
				"storage": swh_storage_backend_config,
	}			}


	@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))
	monkeypatch.setenv("SWH_CONFIG_FILENAME", conf_path)			monkeypatch.setenv("SWH_CONFIG_FILENAME", conf_path)
	return conf_path			return conf_path


	@pytest.fixture(autouse=True)			@pytest.fixture(autouse=True)
	def deposit_autoconfig(deposit_config_path, swh_scheduler_config):			def deposit_autoconfig(deposit_config_path, swh_scheduler_config):
	"""Enforce config for deposit classes inherited from APIConfig."""			"""Enforce config for deposit classes inherited from APIConfig."""
				cfg = read(deposit_config_path)

	scheduler = get_scheduler("local", swh_scheduler_config)			if "scheduler" in cfg:
				# scheduler setup: require the load-deposit tasks (already existing in
				# production)
				scheduler = get_scheduler(**cfg["scheduler"])
	task_type = {			task_type = {
	"type": "load-deposit",			"type": "load-deposit",
	"backend_name": "swh.loader.packages.deposit.tasks.LoadDeposit",			"backend_name": "swh.loader.packages.deposit.tasks.LoadDeposit",
	"description": "Load deposit task",			"description": "Load deposit task",
	}			}
	scheduler.create_task_type(task_type)			scheduler.create_task_type(task_type)

				if "storage" in cfg:
				# `deposit_update` tests requires the SWH_METADATA_AUTHORITY populated (matching
				# production)
				storage = get_storage(**cfg["storage"])
				storage.metadata_authority_add([SWH_METADATA_AUTHORITY])


	@pytest.fixture(scope="session")			@pytest.fixture(scope="session")
	def django_db_setup(request, django_db_blocker, postgresql_proc):			def django_db_setup(request, django_db_blocker, postgresql_proc):
	from django.conf import settings			from django.conf import settings

	settings.DATABASES["default"].update(			settings.DATABASES["default"].update(
	{			{
	("ENGINE", "django.db.backends.postgresql"),			("ENGINE", "django.db.backends.postgresql"),
	▲ Show 20 Lines • Show All 321 Lines • Show Last 20 Lines