| 1 | #!/usr/bin/env python3
|
| 2 | """
|
| 3 | py_fanos_test.py: Tests for py_fanos.py
|
| 4 | """
|
| 5 | import socket
|
| 6 | import sys
|
| 7 | import unittest
|
| 8 |
|
| 9 | import py_fanos # module under test
|
| 10 |
|
| 11 |
|
| 12 | class FanosTest(unittest.TestCase):
|
| 13 |
|
| 14 | def testSendReceive(self):
|
| 15 | left, right = socket.socketpair()
|
| 16 |
|
| 17 | py_fanos.send(left, b'foo')
|
| 18 |
|
| 19 | fd_out = []
|
| 20 | msg = py_fanos.recv(right, fd_out=fd_out)
|
| 21 | self.assertEqual(b'foo', msg)
|
| 22 | self.assertEqual([], fd_out)
|
| 23 |
|
| 24 | py_fanos.send(left, b'spam', [sys.stdin.fileno(), sys.stdout.fileno(), sys.stderr.fileno()])
|
| 25 |
|
| 26 | msg = py_fanos.recv(right, fd_out=fd_out)
|
| 27 | self.assertEqual(b'spam', msg)
|
| 28 | self.assertEqual(3, len(fd_out))
|
| 29 | print(fd_out)
|
| 30 |
|
| 31 | left.close()
|
| 32 | msg = py_fanos.recv(right)
|
| 33 | self.assertEqual(None, msg) # Valid EOF
|
| 34 |
|
| 35 | right.close()
|
| 36 |
|
| 37 |
|
| 38 | class InvalidMessageTests(unittest.TestCase):
|
| 39 | """COPIED to native/fanos_test.py."""
|
| 40 |
|
| 41 | def testInvalidColon(self):
|
| 42 | left, right = socket.socketpair()
|
| 43 |
|
| 44 | left.send(b':') # Should be 3:foo,
|
| 45 | try:
|
| 46 | msg = py_fanos.recv(right)
|
| 47 | except ValueError as e:
|
| 48 | print(type(e))
|
| 49 | print(e)
|
| 50 | else:
|
| 51 | self.fail('Expected failure')
|
| 52 |
|
| 53 | left.close()
|
| 54 | right.close()
|
| 55 |
|
| 56 | def testInvalidDigits(self):
|
| 57 | left, right = socket.socketpair()
|
| 58 |
|
| 59 | left.send(b'34') # EOF in the middle of length
|
| 60 | left.close()
|
| 61 | try:
|
| 62 | msg = py_fanos.recv(right)
|
| 63 | except ValueError as e:
|
| 64 | print(type(e))
|
| 65 | print(e)
|
| 66 | else:
|
| 67 | self.fail('Expected failure')
|
| 68 |
|
| 69 | right.close()
|
| 70 |
|
| 71 | def testInvalidMissingColon(self):
|
| 72 | left, right = socket.socketpair()
|
| 73 |
|
| 74 | left.send(b'34foo') # missing colon
|
| 75 |
|
| 76 | left.close()
|
| 77 | try:
|
| 78 | msg = py_fanos.recv(right)
|
| 79 | except ValueError as e:
|
| 80 | print(type(e))
|
| 81 | print(e)
|
| 82 | else:
|
| 83 | self.fail('Expected failure')
|
| 84 |
|
| 85 | right.close()
|
| 86 |
|
| 87 | def testInvalidMissingComma(self):
|
| 88 | left, right = socket.socketpair()
|
| 89 |
|
| 90 | # Short payload BLOCKS indefinitely?
|
| 91 | #left.send(b'3:fo')
|
| 92 |
|
| 93 | left.send(b'3:foo') # missing comma
|
| 94 |
|
| 95 | left.close()
|
| 96 | try:
|
| 97 | msg = py_fanos.recv(right)
|
| 98 | except ValueError as e:
|
| 99 | print(type(e))
|
| 100 | print(e)
|
| 101 | else:
|
| 102 | self.fail('Expected failure')
|
| 103 |
|
| 104 | right.close()
|
| 105 |
|
| 106 |
|
| 107 | if __name__ == '__main__':
|
| 108 | unittest.main()
|