removing try...finally because using IF NOT EXISTS in lieu of db cleanup
@@ -13,13 +13,15 @@ def test_mock_patient_table():
sa_user = "sa"
sa_password = "sqlServerPassw0rd"
- # sa connection to create/drop the test database
- sql_sa_connection = pymssql.connect(msq_server, user=sa_user, password=sa_password)
+ # sa connection to create the test database
+ with pymssql.connect(
+ msq_server, user=sa_user, password=sa_password
+ ) as sql_sa_connection:
+
sql_sa_connection.autocommit(True)
- try:
# create the test db
- cursor = sql_sa_connection.cursor()
+ with sql_sa_connection.cursor() as cursor:
cursor.execute(
f"""
IF NOT EXISTS (SELECT * FROM sys.databases WHERE name = '{test_db_name}')
@@ -28,16 +30,13 @@ def test_mock_patient_table():
END
"""
)
- cursor.close()
# create dataframes for the Patient and Ident tables
patient_df = pd.DataFrame(
[("Larry", "Fine"), ("Moe", "Howard"), ("Curly", "Howard")],
columns=["First_Name", "Last_Name"],
)
- patient_df.index = (
- patient_df.index + 10001
- ) # use the index+10001 as the Pat_ID1
+ patient_df.index = patient_df.index + 10001 # use the index+10001 as the Pat_ID1
ident_df = pd.DataFrame([("MR8001"), ("MR8002"), ("MR8003")], columns=["IDA"])
ident_df.index = (
@@ -45,9 +44,7 @@ def test_mock_patient_table():
) # use index+10001 in the same order as Pat_ID1
# now use SQLAlchemy to populate the two tables
- conn_str = (
- f"mssql+pymssql://{sa_user}:{sa_password}@{msq_server}/{test_db_name}"
- )
+ conn_str = f"mssql+pymssql://{sa_user}:{sa_password}@{msq_server}/{test_db_name}"
engine = sqlalchemy.create_engine(conn_str, echo=False)
patient_df.to_sql(
@@ -57,15 +54,15 @@ def test_mock_patient_table():
"Ident", engine, if_exists="replace", index=True, index_label="Pat_Id1"
)
- # test a generic query for patient info
- connection = connect(
+ with connect(
msq_server,
port=1433,
database=test_db_name,
username=sa_user,
password=sa_password,
- )
+ ) as connection:
+ # test a generic query for patient info
result_all = execute(
connection,
"""
@@ -87,12 +84,3 @@ def test_mock_patient_table():
moe_patient_name = get_patient_name(connection, "MR8002")
assert moe_patient_name == "HOWARD, Moe"
-
- finally:
- # close the pymedphys connection
- if connection:
- connection.close()
-
- # clean up!
- if sql_sa_connection:
- sql_sa_connection.close()
