1

import unittest

     2

     3

from qmsk.net.lib.event2.base import event_base

     5

from qmsk.net.lib.event2.constants import *

     6

     7

import os

     8

     9

class MyEvent (event) :

    10

    def __init__ (self, base, fd, ev) :

    11

        event.__init__(self, base, fd, ev)

    12

    13

        self.ev_in = ev

    14

        self.ev_out = None

    15

    16

        self.res = None

    17

    18

    def __call__ (self, fd, mask) :

    19

        self.ev_out = mask

    20

    21

        self.res = (mask == self.ev_in)

    22

    24

class TestEventBase (unittest.TestCase) :

    25

    """

    26

        Simple event_base bits

    27

    """

    28

    29

    def setUp (self) :

    30

        self.ev_base = event_base()

    31

    32

    def test_loop_empty (self) :

    33

        """

    34

            loop() on an empty event_base returns False and doesn't block

    35

        """

    36

    37

        self.assertFalse(self.ev_base.loop(nonblock=False, once=False))

    38

    39

    def test_loop_empty_nb (self) :

    40

        """

    41

            loop(nonblock=False) on an empty event_base returns False

    42

        """

    43

    44

        self.assertFalse(self.ev_base.loop(nonblock=True))

    45

    47

class TestEvent (unittest.TestCase) :

    48

    """

    49

        Simple event bits.

    50

    """

    51

    52

    def setUp (self) :

    53

        self.ev_base = event_base()

    54

        self.ev = event(self.ev_base, 0, EV_READ)

    55

    56

    def test_add (self) :

    57

        """

    58

            .add() works.

    59

        """

    60

    61

        self.assertFalse(self.ev.pending(EV_READ))

    62

    63

        self.ev.add()

    64

    65

        self.assertTrue(self.ev.pending(EV_READ))

    66

    67

    def test_add_timeout (self) :

    68

        """

    69

            .add() with timeout works

    70

        """

    71

    72

        self.ev.add(1.1)

    73

        self.assertTrue(self.ev.pending())

    74

    75

    def test_add_del (self) :

    76

        """

    77

            .add() and then .delete() works

    78

        """

    79

    80

        self.assertFalse(self.ev.pending())

    81

    82

        # add it in

    83

        self.ev.add()

    84

    85

        self.assertTrue(self.ev.pending())

    86

    87

        # ensure that the ev_base is indeed not empty

    88

        self.assertTrue(self.ev_base.loop(nonblock=True))

    89

    90

    91

        # remove it again

    92

        self.ev.delete()

    93

    94

        self.assertFalse(self.ev.pending())

    95

    96

        # ensure that the ev_base is indeed empty

    97

        self.assertFalse(self.ev_base.loop(nonblock=True))

    98

    99

    def test_invalid_base (self) :

   100

        """

   101

            Building an event with invalid base (None) fails

   102

        """

   103

   104

        # XXX: this doesn't actually seem to check against None, just the type..?

   105

        self.assertRaises(TypeError, event, None, 0, EV_READ)

   106

   107

    def test_illegal_fd (self) :

   108

        """

   109

            Building an event with an illegal fd (-1) fails

   110

        """

   111

   112

        self.assertRaises(ValueError, event, self.ev_base, -1, EV_READ)

   113

   114

    def test_invalid_fd (self) :

   115

        """

   116

            Building an event with an invalid fd (>= 0) fails when calling .add()

   117

        """

   118

   119

        ev = event(self.ev_base, 666, EV_READ)

   120

   121

        # XXX: be more specific about Exception

   122

        self.assertRaises(Exception, ev.add)

   123

   124

    def test_invalid_mask_zero (self) :

   125

        """

   126

            Building an event with an invalid mask (0)... XXX: works?

   127

        """

   128

   129

        ev = event(self.ev_base, 0, 0)

   130

        ev.add()

   218

   219

class TestEventPipe (unittest.TestCase) :

   220

    """

   221

        Test basic event stuff using a pipe.

   222

    """

   223

   224

    def setUp (self) :

   225

        self.ev_base = event_base()

   226

        self.fd_read, self.fd_write = os.pipe()

   227

   228

    def test_event (self) :

   229

        """

   230

            Constructing an event works

   231

        """

   232

   233

        self.assertTrue(event(self.ev_base, self.fd_read, EV_READ))

   234

   235

    def test_read (self) :

   236

        """

   237

            Read event on pipe with data fires

   238

        """

   239

   240

        ev = MyEvent(self.ev_base, self.fd_read, EV_READ)

   241

        ev.add(0.1) # 100ms

   242

   243

        # trigger

   244

        os.write(self.fd_write, "foo")

   245

   246

        # loop

   247

        self.ev_base.loop(nonblock=True)

   248

   249

        # test

   250

        self.assertEquals(ev.ev_out, EV_READ)

   251

   252

    def test_read_timeout (self) :

   253

        """

   254

            Read event on empty pipe timeouts

   255

        """

   256

   257

        ev = MyEvent(self.ev_base, self.fd_read, EV_READ)

   258

        ev.add(0.0)

   259

   260

        # loop

   261

        self.ev_base.loop(once=True)

   262

   263

        # test

   264

        self.assertEquals(ev.ev_out, EV_TIMEOUT)

   265

   266

    def test_read_nonblock (self) :

   267

        """

   268

            Read event on empty pipe doesn't block loop with nonblock=True

   269

        """

   270

   271

        ev = MyEvent(self.ev_base, self.fd_read, EV_READ)

   272

        ev.add()

   273

   274

        # loop once

   275

        self.ev_base.loop(nonblock=True)

   276

   277

        # test

   278

        self.assertEquals(ev.ev_out, None)

   279

   280

if __name__ == '__main__' :

   281

    unittest.main()