Differential D4013 Diff 14314 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,
	Show All 19 Lines
	}			}


	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):
	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)


	@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(
	{			{
	▲ Show 20 Lines • Show All 322 Lines • Show Last 20 Lines