compn-test.py
1 |
|
2 |
Copyright 2016 Maarten Vangeneugden |
3 |
|
4 |
Licensed under the Apache License, Version 2.0 (the "License"); |
5 |
you may not use this file except in compliance with the License. |
6 |
You may obtain a copy of the License at |
7 |
|
8 |
http://www.apache.org/licenses/LICENSE-2.0 |
9 |
|
10 |
Unless required by applicable law or agreed to in writing, software |
11 |
distributed under the License is distributed on an "AS IS" BASIS, |
12 |
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
13 |
See the License for the specific language governing permissions and |
14 |
limitations under the License. |
15 |
|
16 |
Contact me at: maarten.vangeneugden@student.uhasselt.be |
17 |
""" |
18 |
|
19 |
# That was the legal hurr durr fluff. Now for the actual program: |
20 |
from __future__ import print_function |
21 |
# READ THIS IF YOU USE THIS TEST FOR THE FIRST TIME. |
22 |
# |
23 |
# This test was designed to test the interactive server assignment. It will run a |
24 |
# series of tests to check whether the program works as supposed to. |
25 |
# |
26 |
# This test is built to deal with edge cases, extreme input, and attempts to |
27 |
# break the server. If your server passes this test, you can be assured it's |
28 |
# (relatively speaking) robust enough, even though I cannot make 100% |
29 |
# guarantees. |
30 |
# |
31 |
# To run this test, start the server, and then run this file. |
32 |
# |
33 |
# There are a couple of flags you can set: |
34 |
# -1: Let the tests run for the first server. EXPERIMENTAL, may not work! |
35 |
# -2: Default. Let the tests run for the 'Interactive information server' task. |
36 |
# --verbose: Print all details. |
37 |
# |
38 |
# This test sends large bogus strings to the server (which may have lengths of |
39 |
# up to 100). Be sure to accept enough bytes on your server in order to not |
40 |
# cause the data to queue up too much. (socket.recv(2048) should suffice) |
41 |
# You can also set the max string length in the functions that create them. |
42 |
# |
43 |
# By Maarten Vangeneugden - 1438256 |
44 |
|
45 |
import socket as SocketModule # Altered namespacing for readability. |
46 |
import random # For server input randomization |
47 |
import string # For 'em wacky bogus strings bitch |
48 |
import time # Time control. |
49 |
import itertools # For generating command permutations. |
50 |
import sys # For command-line argument handling. |
51 |
|
52 |
# Constants and program settings (Not necessarily constant, although you can |
53 |
# treat them as such) |
54 |
CLIENT_IP = "127.0.0.1" # If not on localhost, change accordingly. |
55 |
SERVER_ADDRESS = "localhost" |
56 |
SERVER_PORT = 2000 |
57 |
SERVER = 2 # The server to test for. 1 is the first, 2 the interactive one. |
58 |
# The verbose mode can be turned on by the user using the --verbose flag, and |
59 |
# will print all details. Defaults to False, printing only the most necessary |
60 |
# data. |
61 |
VERBOSE_MODE = False |
62 |
RECONNECTIONS_AMOUNT = 2 # The amount of times the client will connect. |
63 |
# The amount of seconds to wait in order to trigger a timeout. |
64 |
# Don't make this as low as possible, but take a reasonable time. 5 is a nice |
65 |
# minimum. |
66 |
TIMEOUT_WAIT = 8 # SET THIS MANUALLY if your server takes longer before timeout. |
67 |
|
68 |
# Functions |
69 |
def verbose_print(string): |
70 |
""" Prints the given string to the output, but only if the user has set the |
71 |
verbose mode. """ |
72 |
if VERBOSE_MODE: |
73 |
print(string) |
74 |
|
75 |
def flag_handling(arguments): |
76 |
""" Checks the given flags, and sets the program's settings accordingly. """ |
77 |
# Settings that can be altered: |
78 |
global SERVER |
79 |
global VERBOSE_MODE |
80 |
|
81 |
for argument in arguments: |
82 |
if argument == "--verbose": # Activate verbose mode |
83 |
VERBOSE_MODE = True |
84 |
print("Verbose mode activated.") |
85 |
elif argument == "-1": |
86 |
SERVER = 1 |
87 |
print("Set the server to test to 1.") |
88 |
elif argument == "-2": |
89 |
SERVER = 2 |
90 |
print("Set the server to interactive (2).") |
91 |
|
92 |
def run_after_erronous_input(socket): |
93 |
""" Sends a series of random bogus commands, and checks whether the server |
94 |
keeps running. """ |
95 |
# Change these settings to your likings if you want to, but be aware that |
96 |
# extreme inputs (like strings of length 10000) may cause problems that |
97 |
# aren't really there. |
98 |
TEST_AMOUNT = 100 # The amount of bogus strings that will be generated. |
99 |
MAX_STRING_LENGTH = 100 # The maximum length of the bogus strings. |
100 |
no_error = True |
101 |
|
102 |
for i in range(TEST_AMOUNT): |
103 |
string_length = random.randint(0, MAX_STRING_LENGTH) |
104 |
bogus_command = \ |
105 |
''.join(random.SystemRandom().choice(string.printable) for _ in range(string_length)) |
106 |
if bogus_command != "TIME" \ |
107 |
and bogus_command != "IP" \ |
108 |
and bogus_command != "EXIT": # Just in case, but imagine the chance... |
109 |
try: |
110 |
verbose_print("Testing if "+ bogus_command +" makes the server \ |
111 |
crash...") |
112 |
socket.send(bogus_command) |
113 |
if SERVER == 2: |
114 |
socket.send("\n") |
115 |
socket.recv(2048) # Receiving what's being sent to discard it. |
116 |
except: |
117 |
print("The server would not accept any more commands after \ |
118 |
receiving the following command: " + bogus_command) |
119 |
no_error = False |
120 |
break # Abort looping the for loop |
121 |
|
122 |
return no_error |
123 |
|
124 |
def check_command_validity(socket, command, answer): |
125 |
""" Initiates a series of tests that will attempt to make the server respond |
126 |
invalid when a correct command sequence is sent, or respond valid when a |
127 |
false command is sent. |
128 |
""" |
129 |
assert command_handled_correctly(socket, command, answer) |
130 |
assert command_misspelled_reject(socket, command, answer) |
131 |
|
132 |
def command_handled_correctly(socket, command, answer): |
133 |
""" Checks whether the server handles the given command correctly. """ |
134 |
if SERVER == 1: # Only test the first command. |
135 |
socket.send(command) |
136 |
response = socket.recv(2048) |
137 |
if response == time.ctime(): |
138 |
return True |
139 |
else: |
140 |
print("When sending "+ command +", the expected answer was "+ \ |
141 |
answer +", but "+ response +" was returned by the server.") |
142 |
return False |
143 |
|
144 |
elif SERVER == 2: # Interactive server |
145 |
command_sequences = generate_correct_commands(command) |
146 |
for sequence in command_sequences: |
147 |
verbose_print("Checking if "+ str(sequence) +" is accepted...") |
148 |
for part in sequence: |
149 |
socket.send(part) |
150 |
socket.send("\n") # Executes the command. |
151 |
response = socket.recv(2048) |
152 |
if response != answer: |
153 |
""" NOTE: It can happen that the time is off by 1 second, |
154 |
because of the time it takes to run tests. If so, this condition |
155 |
will update the answer accordingly. |
156 |
""" |
157 |
if response == time.ctime(): |
158 |
answer = time.ctime() |
159 |
else: |
160 |
print("When sending "+ command +", the expected answer was "+ \ |
161 |
answer +", but "+ response +" was returned by the server.") |
162 |
print("This sequence was sent to the server:") |
163 |
print(sequence) |
164 |
return False |
165 |
else: |
166 |
verbose_print(str(sequence) +" returned a valid response.") |
167 |
|
168 |
# This stage is only reached when the command is handled properly. |
169 |
return True |
170 |
|
171 |
def command_misspelled_reject(socket, correct_command, answer): |
172 |
""" Sends a series of misspelled variations of an accepted command, in an |
173 |
attempt to trick the server in returning a response to said command, which |
174 |
in fact, should be rejected. |
175 |
""" |
176 |
false_commands = generate_false_commands(correct_command) |
177 |
trailing_space_false_commands = [" " + correct_command, correct_command + " "] |
178 |
false_commands.extend(trailing_space_false_commands) |
179 |
while correct_command in false_commands: # FIXME |
180 |
false_commands.remove(correct_command) |
181 |
|
182 |
for false_command in false_commands: |
183 |
verbose_print("Checking if "+ false_command +" is rejected...") |
184 |
socket.send(false_command) |
185 |
if SERVER == 2: |
186 |
socket.send("\n") |
187 |
""" Note that waiting for an answer still breaks when nothing is |
188 |
returned, because: |
189 |
"Incorrect commandos entered by the client should be ignored, and an |
190 |
error must be returned to the client." ~Assignment |
191 |
""" |
192 |
response = socket.recv(2048) |
193 |
if answer == response and false_command != correct_command: |
194 |
print(false_command +" was accepted, which should be rejected!") |
195 |
return False # The server responded to a false command. |
196 |
else: |
197 |
verbose_print(false_command +" successfully rejected!") |
198 |
|
199 |
return True # Not a single false command was accepted. |
200 |
|
201 |
def generate_correct_commands(correct_command): |
202 |
""" Generates all sequences of the given command that, if passed to the |
203 |
interactive server, should yield a correct response. |
204 |
For example: Sending "ABC" makes this function return |
205 |
[ |
206 |
["ABC"], |
207 |
["A", "BC"], |
208 |
["AB", "C"], |
209 |
["A", "B", "C"] |
210 |
] |
211 |
""" |
212 |
correct_commands = [[correct_command]] # Trivial list. |
213 |
for i in range(1, len(correct_command)): |
214 |
# So this is actually kind of like Lisp, but it does the job: Decide |
215 |
# first and rest of the command, and then continue the function |
216 |
# recursively on the rest, concatenating that result on the first, and |
217 |
# appending all that to the list of correct commands. |
218 |
first = [correct_command[:i]] # From begin to i |
219 |
rest = correct_command[i:] # From i to end |
220 |
for recursive_command in generate_correct_commands(rest): |
221 |
correct_commands.append(first + recursive_command) |
222 |
|
223 |
# And finally, return that to the caller: |
224 |
return correct_commands |
225 |
|
226 |
def generate_false_commands(correct_command): |
227 |
""" Generates a list of false commands, similar to the given command. |
228 |
Basically, it creates a permutation of the casing, divided in lists, and so |
229 |
on. |
230 |
It may not be exhaustive, that would actually make running the test run far |
231 |
too long. |
232 |
""" |
233 |
false_commands = [] # This will contain all false commands. |
234 |
if len(correct_command) > 1: |
235 |
# Recursively handle commands less 1 character: |
236 |
false_commands.extend(generate_false_commands( \ |
237 |
correct_command[0:len(correct_command)-1])) |
238 |
false_commands.extend(generate_false_commands( \ |
239 |
correct_command[1:])) |
240 |
# Handling all possible variations of the current length: |
241 |
temp_false_commands = [] |
242 |
for element in itertools.permutations(correct_command, \ |
243 |
len(correct_command)): |
244 |
temp_false_commands.append("".join(element)) |
245 |
|
246 |
swapped_false_commands = [] |
247 |
for false_command in temp_false_commands: |
248 |
for i in range(len(false_command)): |
249 |
new_false_command = false_command[0:i] + \ |
250 |
false_command[i:].swapcase() |
251 |
swapped_false_commands.append(new_false_command) |
252 |
temp_false_commands.extend(swapped_false_commands) |
253 |
|
254 |
# Putting all false commands together: |
255 |
false_commands.extend(temp_false_commands) |
256 |
return false_commands # And returning them all. |
257 |
|
258 |
|
259 |
|
260 |
|
261 |
# --- Start of program --- |
262 |
|
263 |
|
264 |
|
265 |
|
266 |
if len(sys.argv) >= 2: # If the user has set flags, handle them: |
267 |
flag_handling(sys.argv) |
268 |
|
269 |
""" |
270 |
The test will run as follows: |
271 |
1. A predetermined amount of times, a socket will be created, that will try |
272 |
to connect to the server using the given data. |
273 |
2. Random strings will be sent, in an attempt to make the server shut down. |
274 |
3. The client will try different ways to receive the time of the server. |
275 |
4. The client will timeout, and look if the server reacts properly. |
276 |
5. The client will check for IP address data. |
277 |
6. The socket will be closed, and reopened by the loop. |
278 |
7. If the loop ends, a last connection will be started, to shut the server |
279 |
down. |
280 |
8. When that succeeds, all tests have passed. |
281 |
""" |
282 |
|
283 |
for i in range(RECONNECTIONS_AMOUNT): |
284 |
print("Client test (#" + str(i+1) +")") |
285 |
# Creating the socket for use: |
286 |
socket = SocketModule.socket(SocketModule.AF_INET, SocketModule.SOCK_STREAM) |
287 |
socket.connect((SERVER_ADDRESS, SERVER_PORT)) |
288 |
|
289 |
# Testing bogus input: |
290 |
print("Testing if the server breaks on bogus input...") |
291 |
assert run_after_erronous_input(socket) |
292 |
print("Server seems to not crash on erronous input.") |
293 |
|
294 |
# Testing time output: |
295 |
# XXX: It's very likely that the speed of the time test overlaps with the |
296 |
# previous test, causing the server to block TIME. A sleep of 1 should handle |
297 |
# this. |
298 |
time.sleep(1) |
299 |
print("Testing if time gets printed correctly...") |
300 |
check_command_validity(socket, "TIME", time.ctime()) |
301 |
print("Time seems to be handled fine.") |
302 |
|
303 |
# Testing if we can connect after a timeout: |
304 |
print("Testing if timing out works correctly...") |
305 |
# A simple test: Test if it doesn't time out when it shouldn't, and test if |
306 |
# it times out when it should. |
307 |
verbose_print("Test whether the server can be tricked in timing out after 2\ |
308 |
seconds...") |
309 |
time.sleep(2) # This should NOT yield a timeout. |
310 |
socket.send("I didn't time out!") |
311 |
if SERVER == 2: |
312 |
socket.send("\n") |
313 |
data = socket.recv(2048) |
314 |
assert data != "", "The client timed out, but that wasn't supposed to \ |
315 |
happen after 2 seconds!" |
316 |
|
317 |
verbose_print("Server didn't time out after 2 seconds.") |
318 |
verbose_print("Now we will trigger a timeout.") |
319 |
|
320 |
time.sleep(TIMEOUT_WAIT) |
321 |
data = socket.recv(2048) |
322 |
if data == "": # TCP sends "" when a disconnect takes place. |
323 |
print("Timed out as expected!") |
324 |
socket = SocketModule.socket(SocketModule.AF_INET, SocketModule.SOCK_STREAM) |
325 |
socket.connect((SERVER_ADDRESS, SERVER_PORT)) |
326 |
else: |
327 |
assert False, "The program was still connected to the server. The \ |
328 |
client should have timed out by this point, but didn't." |
329 |
|
330 |
|
331 |
# Testing IP data: |
332 |
time.sleep(1) |
333 |
print("Testing if the client's address is sent correctly...") |
334 |
check_command_validity(socket, "IP", CLIENT_IP) |
335 |
print("Addresses seem to be handled fine.") |
336 |
|
337 |
|
338 |
socket.close() # Closing the socket. |
339 |
print("Client test #"+ str(i+1) +" passed!") |
340 |
|
341 |
# Final test: Tell the server to close down, and assert that we can't connect |
342 |
# anymore. |
343 |
print("Checking if the server doesn't exit on wrong EXIT commands...") |
344 |
socket = SocketModule.socket(SocketModule.AF_INET, SocketModule.SOCK_STREAM) |
345 |
socket.connect((SERVER_ADDRESS, SERVER_PORT)) |
346 |
assert command_misspelled_reject(socket, "EXIT", None) |
347 |
# Now, we close the server down, and attempt to connect. |
348 |
socket.send("EXIT\n") |
349 |
socket.close() |
350 |
|
351 |
try: |
352 |
socket.connect((SERVER_ADDRESS, SERVER_PORT)) |
353 |
assert False, "The program was able to connect to the server, despite \ |
354 |
having sent a shutdown command." |
355 |
except SocketModule.error: |
356 |
print("Server shut down on command.") |
357 |
|
358 |
# At this stage, all loops have been done, and all tests have passed. |
359 |
print("All tests passed!") |
360 |