1

import unittest

     2

     3

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

     4

from qmsk.net.lib.event2.event import event

     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

    23

class TestEventBase (unittest.TestCase) :

    24

    """

    25

        Simple event_base bits

    26

    """

    27

    28

    def setUp (self) :

    29

        self.ev_base = event_base()

    30

    31

    def test_loop_empty (self) :

    32

        """

    33

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

    34

        """

    35

    36

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

    37

    38

    def test_loop_empty_nb (self) :

    39

        """

    40

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

    41

        """

    42

    43

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

    44

    45

class TestEvent (unittest.TestCase) :

    46

    """

    47

        Simple event bits.

    48

    """

    49

    50

    def setUp (self) :

    51

        self.ev_base = event_base()

    52

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

    53

    54

    def test_add (self) :

    55

        """

    56

            .add() works.

    57

        """

    58

    59

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

    60

    61

        self.ev.add()

    62

    63

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

    64

    65

    def test_add_timeout (self) :

    66

        """

    67

            .add() with timeout works

    68

        """

    69

    70

        self.ev.add(1.1)

    71

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

    72

    73

    def test_add_del (self) :

    74

        """

    75

            .add() and then .delete() works

    76

        """

    77

    78

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

    79

    80

        # add it in

    81

        self.ev.add()

    82

    83

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

    84

    85

        # ensure that the ev_base is indeed not empty

    86

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

    87

    88

    89

        # remove it again

    90

        self.ev.delete()

    91

    92

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

    93

    94

        # ensure that the ev_base is indeed empty

    95

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

    96

    97

    def test_invalid_base (self) :

    98

        """

    99

            Building an event with invalid base (None) fails

   100

        """

   101

   102

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

   103

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

   104

   105

    def test_illegal_fd (self) :

   106

        """

   107

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

   108

        """

   109

   110

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

   111

   112

    def test_invalid_fd (self) :

   113

        """

   114

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

   115

        """

   116

   117

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

   118

   119

        # XXX: be more specific about Exception

   120

        self.assertRaises(Exception, ev.add)

   121

   122

    def test_invalid_mask_zero (self) :

   123

        """

   124

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

   125

        """

   126

   127

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

   128

        ev.add()

   129

   130

class TestEventPipe (unittest.TestCase) :

   131

    """

   132

        Test basic event stuff using a pipe.

   133

    """

   134

   135

    def setUp (self) :

   136

        self.ev_base = event_base()

   137

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

   138

   139

    def test_event (self) :

   140

        """

   141

            Constructing an event works

   142

        """

   143

   144

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

   145

   146

    def test_read (self) :

   147

        """

   148

            Read event on pipe with data fires

   149

        """

   150

   151

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

   152

        ev.add(0.1) # 100ms

   153

   154

        # trigger

   155

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

   156

   157

        # loop

   158

        self.ev_base.loop(nonblock=True)

   159

   160

        # test

   161

        self.assertEquals(ev.ev_out, EV_READ)

   162

   163

    def test_read_timeout (self) :

   164

        """

   165

            Read event on empty pipe timeouts

   166

        """

   167

   168

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

   169

        ev.add(0.0)

   170

   171

        # loop

   172

        self.ev_base.loop(once=True)

   173

   174

        # test

   175

        self.assertEquals(ev.ev_out, EV_TIMEOUT)

   176

   177

    def test_read_nonblock (self) :

   178

        """

   179

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

   180

        """

   181

   182

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

   183

        ev.add()

   184

   185

        # loop once

   186

        self.ev_base.loop(nonblock=True)

   187

   188

        # test

   189

        self.assertEquals(ev.ev_out, None)

   190

   191

if __name__ == '__main__' :

   192

    unittest.main()