Merge pull request #63 from DiamondLightSource/testdb

Use testdb fixture when possible

tests/test_core.py

@@ -1,12 +1,9 @@
 from __future__ import absolute_import, division, print_function
 
-import context
-import ispyb
 import time
 
-def test_insert_session_for_proposal_code_number(testconfig):
-  with ispyb.open(testconfig) as conn:
-        core = conn.core
+def test_insert_session_for_proposal_code_number(testdb):
+        core = testdb.core
 
         # Test upsert_person:
         params = core.get_person_params()
@@ -70,9 +67,8 @@ def test_insert_session_for_proposal_code_number(testconfig):
         assert phpid is not None
         assert phpid > 0
 
-def test_upsert_sample(testconfig):
-  with ispyb.open(testconfig) as conn:
-        core = conn.core
+def test_upsert_sample(testdb):
+        core = testdb.core
         params = core.get_sample_params()
         params['containerid'] = 1326
         params['crystalid'] = 3918
@@ -87,56 +83,46 @@ def test_upsert_sample(testconfig):
         id = core.upsert_sample(list(params.values()))
         assert id is not None
 
-def test_retrieve_visit_id(testconfig):
-  with ispyb.open(testconfig) as conn:
-        id = conn.core.retrieve_visit_id('cm14451-2')
+def test_retrieve_visit_id(testdb):
+        id = testdb.core.retrieve_visit_id('cm14451-2')
         assert id == 55168
 
-def test_retrieve_current_sessions(testconfig):
-  with ispyb.open(testconfig) as conn:
-        rs = conn.core.retrieve_current_sessions('i03', 24*60*30000)
+def test_retrieve_current_sessions(testdb):
+        rs = testdb.core.retrieve_current_sessions('i03', 24*60*30000)
         assert len(rs) > 0
 
-def test_retrieve_sessions_for_person_login(testconfig):
-  with ispyb.open(testconfig) as conn:
-        rs = conn.core.retrieve_sessions_for_person_login('boaty')
+def test_retrieve_sessions_for_person_login(testdb):
+        rs = testdb.core.retrieve_sessions_for_person_login('boaty')
         assert len(rs) > 0
 
-def test_retrieve_current_sessions_for_person(testconfig):
-  with ispyb.open(testconfig) as conn:
-        rs = conn.core.retrieve_current_sessions_for_person('i03', 'boaty', tolerance_mins=24*60*30000)
+def test_retrieve_current_sessions_for_person(testdb):
+        rs = testdb.core.retrieve_current_sessions_for_person('i03', 'boaty', tolerance_mins=24*60*30000)
         assert len(rs) > 0
 
-def test_retrieve_most_recent_session(testconfig):
-  with ispyb.open(testconfig) as conn:
-        rs = conn.core.retrieve_most_recent_session('i03', 'cm')
+def test_retrieve_most_recent_session(testdb):
+        rs = testdb.core.retrieve_most_recent_session('i03', 'cm')
         assert len(rs) == 1
 
-def test_retrieve_persons_for_proposal(testconfig):
-  with ispyb.open(testconfig) as conn:
-        rs = conn.core.retrieve_persons_for_proposal('cm', 14451)
+def test_retrieve_persons_for_proposal(testdb):
+        rs = testdb.core.retrieve_persons_for_proposal('cm', 14451)
         assert len(rs) == 1
         login = rs[0]['login']
         assert login is not None
 
-def test_retrieve_persons_for_session(testconfig):
-  with ispyb.open(testconfig) as conn:
-        rs = conn.core.retrieve_persons_for_session('cm', 14451, 1)
+def test_retrieve_persons_for_session(testdb):
+        rs = testdb.core.retrieve_persons_for_session('cm', 14451, 1)
         assert len(rs) == 1
         login = rs[0]['login']
         assert login is not None
 
-def test_retrieve_current_cm_sessions(testconfig):
-  with ispyb.open(testconfig) as conn:
-        rs = conn.core.retrieve_current_cm_sessions('i03')
+def test_retrieve_current_cm_sessions(testdb):
+        rs = testdb.core.retrieve_current_cm_sessions('i03')
         assert len(rs) > 0
 
-def test_retrieve_active_plates(testconfig):
-  with ispyb.open(testconfig) as conn:
-        rs = conn.core.retrieve_active_plates('i02-2')
+def test_retrieve_active_plates(testdb):
+        rs = testdb.core.retrieve_active_plates('i02-2')
         assert len(rs) >= 0
 
-def test_retrieve_proposal_title(testconfig):
-  with ispyb.open(testconfig) as conn:
-        title = conn.core.retrieve_proposal_title('cm', 14451)
+def test_retrieve_proposal_title(testdb):
+        title = testdb.core.retrieve_proposal_title('cm', 14451)
         assert title.strip() == 'I03 Commissioning Directory 2016'

tests/test_em_structures.py

@@ -1,11 +1,8 @@
 from __future__ import absolute_import, division, print_function
 
-import context
-import ispyb
 
-def test_insert_movie(testconfig):
-  with ispyb.open(testconfig) as conn:
-        emacquisition = conn.em_acquisition
+def test_insert_movie(testdb):
+        emacquisition = testdb.em_acquisition
         group_params = emacquisition.get_data_collection_group_params()
         group_params['parentid'] = 55168
         group_id = emacquisition.insert_data_collection_group(list(group_params.values()))
@@ -23,9 +20,8 @@ def test_insert_movie(testconfig):
         assert movie_id is not None
         assert movie_id > 0
 
-def test_insert_motion_correction(testconfig):
-  with ispyb.open(testconfig) as conn:
-        emacquisition = conn.em_acquisition
+def test_insert_motion_correction(testdb):
+        emacquisition = testdb.em_acquisition
         group_params = emacquisition.get_data_collection_group_params()
         group_params['parentid'] = 55168
         group_id = emacquisition.insert_data_collection_group(list(group_params.values()))
@@ -46,9 +42,8 @@ def test_insert_motion_correction(testconfig):
         motion_cor_id = emacquisition.insert_motion_correction(list(params.values()))
         assert motion_cor_id is not None
 
-def test_insert_ctf(testconfig):
-  with ispyb.open(testconfig) as conn:
-        emacquisition = conn.em_acquisition
+def test_insert_ctf(testdb):
+        emacquisition = testdb.em_acquisition
         group_params = emacquisition.get_data_collection_group_params()
         group_params['parentid'] = 55168
         group_id = emacquisition.insert_data_collection_group(list(group_params.values()))
@@ -73,9 +68,8 @@ def test_insert_ctf(testconfig):
         ctf_id = emacquisition.insert_ctf(list(params.values()))
         assert ctf_id is not None
 
-def test_insert_drift(testconfig):
-  with ispyb.open(testconfig) as conn:
-        emacquisition = conn.em_acquisition
+def test_insert_drift(testdb):
+        emacquisition = testdb.em_acquisition
         group_params = emacquisition.get_data_collection_group_params()
         group_params['parentid'] = 55168
         group_id = emacquisition.insert_data_collection_group(list(group_params.values()))

tests/test_misc.py

@@ -2,16 +2,13 @@
 
 import threading
 
-import context
-import ispyb
 import ispyb.exception
 import ispyb.model.__future__
 import mysql.connector.errors
 import pytest
 
-def test_multi_threads_upsert(testconfig):
-  with ispyb.open(testconfig) as conn:
-        mxprocessing = conn.mx_processing
+def test_multi_threads_upsert(testdb):
+        mxprocessing = testdb.mx_processing
 
         params = mxprocessing.get_program_params()
         params['cmd_line'] = 'dials -xia2 /path/to/files'
@@ -41,10 +38,9 @@ def test_multi_threads_upsert(testconfig):
         for worker in worker_list:
             worker.join()
 
-def test_retrieve_failure(testconfig):
-  with ispyb.open(testconfig) as conn:
+def test_retrieve_failure(testdb):
     with pytest.raises(ispyb.exception.ISPyBNoResultException):
-      rs = conn.mx_acquisition.retrieve_data_collection_main(0)
+      rs = testdb.mx_acquisition.retrieve_data_collection_main(0)
 
 def test_database_reconnects_on_connection_failure(testconfig, testdb):
   ispyb.model.__future__.enable(testconfig, section='ispyb_mysql_sp')

tests/test_mxacquisition.py

@@ -2,14 +2,11 @@
 
 from datetime import datetime
 
-import context
-import ispyb
 import ispyb.exception
 import pytest
 
-def test_mxacquisition_methods(testconfig):
-  with ispyb.open(testconfig) as conn:
-        mxacquisition = conn.mx_acquisition
+def test_mxacquisition_methods(testdb):
+        mxacquisition = testdb.mx_acquisition
 
         params = mxacquisition.get_data_collection_group_params()
         params['parentid'] = 55168 # sessionId
@@ -41,7 +38,7 @@ def test_mxacquisition_methods(testconfig):
         rs = mxacquisition.retrieve_data_collection_main(id1)
         assert rs[0]['groupId'] == dcgid
 
-        dc = conn.get_data_collection(id1)
+        dc = testdb.get_data_collection(id1)
         assert dc.image_count == 360
         assert dc.dcgid == dcgid
         assert dc.group.dcgid == dcgid

tests/test_mxprocessing.py

@@ -1,12 +1,9 @@
 from __future__ import absolute_import, division, print_function
 
-import context
 import datetime
-import ispyb
 
-def test_processing_jobs(testconfig):
-  with ispyb.open(testconfig) as conn:
-        mxprocessing = conn.mx_processing
+def test_processing_jobs(testdb):
+        mxprocessing = testdb.mx_processing
 
         params = mxprocessing.get_job_params()
         params['datacollectionid'] = 993677
@@ -49,7 +46,7 @@ def test_processing_jobs(testconfig):
 
         # Retrieve same information via object model
 
-        job = conn.get_processing_job(job_id)
+        job = testdb.get_processing_job(job_id)
         assert job.name == 'test_job'
         assert job.DCID == 993677
         assert job.comment == 'Test job by the unit test system ...'
@@ -131,9 +128,8 @@ def test_processing1(testdb):
   assert program.status == 1
   assert program.status_text == 'success'
 
-def test_processing2(testconfig):
-  with ispyb.open(testconfig) as conn:
-        mxprocessing = conn.mx_processing
+def test_processing2(testdb):
+        mxprocessing = testdb.mx_processing
 
         params = mxprocessing.get_program_params()
         params['cmd_line'] = 'ls -ltr'
@@ -152,7 +148,7 @@ def test_processing2(testconfig):
         assert len(pa) > 0
 
         # Find program using the processing job ID and verify stored values
-        programs = conn.get_processing_job(5).programs
+        programs = testdb.get_processing_job(5).programs
         assert programs
         assert len(programs) >= 1
         programs = list(filter(lambda p: p.app_id == id, programs))
@@ -253,9 +249,8 @@ def test_processing2(testconfig):
         id = mxprocessing.upsert_quality_indicators(list(params.values()))
         assert id is not None
 
-def test_post_processing(testconfig):
-  with ispyb.open(testconfig) as conn:
-        mxprocessing = conn.mx_processing
+def test_post_processing(testdb):
+        mxprocessing = testdb.mx_processing
 
         params = mxprocessing.get_run_params()
         params['parentid'] = 596133 # some autoProcScalingId

tests/test_mxscreening.py

@@ -2,14 +2,11 @@
 
 from datetime import datetime
 
-import context
-import ispyb
-
-def test_insert_all_screening(testconfig):
-  with ispyb.open(testconfig) as conn:
-        core = conn.core
-        mxscreening = conn.mx_screening
-        mxacquisition = conn.mx_acquisition
+
+def test_insert_all_screening(testdb):
+        core = testdb.core
+        mxscreening = testdb.mx_screening
+        mxacquisition = testdb.mx_acquisition
 
         test_session = 'cm14451-2'
         session_id = core.retrieve_visit_id(test_session)

tests/test_shipping.py

@@ -1,16 +1,12 @@
 from __future__ import absolute_import, division, print_function
 
-import context
-import ispyb
 
-def test_update_container_assign(testconfig):
-  with ispyb.open(testconfig) as conn:
-        conn.shipping.update_container_assign('i04', 'DLS-0001', 10)
+def test_update_container_assign(testdb):
+        testdb.shipping.update_container_assign('i04', 'DLS-0001', 10)
 
 
-def test_upsert_dewar(testconfig):
-  with ispyb.open(testconfig) as conn:
-        shipping = conn.shipping
+def test_upsert_dewar(testdb):
+        shipping = testdb.shipping
         params = shipping.get_dewar_params()
         params['shipping_id'] = 474
         params['name'] = 'Test-dewar'
@@ -30,7 +26,6 @@ def test_upsert_dewar(testconfig):
         assert did2 is not None
         assert did2 > 0
 
-def test_retrieve_dewars(testconfig):
-  with ispyb.open(testconfig) as conn:
-        rs = conn.shipping.retrieve_dewars_for_proposal_code_number('cm', 1)
+def test_retrieve_dewars(testdb):
+        rs = testdb.shipping.retrieve_dewars_for_proposal_code_number('cm', 1)
         assert len(rs) > 0

tests/test_strictordereddict.py

@@ -1,14 +1,13 @@
-import context
+from __future__ import absolute_import, division, print_function
+
 from ispyb.strictordereddict import StrictOrderedDict
 
+import pytest
+
 def test_keyerror():
     d = StrictOrderedDict([('c',None), ('b',None), ('a',None)])
-    try:
+    with pytest.raises(KeyError):
         d['new_key'] = 'some value'
-    except KeyError:
-        assert True
-    else:
-        assert False
 
 def test_order():
     d = StrictOrderedDict([('c',1), ('b',2), ('a',3)])

tests/test_xmltools.py

@@ -2,21 +2,18 @@
 
 import os
 
-import context
-import ispyb
 from ispyb.xmltools import mx_data_reduction_to_ispyb, xml_file_to_dict
 
-def test_mx_data_reduction_xml_to_ispyb(testconfig):
-  with ispyb.open(testconfig) as conn:
-        mxprocessing = conn.mx_processing
+def test_mx_data_reduction_xml_to_ispyb(testdb):
+        mxprocessing = testdb.mx_processing
 
         xml_file = os.path.abspath(os.path.join(os.path.dirname(__file__), 'data/mx_data_reduction_pipeline_results.xml'))
         # Find the datacollection associated with this data reduction run
         xml_dir = os.path.split(xml_file)[0]
         try:
             dc_id = int(open(os.path.join(xml_dir, '.dc_id'), 'r').read())
             print('Got DC ID %d from file system' % dc_id)
-        except:
+        except Exception:
             dc_id = None
 
         mx_data_reduction_dict = xml_file_to_dict(xml_file)