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