import unittest
from qmsk.net.lib.event2.base import event_base
from qmsk.net.lib.event2.event import event, CallbackEvent
from qmsk.net.lib.event2.constants import *
import os
class MyEvent (event) :
def __init__ (self, base, fd, ev) :
event.__init__(self, base, fd, ev)
self.ev_in = ev
self.ev_out = None
self.res = None
def __call__ (self, fd, mask) :
self.ev_out = mask
self.res = (mask == self.ev_in)
class TestEventBase (unittest.TestCase) :
"""
Simple event_base bits
"""
def setUp (self) :
self.ev_base = event_base()
def test_loop_empty (self) :
"""
loop() on an empty event_base returns False and doesn't block
"""
self.assertFalse(self.ev_base.loop(nonblock=False, once=False))
def test_loop_empty_nb (self) :
"""
loop(nonblock=False) on an empty event_base returns False
"""
self.assertFalse(self.ev_base.loop(nonblock=True))
class TestEvent (unittest.TestCase) :
"""
Simple event bits.
"""
def setUp (self) :
self.ev_base = event_base()
self.ev = event(self.ev_base, 0, EV_READ)
def test_add (self) :
"""
.add() works.
"""
self.assertFalse(self.ev.pending(EV_READ))
self.ev.add()
self.assertTrue(self.ev.pending(EV_READ))
def test_add_timeout (self) :
"""
.add() with timeout works
"""
self.ev.add(1.1)
self.assertTrue(self.ev.pending())
def test_add_del (self) :
"""
.add() and then .delete() works
"""
self.assertFalse(self.ev.pending())
# add it in
self.ev.add()
self.assertTrue(self.ev.pending())
# ensure that the ev_base is indeed not empty
self.assertTrue(self.ev_base.loop(nonblock=True))
# remove it again
self.ev.delete()
self.assertFalse(self.ev.pending())
# ensure that the ev_base is indeed empty
self.assertFalse(self.ev_base.loop(nonblock=True))
def test_invalid_base (self) :
"""
Building an event with invalid base (None) fails
"""
# XXX: this doesn't actually seem to check against None, just the type..?
self.assertRaises(TypeError, event, None, 0, EV_READ)
def test_illegal_fd (self) :
"""
Building an event with an illegal fd (-1) fails
"""
self.assertRaises(ValueError, event, self.ev_base, -1, EV_READ)
def test_invalid_fd (self) :
"""
Building an event with an invalid fd (>= 0) fails when calling .add()
"""
ev = event(self.ev_base, 666, EV_READ)
# XXX: be more specific about Exception
self.assertRaises(Exception, ev.add)
def test_invalid_mask_zero (self) :
"""
Building an event with an invalid mask (0)... XXX: works?
"""
ev = event(self.ev_base, 0, 0)
ev.add()
def test_init_nonalive (self) :
"""
An event is not alive after init
"""
ev = event(self.ev_base, 0, 0)
self.assertFalse(ev.alive)
def test_add_alive (self) :
"""
An event is considered alive and pending after add
"""
ev = event(self.ev_base, 0, EV_READ)
ev.add()
self.assertTrue(ev.alive)
def test_del_nonalive (self) :
"""
An event is not considered alive anymore after del
"""
ev = event(self.ev_base, 0, EV_READ)
ev.add()
ev.delete()
self.assertFalse(ev.alive)
def test_fire_nonalive (self) :
"""
An event is not considered alive anymore during callback execution the callback fires, nor afterwards
"""
fired = []
def cb (ev, what) :
fired.append(1)
self.assertFalse(ev.alive)
ev = CallbackEvent(self.ev_base, 0, EV_READ, cb)
ev.add(-1)
self.ev_base.loop()
# fired
self.assertTrue(fired)
# not alive anymore
self.assertFalse(ev.alive)
def test_fire_add_alive (self) :
"""
An event is considered alive if it is re-added during callback exec
"""
ev = CallbackEvent(self.ev_base, 0, EV_READ,
lambda ev, what: ev.add()
)
ev.add(0)
self.ev_base.loop(once=True, nonblock=True)
# still alive
self.assertTrue(ev.alive)
def test_persistent_fire_alive (self) :
"""
A persistent event is still considered alive during and after callback exec
"""
ev = CallbackEvent(self.ev_base, 0, EV_READ,
lambda ev, what: self.assertTrue(ev.alive)
)
ev.add()
self.ev_base.loop(once=True, nonblock=True)
# persists -> stays alive
self.assertTrue(ev.alive)
class TestEventPipe (unittest.TestCase) :
"""
Test basic event stuff using a pipe.
"""
def setUp (self) :
self.ev_base = event_base()
self.fd_read, self.fd_write = os.pipe()
def test_event (self) :
"""
Constructing an event works
"""
self.assertTrue(event(self.ev_base, self.fd_read, EV_READ))
def test_read (self) :
"""
Read event on pipe with data fires
"""
ev = MyEvent(self.ev_base, self.fd_read, EV_READ)
ev.add(0.1) # 100ms
# trigger
os.write(self.fd_write, "foo")
# loop
self.ev_base.loop(nonblock=True)
# test
self.assertEquals(ev.ev_out, EV_READ)
def test_read_timeout (self) :
"""
Read event on empty pipe timeouts
"""
ev = MyEvent(self.ev_base, self.fd_read, EV_READ)
ev.add(0.0)
# loop
self.ev_base.loop(once=True)
# test
self.assertEquals(ev.ev_out, EV_TIMEOUT)
def test_read_nonblock (self) :
"""
Read event on empty pipe doesn't block loop with nonblock=True
"""
ev = MyEvent(self.ev_base, self.fd_read, EV_READ)
ev.add()
# loop once
self.ev_base.loop(nonblock=True)
# test
self.assertEquals(ev.ev_out, None)
if __name__ == '__main__' :
unittest.main()