clarity_tests.py
2.42 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
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 #####
#because we dont' want to hit clarity more than necessary, we run tests one at a time
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_temp_table_persistence(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)
#we expect the global temp table to disappear with new connection
with eng.connect() as conn:
with self.assertRaises(Exception) as e:
res = conn.execute('SELECT * FROM ##COHORT')
print(e.exception)
# def test_raw_connection(self):
# eng = get_clarity_engine()
# 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()