def testQueueTasklet(self):
    def f(unused_number, unused_string, unused_a, unused_b): return 1

    def g(unused_number, unused_string): return 2

    def h(unused_c, unused_d): return 3
    t_before = time.time()
    eventloop.queue_call(1, f, 42, 'hello', unused_a=1, unused_b=2)
    eventloop.queue_call(3, h, unused_c=3, unused_d=4)
    eventloop.queue_call(2, g, 100, 'abc')
    t_after = time.time()
    self.assertEqual(len(self.ev.queue), 3)
    [(t1, f1, a1, k1), (t2, f2, a2, k2), (t3, f3, a3, k3)] = self.ev.queue
    self.assertTrue(t1 < t2)
    self.assertTrue(t3 < t2)
    self.assertTrue(abs(t1 - (t_before + 1)) <= t_after - t_before)
    self.assertTrue(abs(t2 - (t_before + 2)) <= t_after - t_before)
    self.assertTrue(abs(t3 - (t_before + 3)) <= t_after - t_before)
    self.assertEqual(f1, f)
    self.assertEqual(f2, g)
    self.assertEqual(f3, h)
    self.assertEqual(a1, (42, 'hello'))
    self.assertEqual(a2, (100, 'abc'))
    self.assertEqual(a3, ())
    self.assertEqual(k1, {'unused_a': 1, 'unused_b': 2})
    self.assertEqual(k2, {})
    self.assertEqual(k3, {'unused_c': 3, 'unused_d': 4})
    # Delete queued events (they would fail or take a long time).
    ev = eventloop.get_event_loop()
    ev.queue = []
    ev.rpcs = {}
dataset/ETHPy150Open GoogleCloudPlatform/datastore-ndb-python/ndb/eventloop_test.py/EventLoopTests.testQueueTasklet/swapped operands of `<`