1import unittest 2from functools import partial 3import time 4 5import molequeue 6 7class TestClient(unittest.TestCase): 8 9 def test_submit_job(self): 10 client = molequeue.Client() 11 client.connect_to_server('MoleQueue') 12 13 job = molequeue.Job() 14 job.queue = 'salix' 15 job.program = 'sleep (testing)' 16 17 file_path = molequeue.FilePath() 18 file_path.path = "/tmp/test" 19 20 job.input_file = file_path 21 22 molequeue_id = client.submit_job(job) 23 24 print "MoleQueue ID: ", molequeue_id 25 26 self.assertTrue(isinstance(molequeue_id, int)) 27 28 client.disconnect() 29 30 def test_notification_callback(self): 31 client = molequeue.Client() 32 client.connect_to_server('MoleQueue') 33 34 self.callback_count = 0 35 36 def callback_counter(testcase, msg): 37 testcase.callback_count +=1 38 39 callback = partial(callback_counter, self) 40 41 client.register_notification_callback(callback) 42 client.register_notification_callback(callback) 43 44 job = molequeue.Job() 45 job.queue = 'salix' 46 job.program = 'sleep (testing)' 47 48 molequeue_id = client.submit_job(job) 49 50 # wait for notification 51 time.sleep(1) 52 53 self.assertIs(self.callback_count, 2) 54 55 client.disconnect() 56 57 def test_wait_for_response_timeout(self): 58 client = molequeue.Client() 59 # Fake up the request 60 client._request_response_map[1] = None 61 start = time.time() 62 response = client._wait_for_response(1, 3) 63 end = time.time() 64 65 self.assertEqual(response, None) 66 self.assertEqual(int(end - start), 3) 67 68 def test_lookup_job(self): 69 client = molequeue.Client() 70 client.connect_to_server('MoleQueue') 71 72 expected_job = molequeue.Job() 73 expected_job.queue = 'salix' 74 expected_job.program = 'sleep (testing)' 75 expected_job.description = 'This is a test job' 76 expected_job.hide_from_gui = True 77 expected_job.popup_on_state_change = False 78 79 file_contents = molequeue.FileContents() 80 file_contents.filename = 'test.in' 81 file_contents.contents = 'Hello' 82 expected_job.input_file = file_contents 83 84 molequeue_id = client.submit_job(expected_job) 85 86 job = client.lookup_job(molequeue_id) 87 88 self.assertEqual(molequeue_id, job.molequeue_id()) 89 self.assertEqual(job.job_state(), molequeue.JobState.ACCEPTED) 90 self.assertEqual(job.queue_id(), None) 91 self.assertEqual(job.queue, expected_job.queue) 92 self.assertEqual(job.program, expected_job.program) 93 self.assertEqual(job.description, expected_job.description) 94 self.assertEqual(job.hide_from_gui, 95 expected_job.hide_from_gui) 96 self.assertEqual(job.popup_on_state_change, 97 expected_job.popup_on_state_change) 98 99 client.disconnect() 100 101 def test_request_queue_list_update(self): 102 client = molequeue.Client() 103 client.connect_to_server('MoleQueue') 104 105 queues = client.request_queue_list_update() 106 107 for q in queues: 108 print q.name, ", ", q.programs; 109 110 client.disconnect() 111 112 113if __name__ == '__main__': 114 unittest.main() 115