clarity_tests.py
4.75 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
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
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, host='clarityprod.uphs.upenn.edu'):
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 #####
#because we are dealing with wierd hanging issues, we run tests one at a time
#because we don't want to hit clarity more than necessary, we use dev server
class TestStuff(unittest.TestCase):
#Test a basic connect and execute
def test_basic_conn_execute(self):
eng = get_clarity_engine()
with eng.connect() as conn:
res = conn.execute('SELECT TOP 3 PAT_ID FROM PAT_ENC')
self.assertEqual(len(list(res)), 3)
def test_dev_conn_execute(self):
eng = get_clarity_engine(host='claritydev.uphs.upenn.edu')
with eng.connect() as conn:
res = conn.execute('SELECT TOP 3 PAT_ID FROM PAT_ENC')
self.assertEqual(len(list(res)), 3)
#This hangs...sometimes...why? Hung on the second time I ran it.
#Is it that it hangs after the raw connection
def test_temp_table_persistence(self):
eng = get_clarity_engine(host='claritydev.uphs.upenn.edu')
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)
#we expect the global temp table to disappear with new connection
import time
print("Sleeping 3 seconds...")
time.sleep(3)
print("Done sleeping")
with eng.connect() as conn:
print("`eng.connect() as conn` finished executing")
with self.assertRaises(Exception) as e:
res = conn.execute('SELECT * FROM ##COHORT')
print(e.exception)
def will_this_hang(self):
eng = get_clarity_engine(host='claritydev.uphs.upenn.edu')
print("Try first connection as raw connection...")
with eng.raw_connection().cursor() as cur:
print("connect executed")
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)
print("recreate engine")
eng = get_clarity_engine(host='claritydev.uphs.upenn.edu', timeout=60) #if I get rid of this line it doesn't hang
print("Try second connection as regular connection...")
with eng.connect() as conn:
print("connect executed")
with self.assertRaises(Exception) as e:
res = conn.execute('SELECT * FROM ##COHORT') #I think this is where it hangs? or not
print(e.exception)
print("Try third connection as regular connection...")
with eng.connect() as conn:
print("connect executed")
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)
#we expect the global temp table to disappear with new connection
def test_raw_connection(self):
eng = get_clarity_engine(host='claritydev.uphs.upenn.edu')
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__':
# unittest.main()
t = TestStuff()
t.test_basic_conn_execute()
t.test_dev_conn_execute()
t.test_temp_table_persistence()
t.will_this_hang()
t.test_raw_connection()