Reformat using black
This commit is contained in:
parent
efee4ad081
commit
a87fb49918
221 changed files with 19127 additions and 7310 deletions
|
@ -4,7 +4,7 @@ import struct
|
|||
import time
|
||||
import zlib
|
||||
|
||||
size_pack = struct.Struct('!I')
|
||||
size_pack = struct.Struct("!I")
|
||||
|
||||
|
||||
def open_connection():
|
||||
|
@ -13,69 +13,70 @@ def open_connection():
|
|||
|
||||
|
||||
def zlibify(data):
|
||||
data = zlib.compress(data.encode('utf-8'))
|
||||
data = zlib.compress(data.encode("utf-8"))
|
||||
return size_pack.pack(len(data)) + data
|
||||
|
||||
|
||||
def dezlibify(data, skip_head=True):
|
||||
if skip_head:
|
||||
data = data[size_pack.size:]
|
||||
return zlib.decompress(data).decode('utf-8')
|
||||
data = data[size_pack.size :]
|
||||
return zlib.decompress(data).decode("utf-8")
|
||||
|
||||
|
||||
def main():
|
||||
global host, port
|
||||
import argparse
|
||||
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument('-l', '--host', default='localhost')
|
||||
parser.add_argument('-p', '--port', default=9999, type=int)
|
||||
parser.add_argument("-l", "--host", default="localhost")
|
||||
parser.add_argument("-p", "--port", default=9999, type=int)
|
||||
args = parser.parse_args()
|
||||
host, port = args.host, args.port
|
||||
|
||||
print('Opening idle connection:', end=' ')
|
||||
print("Opening idle connection:", end=" ")
|
||||
s1 = open_connection()
|
||||
print('Success')
|
||||
print('Opening hello world connection:', end=' ')
|
||||
print("Success")
|
||||
print("Opening hello world connection:", end=" ")
|
||||
s2 = open_connection()
|
||||
print('Success')
|
||||
print('Sending Hello, World!', end=' ')
|
||||
s2.sendall(zlibify('Hello, World!'))
|
||||
print('Success')
|
||||
print('Testing blank connection:', end=' ')
|
||||
print("Success")
|
||||
print("Sending Hello, World!", end=" ")
|
||||
s2.sendall(zlibify("Hello, World!"))
|
||||
print("Success")
|
||||
print("Testing blank connection:", end=" ")
|
||||
s3 = open_connection()
|
||||
s3.close()
|
||||
print('Success')
|
||||
print("Success")
|
||||
result = dezlibify(s2.recv(1024))
|
||||
assert result == 'Hello, World!'
|
||||
assert result == "Hello, World!"
|
||||
print(result)
|
||||
s2.close()
|
||||
print('Large random data test:', end=' ')
|
||||
print("Large random data test:", end=" ")
|
||||
s4 = open_connection()
|
||||
data = os.urandom(1000000).decode('iso-8859-1')
|
||||
print('Generated', end=' ')
|
||||
data = os.urandom(1000000).decode("iso-8859-1")
|
||||
print("Generated", end=" ")
|
||||
s4.sendall(zlibify(data))
|
||||
print('Sent', end=' ')
|
||||
result = b''
|
||||
print("Sent", end=" ")
|
||||
result = b""
|
||||
while len(result) < size_pack.size:
|
||||
result += s4.recv(1024)
|
||||
size = size_pack.unpack(result[:size_pack.size])[0]
|
||||
result = result[size_pack.size:]
|
||||
size = size_pack.unpack(result[: size_pack.size])[0]
|
||||
result = result[size_pack.size :]
|
||||
while len(result) < size:
|
||||
result += s4.recv(1024)
|
||||
print('Received', end=' ')
|
||||
print("Received", end=" ")
|
||||
assert dezlibify(result, False) == data
|
||||
print('Success')
|
||||
print("Success")
|
||||
s4.close()
|
||||
print('Test malformed connection:', end=' ')
|
||||
print("Test malformed connection:", end=" ")
|
||||
s5 = open_connection()
|
||||
s5.sendall(data[:100000].encode('utf-8'))
|
||||
s5.sendall(data[:100000].encode("utf-8"))
|
||||
s5.close()
|
||||
print('Success')
|
||||
print('Waiting for timeout to close idle connection:', end=' ')
|
||||
print("Success")
|
||||
print("Waiting for timeout to close idle connection:", end=" ")
|
||||
time.sleep(6)
|
||||
print('Done')
|
||||
print("Done")
|
||||
s1.close()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue