Changeset View
Changeset View
Standalone 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 base64 | import base64 | ||||
import pytest | import pytest | ||||
import psycopg2 | import psycopg2 | ||||
from django.urls import reverse | from django.urls import reverse | ||||
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 ( | ||||
COL_IRI, EDIT_SE_IRI, DEPOSIT_STATUS_DEPOSITED, DEPOSIT_STATUS_REJECTED, | COL_IRI, EDIT_SE_IRI, DEPOSIT_STATUS_DEPOSITED, DEPOSIT_STATUS_REJECTED, | ||||
DEPOSIT_STATUS_PARTIAL, DEPOSIT_STATUS_LOAD_SUCCESS, | DEPOSIT_STATUS_PARTIAL, DEPOSIT_STATUS_LOAD_SUCCESS, | ||||
DEPOSIT_STATUS_LOAD_FAILURE | DEPOSIT_STATUS_VERIFIED, DEPOSIT_STATUS_LOAD_FAILURE | ||||
) | ) | ||||
from swh.deposit.tests.common import create_arborescence_archive | from swh.deposit.tests.common import create_arborescence_archive | ||||
TEST_USER = { | TEST_USER = { | ||||
'username': 'test', | 'username': 'test', | ||||
'password': 'password', | 'password': 'password', | ||||
'email': 'test@example.org', | 'email': 'test@example.org', | ||||
Show All 19 Lines | """This hook is done prior to django loading. | ||||
Used to initialize the deposit's server db. | Used to initialize the deposit's server db. | ||||
""" | """ | ||||
import project.app.signals | import project.app.signals | ||||
def prepare_db(*args, **kwargs): | def prepare_db(*args, **kwargs): | ||||
from django.conf import settings | from django.conf import settings | ||||
db_name = 'tests' | db_name = 'tests' | ||||
print('before: %s' % settings.DATABASES) | |||||
# work around db settings for django | # work around db settings for django | ||||
for k, v in [ | 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 | settings.DATABASES['default'][k] = v | ||||
print('after: %s' % settings.DATABASES) | |||||
execute_sql('DROP DATABASE IF EXISTS %s' % db_name) | execute_sql('DROP DATABASE IF EXISTS %s' % db_name) | ||||
execute_sql('CREATE DATABASE %s TEMPLATE template0' % db_name) | execute_sql('CREATE DATABASE %s TEMPLATE template0' % db_name) | ||||
project.app.signals.something = prepare_db | project.app.signals.something = prepare_db | ||||
@pytest.fixture(autouse=True, scope='session') | |||||
def swh_proxy(): | |||||
"""Automatically inject this fixture in all tests to ensure no outside | |||||
connection takes place. | |||||
""" | |||||
os.environ['http_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) | ||||
▲ Show 20 Lines • Show All 150 Lines • ▼ Show 20 Lines | def _deposit(sample_archive, deposit_collection, authenticated_client, | ||||
) | ) | ||||
return _deposit | return _deposit | ||||
deposited_deposit = deposit_factory() | deposited_deposit = deposit_factory() | ||||
rejected_deposit = deposit_factory(deposit_status=DEPOSIT_STATUS_REJECTED) | rejected_deposit = deposit_factory(deposit_status=DEPOSIT_STATUS_REJECTED) | ||||
partial_deposit = deposit_factory(deposit_status=DEPOSIT_STATUS_PARTIAL) | partial_deposit = deposit_factory(deposit_status=DEPOSIT_STATUS_PARTIAL) | ||||
verified_deposit = deposit_factory(deposit_status=DEPOSIT_STATUS_VERIFIED) | |||||
completed_deposit = deposit_factory(deposit_status=DEPOSIT_STATUS_LOAD_SUCCESS) | completed_deposit = deposit_factory(deposit_status=DEPOSIT_STATUS_LOAD_SUCCESS) | ||||
failed_deposit = deposit_factory(deposit_status=DEPOSIT_STATUS_LOAD_FAILURE) | failed_deposit = deposit_factory(deposit_status=DEPOSIT_STATUS_LOAD_FAILURE) | ||||
@pytest.fixture | @pytest.fixture | ||||
def partial_deposit_with_metadata( | def partial_deposit_with_metadata( | ||||
sample_archive, deposit_collection, authenticated_client, | sample_archive, deposit_collection, authenticated_client, | ||||
atom_dataset): | atom_dataset): | ||||
▲ Show 20 Lines • Show All 58 Lines • Show Last 20 Lines |