7 class EventTestBase(object):
8 def on_event(self, ev):
10 self.assertEqual(200, ev['status'])
14 self.assertEqual(self.h[u'uuid'], ev[u'object_uuid'])
23 self.subscribed = threading.Event()
24 self.done = threading.Event()
26 run_test_server.authorize_with("admin")
27 self.ws = arvados.events.subscribe(arvados.api('v1'), [['object_uuid', 'is_a', 'arvados#human']], self.on_event, poll_fallback=2)
28 self.assertIsInstance(self.ws, self.WS_TYPE)
29 self.subscribed.wait(10)
30 self.h = arvados.api('v1').humans().create(body={}).execute()
32 self.assertEqual(3, self.state)
34 class WebsocketTest(run_test_server.TestCaseWithServers, EventTestBase):
35 MAIN_SERVER = {'websockets': True}
36 WS_TYPE = arvados.events.EventClient
41 super(WebsocketTest, self).tearDown()
44 class PollClientTest(run_test_server.TestCaseWithServers, EventTestBase):
46 WS_TYPE = arvados.events.PollClient
51 super(PollClientTest, self).tearDown()