1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
|
#!/usr/bin/env python
import unittest
import pexpect
from re import search, escape
from os import kill
from signal import SIGINT, SIGUSR1
class WorkerTests(unittest.TestCase):
__exec_echo_lol_output__ = '0\r\n4\r\nlol\r\n\r\n0\r\n\r\n'
def prompt(self, n):
return ('#%i#\r\n' % n)
def setUp(self):
self.worker = pexpect.spawn('python ../worker/worker.py')
# worker should always be very fast to span
self.expect(self.prompt(0))
def sig(self, sig):
kill(self.worker.pid, sig)
def lose_expect(self, v, timeout = 3):
self.worker.expect(v, timeout)
def expect(self, v, timeout = 3):
self.lose_expect(v, timeout)
self.assertEqual(self.worker.before, '')
def send(self, text):
self.worker.send(text)
self.expect(escape(text))
def sendlines(self, lines):
for line in lines:
self.worker.send(line+'\n')
for line in lines:
self.expect(escape(line)+'\r\n')
def __finishes_ok__(self):
self.expect(pexpect.EOF)
self.worker.close()
self.assertEqual(self.worker.exitstatus, 0)
def test_exit(self):
self.sendlines(['exit'])
self.__finishes_ok__()
def test_ctrlc_on_cmd_prompt_quits(self):
self.sig(SIGINT)
self.expect(pexpect.EOF)
self.__finishes_ok__()
def test_ctrlc_when_entering_command_quits(self):
# "Mon clavier se blo" -- French reference
self.send('glo')
self.sig(SIGINT)
self.expect(pexpect.EOF)
def test_ctrlc_on_readparms_drops(self):
self.sendlines(['exec'])
self.sig(SIGINT)
self.expect(self.prompt(0))
self.sendlines(['glob'])
self.sig(SIGINT)
self.expect(self.prompt(0))
def test_basic_noop(self):
self.sendlines(['noop'])
self.expect(self.prompt(1))
self.test_exit()
def test_basic_ping(self):
self.sendlines(['ping'])
self.expect('ALIVE\r\n' + self.prompt(1))
self.test_exit()
def test_basic_glob(self):
self.sendlines(['glob', '/*bin'])
self.expect('2\r\n(/bin\r\n/sbin|/sbin\r\n/bin)\r\n'+self.prompt(1))
self.test_exit()
def test_empty_glob(self):
self.sendlines(['glob', '/?kyzh?'])
self.expect('0\r\n'+self.prompt(1))
self.test_exit()
def test_increasing_counter(self):
for req_counter in range(1, 5):
self.sendlines(['noop'])
self.expect(self.prompt(req_counter))
for req_counter in range(5, 10):
self.sendlines(['ping'])
self.expect('ALIVE\r\n'+self.prompt(req_counter))
self.test_exit()
def test_queuecommands(self):
self.worker.send('ping\n'*5)
self.worker.send('exec\necho lol\n'*5)
for req_counter in range(1,6):
self.lose_expect('ALIVE\r\n'+self.prompt(req_counter))
for req_counter in range(6,11):
self.lose_expect(self.__exec_echo_lol_output__+self.prompt(req_counter))
self.test_exit()
if __name__ == '__main__':
unittest.main()
|
