160 lines
4.7 KiB
Python
160 lines
4.7 KiB
Python
import glob
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
from contextlib import closing, contextmanager
|
|
|
|
import django_webtest
|
|
import psycopg2
|
|
import pytest
|
|
from django.contrib.auth.models import User
|
|
from django.core.management import call_command
|
|
from django.db import transaction
|
|
|
|
|
|
def pytest_addoption(parser):
|
|
parser.addoption(
|
|
'--bijoe-store-table',
|
|
action='store_true',
|
|
default=False,
|
|
help='Store tables value in new_tables.json',
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def app(settings, request):
|
|
settings.TEMPLATE_DEBUG = True
|
|
settings.DEBUG = True
|
|
wtm = django_webtest.WebTestMixin()
|
|
wtm._patch_settings()
|
|
request.addfinalizer(wtm._unpatch_settings)
|
|
return django_webtest.DjangoTestApp()
|
|
|
|
|
|
@pytest.fixture
|
|
def john_doe(db):
|
|
u = User(username='john.doe', first_name='John', last_name='Doe', email='john.doe@example.net')
|
|
u.set_password('john.doe')
|
|
u.save()
|
|
return u
|
|
|
|
|
|
@pytest.fixture
|
|
def admin(db):
|
|
u = User(username='super.user', first_name='Super', last_name='User', email='super.user@example.net')
|
|
u.set_password('super.user')
|
|
u.is_superuser = True
|
|
u.is_staff = True
|
|
u.save()
|
|
return u
|
|
|
|
|
|
SCHEMA_PATHS = os.path.join(os.path.dirname(__file__), 'fixtures/')
|
|
|
|
|
|
@contextmanager
|
|
def load_schema_db(schema):
|
|
import random
|
|
|
|
database_name = 'bijoe_test_%s' % random.getrandbits(20)
|
|
tmpdir = tempfile.mkdtemp()
|
|
bijoe_schema_dir = os.path.join(tmpdir, 'schemas')
|
|
os.makedirs(bijoe_schema_dir)
|
|
|
|
try:
|
|
with closing(psycopg2.connect(user=os.environ.get('USER'), dbname='postgres')) as conn:
|
|
conn.set_isolation_level(0)
|
|
with conn.cursor() as cursor:
|
|
cursor.execute('CREATE DATABASE %s' % database_name)
|
|
|
|
schema_dir = os.path.join(SCHEMA_PATHS, schema)
|
|
|
|
# copy schemas and set pg_dsn
|
|
for schema_path in glob.glob(os.path.join(schema_dir, '*_schema.json')):
|
|
with open(schema_path) as f:
|
|
schema = json.load(f)
|
|
schema['pg_dsn'] = 'dbname=%s' % database_name
|
|
new_schema_path = os.path.join(bijoe_schema_dir, os.path.basename(schema_path))
|
|
with open(new_schema_path, 'w') as fp:
|
|
json.dump(schema, fp)
|
|
|
|
# load data
|
|
for sql_path in sorted(glob.glob(os.path.join(schema_dir, '*.sql'))):
|
|
process = subprocess.Popen(
|
|
[
|
|
'psql',
|
|
'-c',
|
|
'\\set ON_ERROR_STOP on',
|
|
'--single-transaction',
|
|
database_name,
|
|
'-f',
|
|
sql_path,
|
|
],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
)
|
|
stdout, stderr = process.communicate()
|
|
return_code = process.returncode
|
|
assert return_code == 0, [stdout, stderr]
|
|
|
|
# load fixtures
|
|
fixtures = sorted(glob.glob(os.path.join(schema_dir, '*_fixture.json')))
|
|
|
|
d = {
|
|
'schema_dir': schema_dir,
|
|
'database_name': database_name,
|
|
'bijoe_schemas': [os.path.join(bijoe_schema_dir, '*_schema.json')],
|
|
'fixtures': fixtures,
|
|
}
|
|
tables_path = os.path.join(schema_dir, 'tables.json')
|
|
if os.path.exists(tables_path):
|
|
with open(tables_path) as f:
|
|
d['tables'] = json.load(f)
|
|
yield d
|
|
finally:
|
|
with closing(psycopg2.connect(user=os.environ.get('USER'), dbname='postgres')) as conn:
|
|
conn.set_isolation_level(0)
|
|
with conn.cursor() as cursor:
|
|
cursor.execute('DROP DATABASE IF EXISTS %s' % database_name)
|
|
shutil.rmtree(tmpdir)
|
|
|
|
|
|
@pytest.fixture(scope='module')
|
|
def schema1_db():
|
|
with load_schema_db('schema1') as d:
|
|
yield d
|
|
|
|
|
|
@pytest.fixture
|
|
def schema1(db, schema1_db, settings):
|
|
settings.BIJOE_SCHEMAS = schema1_db['bijoe_schemas']
|
|
for json_fixture in schema1_db['fixtures']:
|
|
call_command('loaddata', json_fixture)
|
|
return schema1_db
|
|
|
|
|
|
@pytest.fixture(scope='module')
|
|
def schema2_db():
|
|
with load_schema_db('schema2') as d:
|
|
yield d
|
|
|
|
|
|
@pytest.fixture(scope='module')
|
|
def schema2_fixtures(schema2_db, django_db_setup, django_db_blocker):
|
|
with django_db_blocker.unblock():
|
|
with transaction.atomic():
|
|
try:
|
|
for json_fixture in schema2_db['fixtures']:
|
|
call_command('loaddata', json_fixture)
|
|
yield
|
|
finally:
|
|
transaction.set_rollback(True)
|
|
|
|
|
|
@pytest.fixture
|
|
def schema2(schema2_db, schema2_fixtures, settings, django_db_blocker):
|
|
settings.BIJOE_SCHEMAS = schema2_db['bijoe_schemas']
|
|
yield schema2_db
|