-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathconftest.py
99 lines (74 loc) · 2.11 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
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
from sqlalchemy.engine import reflection
from sqlalchemy.schema import (MetaData, Table, DropTable, ForeignKeyConstraint, DropConstraint)
import pytest
from api import api as ctf_api
from api.app import app as ctf_app
from api.config import options
from api.api.models import db as ctf_db
def db_DropEverything(db):
# From http://www.sqlalchemy.org/trac/wiki/UsageRecipes/DropEverything
conn=db.engine.connect()
# the transaction only applies if the DB supports
# transactional DDL, i.e. Postgresql, MS SQL Server
trans = conn.begin()
inspector = reflection.Inspector.from_engine(db.engine)
# gather all data first before dropping anything.
# some DBs lock after things have been dropped in
# a transaction.
metadata = MetaData()
tbs = []
all_fks = []
for table_name in inspector.get_table_names():
fks = []
for fk in inspector.get_foreign_keys(table_name):
if not fk['name']:
continue
fks.append(
ForeignKeyConstraint((),(),name=fk['name'])
)
t = Table(table_name,metadata,*fks)
tbs.append(t)
all_fks.extend(fks)
for fkc in all_fks:
conn.execute(DropConstraint(fkc))
for table in tbs:
conn.execute(DropTable(table))
trans.commit()
@pytest.fixture(scope="session")
def app(request):
app = ctf_app
app.config.from_object(options)
app.config["TESTING"] = True
ctx = app.test_request_context()
ctx.push()
def teardown():
ctx.pop()
request.addfinalizer(teardown)
return app
@pytest.fixture(scope="session")
def client(app):
return app.test_client()
@pytest.fixture(scope="class")
def db(request, app):
ctf_db.reflect()
db_DropEverything(ctf_db)
ctf_db.create_all()
def teardown():
ctf_db.session.close_all()
ctf_db.reflect()
db_DropEverything(ctf_db)
request.addfinalizer(teardown)
return ctf_db
@pytest.fixture(scope="class")
def session(request, db):
connection = db.engine.connect()
transaction = connection.begin()
options = dict(bind=connection, binds={})
session = db.create_scoped_session(options=options)
db.session = session
def teardown():
transaction.rollback()
connection.close()
session.remove()
request.addfinalizer(teardown)
return session