import unittest

from qmsk.net.lib.event2.base import event_base
from qmsk.net.lib.event2.event import event
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()

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()