test/util/tunchannel.py

   1 #!/usr/bin/env python
   2
   3 from nepi.util import tunchannel
   4 import socket
   5 import time
   6 import threading
   7 import unittest
   8
   9 class TunnChannelTestCase(unittest.TestCase):
  10     def test_send_suspend_terminate(self):
  11         def tun_fwd(local, remote, TERMINATE, SUSPEND, STOPPED):
  12             tunchannel.tun_fwd(local, remote, True, True, None, True,
  13                 TERMINATE, SUSPEND, None)
  14             STOPPED.append(None)
  15
  16         TERMINATE = []
  17         SUSPEND = []
  18         STOPPED = []
  19
  20         s1, s2 = socket.socketpair()
  21         s3, s4 = socket.socketpair()
  22         s4.settimeout(2.0)
  23
  24         t = threading.Thread(target=tun_fwd, args=[s2, s3, TERMINATE, SUSPEND, STOPPED])
  25         t.start()
  26
  27         txt = "0000|received"
  28         s1.send(txt)
  29         rtxt = s4.recv(len(txt))
  30
  31         self.assertTrue(rtxt == txt[4:])
  32
  33         # Let's try to suspend execution now
  34         cond = threading.Condition()
  35         SUSPEND.insert(0, cond)
  36
  37         txt = "0000|suspended"
  38         s1.send(txt)
  39
  40         rtxt = "timeout"
  41         try:
  42             rtxt = s4.recv(len(txt))
  43         except socket.timeout:
  44             pass
  45
  46         self.assertTrue(rtxt == "timeout")
  47
  48         # Let's see if we can resume and receive the message
  49         cond = SUSPEND[0]
  50         SUSPEND.remove(cond)
  51         cond.acquire()
  52         cond.notify()
  53         cond.release()
  54
  55         rtxt = s4.recv(len(txt))
  56         self.assertTrue(rtxt == txt[4:])
  57
  58         # Stop forwarding
  59         TERMINATE.append(None)
  60
  61         txt = "0000|never received"
  62         s1.send(txt)
  63
  64         rtxt = "timeout"
  65         try:
  66             rtxt = s4.recv(len(txt))
  67         except socket.timeout:
  68             pass
  69
  70         self.assertTrue(rtxt == "timeout")
  71         self.assertTrue(STOPPED)
  72
  73 if __name__ == '__main__':
  74     unittest.main()
  75