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
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
|
#!/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_basic_sigusr1(self):
self.sig(SIGUSR1)
self.expect('ALIVE\r\n')
self.test_exit()
def test_sigusr1_when_entering_command(self):
self.worker.send('pin')
self.sig(SIGUSR1)
self.expect('pinALIVE\r\n')
self.sendlines(['g'])
self.expect('ALIVE\r\n' + self.prompt(1))
self.test_exit()
def test_sigusr1_on_readparms(self):
self.worker.send('exec\nech')
self.sig(SIGUSR1)
self.worker.send('o lol\n')
self.expect('exec\r\nechALIVE\r\no lol\r\n'+
self.__exec_echo_lol_output__ + self.prompt(1))
self.test_exit()
def test_exec_continues_after_sigusr1(self):
self.worker.send('exec\nsleep 0.1; exec echo lol\n')
self.sig(SIGUSR1)
self.expect('exec\r\nsleep 0.1; exec echo lol\r\nALIVE\r\n'+
self.__exec_echo_lol_output__ + 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()
|
