Differential D2474 Diff 8844 swh/deposit/tests/conftest.py

Changeset View

Standalone View

swh/deposit/tests/conftest.py

# Copyright (C) 2019 The Software Heritage developers		# Copyright (C) 2019 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 os		import os
import base64		import base64
import pytest		import pytest
import psycopg2		import psycopg2

from django.urls import reverse		from django.urls import reverse
		from django.test.utils import setup_databases # type: ignore
		# mypy is asked to ignore the import statement above because setup_databases
		# is not part of the d.t.utils.__all__ variable.

from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT		from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
from rest_framework import status		from rest_framework import status
from rest_framework.test import APIClient		from rest_framework.test import APIClient
from typing import Mapping		from typing import Mapping

from swh.scheduler.tests.conftest import * # noqa		from swh.scheduler.tests.conftest import * # noqa
from swh.deposit.parsers import parse_xml		from swh.deposit.parsers import parse_xml
from swh.deposit.config import (		from swh.deposit.config import (
Show All 11 Lines	TEST_USER = {
'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': {		'collection': {
'name': 'test'		'name': 'test'
},		},
}		}


def execute_sql(sql):		@pytest.fixture(scope='session')
"""Execute sql to postgres db"""		def django_db_setup(
with psycopg2.connect(database='postgres') as conn:		request,
conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)		django_db_blocker,
cur = conn.cursor()		postgresql_proc):
cur.execute(sql)


@pytest.hookimpl(tryfirst=True)
def pytest_load_initial_conftests(early_config, parser, args):
"""This hook is done prior to django loading.
Used to initialize the deposit's server db.

"""
import project.app.signals # type: ignore

def prepare_db(args, *kwargs):
from django.conf import settings		from django.conf import settings
db_name = 'tests'		settings.DATABASES['default'].update({
# work around db settings for django
for k, v in [
('ENGINE', 'django.db.backends.postgresql'),		('ENGINE', 'django.db.backends.postgresql'),
('NAME', 'tests'),		('NAME', 'tests'),
('USER', postgresql_proc.user), # noqa		('USER', postgresql_proc.user), # noqa
('HOST', postgresql_proc.host), # noqa		('HOST', postgresql_proc.host), # noqa
('PORT', postgresql_proc.port), # noqa		('PORT', postgresql_proc.port), # noqa
]:		})
settings.DATABASES['default'][k] = v		with django_db_blocker.unblock():
		setup_databases(
		verbosity=request.config.option.verbose,
		interactive=False,
		keepdb=False)

execute_sql('DROP DATABASE IF EXISTS %s' % db_name)
execute_sql('CREATE DATABASE %s TEMPLATE template0' % db_name)

project.app.signals.something = prepare_db		def execute_sql(sql):
		"""Execute sql to postgres db"""
		with psycopg2.connect(database='postgres') as conn:
		conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
		cur = conn.cursor()
		cur.execute(sql)


@pytest.fixture(autouse=True, scope='session')		@pytest.fixture(autouse=True, scope='session')
def swh_proxy():		def swh_proxy():
"""Automatically inject this fixture in all tests to ensure no outside		"""Automatically inject this fixture in all tests to ensure no outside
connection takes place.		connection takes place.

"""		"""
os.environ['http_proxy'] = 'http://localhost:999'		os.environ['http_proxy'] = 'http://localhost:999'
os.environ['https_proxy'] = 'http://localhost:999'		os.environ['https_proxy'] = 'http://localhost:999'


def create_deposit_collection(collection_name: str):		def create_deposit_collection(collection_name: str):
"""Create a deposit collection with name collection_name		"""Create a deposit collection with name collection_name

"""		"""
from swh.deposit.models import DepositCollection		from swh.deposit.models import DepositCollection
try:		try:
collection = DepositCollection._default_manager.get(		collection = DepositCollection._default_manager.get(
name=collection_name)		name=collection_name)
except DepositCollection.DoesNotExist:		except DepositCollection.DoesNotExist:
		ardumontUnsubmitted Not Done Inline Actions `.package.` not that it matters i think. ardumont: `.package.` not that it matters i think.
collection = DepositCollection(name=collection_name)		collection = DepositCollection(name=collection_name)
		ardumontUnsubmitted Not Done Inline Actions because why not? i don't know, maybe to ease readability... but in the end, task types usually are self explanatory. so maybe this can be dropped now that we know. ardumont: because why not? i don't know, maybe to ease readability... but in the end, task types usually…
collection.save()		collection.save()
return collection		return collection


def deposit_collection_factory(		def deposit_collection_factory(
collection_name=TEST_USER['collection']['name']):		collection_name=TEST_USER['collection']['name']):
@pytest.fixture		@pytest.fixture
def _deposit_collection(db, collection_name=collection_name):		def _deposit_collection(db, collection_name=collection_name):
▲ Show 20 Lines • Show All 237 Lines • Show Last 20 Lines