-
Notifications
You must be signed in to change notification settings - Fork 0
/
conftest.py
82 lines (67 loc) · 2.48 KB
/
conftest.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
import dateutil.parser as dtparse
import importlib
import json
import os
import pytest
from sqlalchemy import Table
from nngmail import app, db
from nngmail.models import AddresseeEnum
@pytest.fixture
def client():
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:'
app.config['TESTING'] = True
client = app.test_client()
with app.app_context():
db.create_all()
yield client
def load_fixtures(filepath):
enum_map = {}
for tt in AddresseeEnum:
enum_map[tt.value] = tt
def _datetime_parser(dct):
for key, value in list(dct.items()):
try:
dct[key] = dtparse.parse(value)
except Exception:
pass
if key == 'type_':
dct[key] = enum_map[value]
return dct
with open(filepath, 'rb') as fin:
fixtures = json.load(fin, object_hook=_datetime_parser)
conn = db.engine.connect()
metadata = db.metadata
for fixture in fixtures:
if 'model' in fixture:
module_name, class_name = fixture['model'].rsplit('.', 1)
module = importlib.import_module(module_name)
model = getattr(module, class_name)
for fields in fixture['records']:
obj = model(**fields)
db.session.add(obj)
db.session.commit()
elif 'table' in fixture:
table = Table(fixture['table'], metadata)
conn.execute(table.insert(), fixture['records'])
else:
raise ValueError("Fixture missing a 'model' or 'table' field: "
"{0}".format(json.dumps(fixture)))
@pytest.fixture
def fixtures(client):
fixtures_dirs = [os.path.join(app.root_path, 'tests/fixtures'),
os.path.join(app.root_path, 'api/tests/fixtures')]
for directory in app.config.get('FIXTURES_DIRS', []):
if not os.path.isabs(directory):
directory = os.path.abspath(os.path.join(app.root_path, directory))
fixtures_dirs.append(directory)
for filename in ['db-fixtures.json']:
for directory in fixtures_dirs:
print(directory)
filepath = os.path.join(directory, filename)
if os.path.exists(filepath):
# TODO load the data into the database
load_fixtures(filepath)
break
else:
raise IOError("Error loading '{0}'. "
"File could not be found".format(filename))