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
|
import unittest
from functools import partial
import time
import molequeue
class TestClient(unittest.TestCase):
def test_submit_job(self):
client = molequeue.Client()
client.connect_to_server('MoleQueue')
job = molequeue.Job()
job.queue = 'salix'
job.program = 'sleep (testing)'
file_path = molequeue.FilePath()
file_path.path = "/tmp/test"
job.input_file = file_path
molequeue_id = client.submit_job(job)
print "MoleQueue ID: ", molequeue_id
self.assertTrue(isinstance(molequeue_id, int))
client.disconnect()
def test_notification_callback(self):
client = molequeue.Client()
client.connect_to_server('MoleQueue')
self.callback_count = 0
def callback_counter(testcase, msg):
testcase.callback_count +=1
callback = partial(callback_counter, self)
client.register_notification_callback(callback)
client.register_notification_callback(callback)
job = molequeue.Job()
job.queue = 'salix'
job.program = 'sleep (testing)'
molequeue_id = client.submit_job(job)
# wait for notification
time.sleep(1)
self.assertIs(self.callback_count, 2)
client.disconnect()
def test_wait_for_response_timeout(self):
client = molequeue.Client()
# Fake up the request
client._request_response_map[1] = None
start = time.time()
response = client._wait_for_response(1, 3)
end = time.time()
self.assertEqual(response, None)
self.assertEqual(int(end - start), 3)
def test_lookup_job(self):
client = molequeue.Client()
client.connect_to_server('MoleQueue')
expected_job = molequeue.Job()
expected_job.queue = 'salix'
expected_job.program = 'sleep (testing)'
expected_job.description = 'This is a test job'
expected_job.hide_from_gui = True
expected_job.popup_on_state_change = False
file_contents = molequeue.FileContents()
file_contents.filename = 'test.in'
file_contents.contents = 'Hello'
expected_job.input_file = file_contents
molequeue_id = client.submit_job(expected_job)
job = client.lookup_job(molequeue_id)
self.assertEqual(molequeue_id, job.molequeue_id())
self.assertEqual(job.job_state(), molequeue.JobState.ACCEPTED)
self.assertEqual(job.queue_id(), None)
self.assertEqual(job.queue, expected_job.queue)
self.assertEqual(job.program, expected_job.program)
self.assertEqual(job.description, expected_job.description)
self.assertEqual(job.hide_from_gui,
expected_job.hide_from_gui)
self.assertEqual(job.popup_on_state_change,
expected_job.popup_on_state_change)
client.disconnect()
def test_request_queue_list_update(self):
client = molequeue.Client()
client.connect_to_server('MoleQueue')
queues = client.request_queue_list_update()
for q in queues:
print q.name, ", ", q.programs;
client.disconnect()
if __name__ == '__main__':
unittest.main()
|
