-
Notifications
You must be signed in to change notification settings - Fork 2
/
tests.py
66 lines (53 loc) · 1.5 KB
/
tests.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
import unittest
from os import path
from sqlalchemy.exc import IntegrityError, OperationalError
from devkit import create_userbase
class DBTester(unittest.TestCase):
def __init__(self, app=None, db=None):
super().__init__()
self.app = app
self.db = db
def init_app(self, app, db):
self.app = app
self.db = db
def test_db_ready(self):
print('testing database ...')
if self.test_db_exist():
if self.test_db_connection():
return True
return False
def test_db_exist(self):
if 'sqlite:///' in self.app.config['SQLALCHEMY_DATABASE_URI']:
db_uri = str(self.app.config['SQLALCHEMY_DATABASE_URI']).replace('sqlite:///', '')
if path.exists(db_uri):
return True
print('(!) ERROR: No active db has been detected for the current working app.')
return False
def test_db_connection(self):
if self.test_db_read():
if self.test_db_write():
print('CONNECTION ON : {}'.format(self.app.config['SQLALCHEMY_DATABASE_URI']))
return True
print('(!) ERROR : failed connection')
print('(!) INFO : have you run upgrade() from last migration file?')
return False
@staticmethod
def test_db_read():
from app.dbModels import User, Owner, Pub
try:
User.query.first()
Owner.query.first()
Pub.query.first()
except OperationalError:
return False
return True
def test_db_write(self):
try:
create_userbase(test_db=True)
except IntegrityError:
self.test_db_write()
except RuntimeError:
return False
return True
if __name__ == '__main__':
unittest.main()