sqlalchemy_clarity_tests.py
2.98 KB
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
import unittest
import sqlalchemy
from unittest.mock import MagicMock
def get_mssql_engine(
username="lynchse",
host="clarityprod.uphs.upenn.edu",
database="clarity_snapshot_db",
domain="UPHS",
port="1433",
timeout=600,
password=None,
):
from sqlalchemy import create_engine
user = domain + "\\" + username
clarity_engine = create_engine(f"mssql+pymssql://{user}:{password}@{host}:{port}/{database}?timeout={timeout}")
return clarity_engine
selahcredsfilename = 'C:\\Users\\LynchSe\\Documents\\selah_clarity_credentials.txt'
def get_clarity_engine(credsfilename = selahcredsfilename, timeout=600):
with open(credsfilename, 'r') as credsfile:
name = credsfile.readline().strip()
pw = credsfile.readline().strip()
clarity_engine = get_mssql_engine(username=name, password=pw, timeout=timeout)
return clarity_engine
##### BEGIN ACTUAL TESTS #####
class TestStuff(unittest.TestCase):
def test_sqlite(self):
from sqlalchemy import create_engine
sqlite_db_filepath = "C:\\Users\\LynchSe\\Documents\\Data\\database.db"
e = create_engine('sqlite:///{}'.format(sqlite_db_filepath))
c = e.connect()
c.execute('SELECT 1;')
c.execute('CREATE TABLE IF NOT EXISTS COHORT (EMPI VARCHAR(90) NOT NULL);')
c.execute("INSERT INTO COHORT (EMPI) values ('8001111117'),('1000000000'),('8333333002');")
res = c.execute("SELECT EMPI FROM COHORT;")
for line in res:
print(type(line))
print(line)
c.close()
import subprocess
subprocess.check_output('rm {}'.format(sqlite_db_filepath))
def test_drop_cohort(self):
eng = get_clarity_engine(timeout=10)
print("Am I about to hang like a shitty database engine?")
with eng.connect() as conn:
conn.execute('DROP TABLE IF EXISTS ##cohort;')
def test_sqalch_insert_read(self):
eng = get_clarity_engine()
with eng.connect() as conn:
conn.execute('DROP TABLE IF EXISTS ##COHORT')
conn.execute('SELECT TOP 3 PAT_ID INTO ##COHORT FROM PAT_ENC')
res = conn.execute('SELECT * FROM ##COHORT')
self.assertEqual(len(list(res)), 3)
with eng.raw_connection().cursor() as cur:
cur.execute('DROP TABLE IF EXISTS ##COHORT')
cur.execute('SELECT TOP 3 PAT_ID INTO ##COHORT FROM PAT_ENC')
cur.execute('SELECT * FROM ##COHORT')
self.assertEqual(len([row for row in cur]), 3)
if __name__ == '__main__':
tests_to_run = [
"test_sqlite"
# ,"try_a_mock"
# ,"test_sqalch_insert_read"
# ,"test_shitty_hang"
]
suite = unittest.TestSuite()
for test in tests_to_run:
suite.addTest(TestStuff(test))
runner = unittest.TextTestRunner()
runner.run(suite)
# unittest.main()