[LABI] Added directory with mid-semester project
This commit is contained in:
parent
568765c1ce
commit
6db8c56290
|
|
@ -0,0 +1,11 @@
|
||||||
|
# labiaprof
|
||||||
|
Trabalho de aprofundamento de laboratórios de informática
|
||||||
|
|
||||||
|
## Grupo
|
||||||
|
### Rúben Gomes
|
||||||
|
* 113435
|
||||||
|
* rlcg@ua.pt
|
||||||
|
|
||||||
|
### Tiago Garcia
|
||||||
|
* 114184
|
||||||
|
* tiago.rgarcia@ua.pt
|
||||||
|
|
@ -0,0 +1,394 @@
|
||||||
|
#!/usr/bin/python3
|
||||||
|
import os
|
||||||
|
import re
|
||||||
|
import sys
|
||||||
|
import socket
|
||||||
|
import json
|
||||||
|
import base64
|
||||||
|
from common_comm import send_dict, recv_dict, sendrecv_dict
|
||||||
|
|
||||||
|
from Crypto.Cipher import AES
|
||||||
|
from Crypto.Hash import SHA256
|
||||||
|
|
||||||
|
|
||||||
|
class Tcolors:
|
||||||
|
HEADER = '\033[95m'
|
||||||
|
OKBLUE = '\033[94m'
|
||||||
|
OKCYAN = '\033[96m'
|
||||||
|
OKGREEN = '\033[92m'
|
||||||
|
WARNING = '\033[93m'
|
||||||
|
FAIL = '\033[91m'
|
||||||
|
ENDC = '\033[0m'
|
||||||
|
BOLD = '\033[1m'
|
||||||
|
UNDERLINE = '\033[4m'
|
||||||
|
|
||||||
|
|
||||||
|
# Function to encript values for sending in json format
|
||||||
|
# return int data encrypted in a 16 bytes binary string coded in base64
|
||||||
|
def encrypt_intvalue(cipherkey, data):
|
||||||
|
key = base64.b64decode(cipherkey)
|
||||||
|
cipher = AES.new(key, AES.MODE_ECB)
|
||||||
|
data = cipher.encrypt(bytes("%16d" % data, "utf8"))
|
||||||
|
return str(base64.b64encode(data), "utf8")
|
||||||
|
|
||||||
|
|
||||||
|
# Function to decript values received in json format
|
||||||
|
# return int data decrypted from a 16 bytes binary strings coded in base64
|
||||||
|
def decrypt_intvalue(cipherkey, data_arg):
|
||||||
|
cipher = AES.new(base64.b64decode(cipherkey), AES.MODE_ECB)
|
||||||
|
data = base64.b64decode(data_arg)
|
||||||
|
data = cipher.decrypt(data)
|
||||||
|
return int(str(data, "utf8"))
|
||||||
|
|
||||||
|
|
||||||
|
# verify if response from server is valid or is an error message and act accordingly - já está implementada
|
||||||
|
def validate_response(client_sock, response):
|
||||||
|
if not response["status"]:
|
||||||
|
print(f"{Tcolors.FAIL}Error: {response['error']}{Tcolors.ENDC}")
|
||||||
|
client_sock.close()
|
||||||
|
sys.exit(3)
|
||||||
|
|
||||||
|
|
||||||
|
# process QUIT operation
|
||||||
|
def quit_action(client_sock, has_started):
|
||||||
|
print(f"{Tcolors.ENDC}Quitting...")
|
||||||
|
if has_started:
|
||||||
|
senddata = {"op": "QUIT"}
|
||||||
|
recvdata = sendrecv_dict(client_sock, senddata)
|
||||||
|
|
||||||
|
try:
|
||||||
|
# status = False
|
||||||
|
if not recvdata["status"]:
|
||||||
|
print(f"{Tcolors.ENDC}{Tcolors.FAIL}Error: {recvdata['error']}{Tcolors.ENDC}")
|
||||||
|
return
|
||||||
|
except TypeError:
|
||||||
|
print(f"{Tcolors.ENDC}{Tcolors.FAIL}Error: an error occurred with the server.{Tcolors.ENDC}")
|
||||||
|
print(f"{Tcolors.ENDC}{Tcolors.WARNING}Client not removed from server, quitting...{Tcolors.ENDC}")
|
||||||
|
client_sock.close()
|
||||||
|
exit(1)
|
||||||
|
|
||||||
|
# status = True
|
||||||
|
print(f"{Tcolors.OKGREEN}Client quit with success")
|
||||||
|
client_sock.close()
|
||||||
|
exit(0)
|
||||||
|
|
||||||
|
|
||||||
|
# Outcomming message structure:
|
||||||
|
# { op = "START", client_id, [cipher] }
|
||||||
|
# { op = "QUIT" }
|
||||||
|
# { op = "NUMBER", number }
|
||||||
|
# { op = "STOP", [shasum] }
|
||||||
|
# { op = "GUESS", choice }
|
||||||
|
#
|
||||||
|
# Incomming message structure:
|
||||||
|
# { op = "START", status }
|
||||||
|
# { op = "QUIT" , status }
|
||||||
|
# { op = "NUMBER", status }
|
||||||
|
# { op = "STOP", status, value }
|
||||||
|
# { op = "GUESS", status, result }
|
||||||
|
|
||||||
|
#
|
||||||
|
# Suport for executing the client pretended behaviour
|
||||||
|
#
|
||||||
|
|
||||||
|
# returns a valid number
|
||||||
|
def returnValidNum():
|
||||||
|
while 1:
|
||||||
|
try:
|
||||||
|
num = int(input(f"\n{Tcolors.ENDC}{Tcolors.BOLD}Number > {Tcolors.UNDERLINE}"))
|
||||||
|
except ValueError:
|
||||||
|
print(f"{Tcolors.ENDC}{Tcolors.WARNING}Invalid input{Tcolors.ENDC}")
|
||||||
|
continue
|
||||||
|
break
|
||||||
|
return num
|
||||||
|
|
||||||
|
|
||||||
|
# verify if port is valid
|
||||||
|
def verify_port(port):
|
||||||
|
# verify if port is a number
|
||||||
|
if not port.isdigit():
|
||||||
|
return {"status": False, "error": "Port must be an integer"}
|
||||||
|
# verify if port is between 1024 and 65535
|
||||||
|
if not (1024 <= int(port) <= 65535):
|
||||||
|
return {"status": False, "error": "Port number must be between 1024 and 65535"}
|
||||||
|
return {"status": True, "port": int(port)}
|
||||||
|
|
||||||
|
|
||||||
|
# verify if hostname is valid
|
||||||
|
def verify_hostname(hostname):
|
||||||
|
if hostname == "localhost":
|
||||||
|
return {"status": True}
|
||||||
|
if not (re.match(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$', hostname) and all(0 <= int(n) < 256 for n in hostname.split('.'))):
|
||||||
|
return {"status": False, "error": "Invalid DNS address"}
|
||||||
|
return {"status": True}
|
||||||
|
|
||||||
|
|
||||||
|
def run_client(client_sock, client_id):
|
||||||
|
# Print the welcome message
|
||||||
|
print(f"{Tcolors.OKCYAN}{Tcolors.BOLD}{Tcolors.UNDERLINE}Number characteristics guesser game!{Tcolors.ENDC}\n")
|
||||||
|
|
||||||
|
# client runtime global variables
|
||||||
|
has_stopped = False
|
||||||
|
has_started = False
|
||||||
|
cipherkey = None
|
||||||
|
numbers = []
|
||||||
|
|
||||||
|
while 1:
|
||||||
|
option = input(f"{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}")
|
||||||
|
|
||||||
|
# start option
|
||||||
|
if option.upper() == "START":
|
||||||
|
if has_started:
|
||||||
|
print(f"{Tcolors.ENDC}{Tcolors.WARNING}Client already started\n{Tcolors.ENDC}")
|
||||||
|
continue
|
||||||
|
|
||||||
|
while 1:
|
||||||
|
# ask user if cipher is needed
|
||||||
|
choice = input(f"\n{Tcolors.ENDC}Do you wish to use a cipher? {Tcolors.BOLD}(Y/N)\n> {Tcolors.UNDERLINE}")
|
||||||
|
if choice.upper() == "Y":
|
||||||
|
# create cipher key for server
|
||||||
|
cipherkey = base64.b64encode(os.urandom(16)).decode()
|
||||||
|
break
|
||||||
|
elif choice.upper() == "N":
|
||||||
|
# do nothing since cipher will be None
|
||||||
|
break
|
||||||
|
else:
|
||||||
|
# loop if invalid option
|
||||||
|
print(f"{Tcolors.ENDC}{Tcolors.WARNING}Invalid input{Tcolors.ENDC}")
|
||||||
|
continue
|
||||||
|
|
||||||
|
# send dict and receive response
|
||||||
|
senddata = {"op": "START", "client_id": client_id, "cipher": cipherkey}
|
||||||
|
recvdata = sendrecv_dict(client_sock, senddata)
|
||||||
|
|
||||||
|
try:
|
||||||
|
# status = False
|
||||||
|
if not recvdata["status"]:
|
||||||
|
print(f"{Tcolors.ENDC}{Tcolors.FAIL}Error: {recvdata['error']}{Tcolors.ENDC}")
|
||||||
|
print(f"{Tcolors.ENDC}{Tcolors.WARNING}Client not added, quitting...{Tcolors.ENDC}")
|
||||||
|
client_sock.close()
|
||||||
|
exit(1)
|
||||||
|
except TypeError:
|
||||||
|
print(f"{Tcolors.ENDC}{Tcolors.FAIL}Error: an error occurred with the server, try again later{Tcolors.ENDC}")
|
||||||
|
print(f"{Tcolors.ENDC}{Tcolors.WARNING}Client not added, quitting...{Tcolors.ENDC}")
|
||||||
|
client_sock.close()
|
||||||
|
exit(1)
|
||||||
|
|
||||||
|
# status = True
|
||||||
|
has_started = True
|
||||||
|
print(f"{Tcolors.ENDC}{Tcolors.OKGREEN}\nClient added with success{Tcolors.ENDC}\n")
|
||||||
|
|
||||||
|
elif option.upper() == "QUIT":
|
||||||
|
quit_action(client_sock, has_started)
|
||||||
|
continue
|
||||||
|
|
||||||
|
elif option.upper() == "NUMBER":
|
||||||
|
if not has_started:
|
||||||
|
print(f"{Tcolors.ENDC}{Tcolors.WARNING}You must start the game first\n{Tcolors.ENDC}")
|
||||||
|
continue
|
||||||
|
|
||||||
|
if has_stopped:
|
||||||
|
print(f"{Tcolors.ENDC}{Tcolors.WARNING}You can't add more numbers\n{Tcolors.ENDC}")
|
||||||
|
continue
|
||||||
|
# verify if number is int
|
||||||
|
num = returnValidNum()
|
||||||
|
|
||||||
|
# encrypt the number is a cipher is being used
|
||||||
|
if cipherkey is not None:
|
||||||
|
encrypted_num = encrypt_intvalue(cipherkey, num)
|
||||||
|
else:
|
||||||
|
encrypted_num = num
|
||||||
|
|
||||||
|
# send dict and receive response
|
||||||
|
senddata = {"op": "NUMBER", "number": encrypted_num}
|
||||||
|
recvdata = sendrecv_dict(client_sock, senddata)
|
||||||
|
|
||||||
|
try:
|
||||||
|
# status = False
|
||||||
|
if not recvdata["status"]:
|
||||||
|
print(f"{Tcolors.ENDC}{Tcolors.FAIL}Error: {recvdata['error']}{Tcolors.ENDC}")
|
||||||
|
client_sock.close()
|
||||||
|
continue
|
||||||
|
# status = True
|
||||||
|
numbers.append(num)
|
||||||
|
print(f"{Tcolors.ENDC}{Tcolors.OKGREEN}Number added with success{Tcolors.ENDC}\n")
|
||||||
|
except TypeError:
|
||||||
|
print(f"{Tcolors.ENDC}{Tcolors.FAIL}Error: an error occurred with the server\nSocket has been closed, try to start again{Tcolors.ENDC}")
|
||||||
|
client_sock.close()
|
||||||
|
has_started = False
|
||||||
|
cipherkey = None
|
||||||
|
numbers = []
|
||||||
|
continue
|
||||||
|
|
||||||
|
elif option.upper() == "STOP":
|
||||||
|
# check if client has started the game
|
||||||
|
if not has_started:
|
||||||
|
print(f"{Tcolors.ENDC}{Tcolors.WARNING}You must start the game first\n{Tcolors.ENDC}")
|
||||||
|
continue
|
||||||
|
# check if client has stopped adding numbers
|
||||||
|
if has_stopped:
|
||||||
|
print(f"{Tcolors.ENDC}{Tcolors.WARNING}You can't stop the game again\n{Tcolors.ENDC}")
|
||||||
|
continue
|
||||||
|
|
||||||
|
# creates the synthesis for the number list
|
||||||
|
hasher = SHA256.new()
|
||||||
|
for number in numbers:
|
||||||
|
hasher.update(bytes(str(number), "utf8"))
|
||||||
|
|
||||||
|
# send dict and receive response
|
||||||
|
senddata = {"op": "STOP", "shasum": hasher.hexdigest()}
|
||||||
|
recvdata = sendrecv_dict(client_sock, senddata)
|
||||||
|
|
||||||
|
try:
|
||||||
|
# status = False
|
||||||
|
if not recvdata["status"]:
|
||||||
|
print(f"{Tcolors.ENDC}{Tcolors.FAIL}Error: {recvdata['error']}{Tcolors.ENDC}")
|
||||||
|
continue
|
||||||
|
except TypeError:
|
||||||
|
print(f"{Tcolors.ENDC}{Tcolors.FAIL}Error: an error occurred with the server\nSocket has been closed, try to start again{Tcolors.ENDC}")
|
||||||
|
client_sock.close()
|
||||||
|
has_started = False
|
||||||
|
cipherkey = None
|
||||||
|
numbers = []
|
||||||
|
continue
|
||||||
|
|
||||||
|
# decipher data if using encryption
|
||||||
|
data = recvdata["value"]
|
||||||
|
if cipherkey is not None:
|
||||||
|
data = decrypt_intvalue(cipherkey, data)
|
||||||
|
has_stopped = True
|
||||||
|
# status = True
|
||||||
|
print(f"{Tcolors.ENDC}{Tcolors.OKGREEN}\nChosen number: {Tcolors.UNDERLINE}{data}{Tcolors.ENDC}\n")
|
||||||
|
|
||||||
|
elif option.upper() == "GUESS":
|
||||||
|
# check if client has started the game
|
||||||
|
if not has_started:
|
||||||
|
print(f"{Tcolors.ENDC}{Tcolors.WARNING}You must start the game first\n{Tcolors.ENDC}")
|
||||||
|
continue
|
||||||
|
# check if client has stopped adding numbers
|
||||||
|
if not has_stopped:
|
||||||
|
print(f"{Tcolors.ENDC}{Tcolors.WARNING}You can't guess before stopping the game\n{Tcolors.ENDC}")
|
||||||
|
continue
|
||||||
|
|
||||||
|
# print the possible choices
|
||||||
|
print(f"""
|
||||||
|
{Tcolors.ENDC}Choose one of the following options:
|
||||||
|
1 - first
|
||||||
|
2 - last
|
||||||
|
3 - min
|
||||||
|
4 - max
|
||||||
|
5 - median
|
||||||
|
6 - min, first
|
||||||
|
7 - max, first
|
||||||
|
8 - min, last
|
||||||
|
9 - max, last
|
||||||
|
10 - median, first
|
||||||
|
11 - median, last
|
||||||
|
""")
|
||||||
|
while True:
|
||||||
|
try:
|
||||||
|
choice_num = int(input(f"{Tcolors.BOLD}\n> {Tcolors.UNDERLINE}"))
|
||||||
|
if choice_num == 1:
|
||||||
|
choice = ["first"]
|
||||||
|
elif choice_num == 2:
|
||||||
|
choice = ["last"]
|
||||||
|
elif choice_num == 3:
|
||||||
|
choice = ["min"]
|
||||||
|
elif choice_num == 4:
|
||||||
|
choice = ["max"]
|
||||||
|
elif choice_num == 5:
|
||||||
|
choice = ["median"]
|
||||||
|
elif choice_num == 6:
|
||||||
|
choice = ["min", "first"]
|
||||||
|
elif choice_num == 7:
|
||||||
|
choice = ["max", "first"]
|
||||||
|
elif choice_num == 8:
|
||||||
|
choice = ["min", "last"]
|
||||||
|
elif choice_num == 9:
|
||||||
|
choice = ["max", "last"]
|
||||||
|
elif choice_num == 10:
|
||||||
|
choice = ["median", "first"]
|
||||||
|
elif choice_num == 11:
|
||||||
|
choice = ["median", "last"]
|
||||||
|
else:
|
||||||
|
print(f"{Tcolors.ENDC}{Tcolors.WARNING}Invalid input{Tcolors.ENDC}")
|
||||||
|
continue
|
||||||
|
break
|
||||||
|
except ValueError:
|
||||||
|
print(f"{Tcolors.ENDC}{Tcolors.WARNING}Invalid input{Tcolors.ENDC}")
|
||||||
|
continue
|
||||||
|
|
||||||
|
# send dict and receive response
|
||||||
|
senddata = {"op": "GUESS", "choice": choice}
|
||||||
|
recvdata = sendrecv_dict(client_sock, senddata)
|
||||||
|
|
||||||
|
try:
|
||||||
|
# status = False
|
||||||
|
if not recvdata["status"]:
|
||||||
|
print(f"{Tcolors.ENDC}{Tcolors.FAIL}Error: {recvdata['error']}{Tcolors.ENDC}")
|
||||||
|
continue
|
||||||
|
except TypeError:
|
||||||
|
print(f"{Tcolors.ENDC}{Tcolors.FAIL}Error: an error occurred with the server\nSocket has been closed, try to start again{Tcolors.ENDC}")
|
||||||
|
client_sock.close()
|
||||||
|
has_started = False
|
||||||
|
cipherkey = None
|
||||||
|
numbers = []
|
||||||
|
continue
|
||||||
|
|
||||||
|
# status = True
|
||||||
|
print(f"\n\n{Tcolors.ENDC}{Tcolors.BOLD}{Tcolors.OKBLUE}{'='*15}\n\n{Tcolors.UNDERLINE}{Tcolors.OKCYAN}"
|
||||||
|
+ ("You are right!" if recvdata["result"] else "You are wrong!")
|
||||||
|
+ f"{Tcolors.ENDC}{Tcolors.BOLD}{Tcolors.OKBLUE}\n\n{'='*15}{Tcolors.ENDC}\n\n")
|
||||||
|
quit_action(client_sock, has_started)
|
||||||
|
else:
|
||||||
|
print(f"{Tcolors.ENDC}{Tcolors.WARNING}Invalid option!\n{Tcolors.ENDC}")
|
||||||
|
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def main():
|
||||||
|
# validate the number of arguments and eventually print error message and exit with error
|
||||||
|
# verify type of arguments and eventually print error message and exit with error
|
||||||
|
if len(sys.argv) not in [3, 4]:
|
||||||
|
print(f"{Tcolors.WARNING}Usage: python3 client.py client_id port DNS{Tcolors.ENDC}")
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
# check if indicated port is valid and get its value
|
||||||
|
port = sys.argv[2]
|
||||||
|
verified = verify_port(port)
|
||||||
|
if not verified["status"]:
|
||||||
|
print(f"{Tcolors.WARNING}{verified['error']}{Tcolors.ENDC}")
|
||||||
|
sys.exit(1)
|
||||||
|
port = verified["port"]
|
||||||
|
|
||||||
|
# get the ip address of the DNS and get its value
|
||||||
|
hostname = sys.argv[3] if len(sys.argv) == 4 else socket.gethostbyname(socket.gethostname())
|
||||||
|
verified = verify_hostname(hostname)
|
||||||
|
if not verified["status"]:
|
||||||
|
print(f"{Tcolors.WARNING}{verified['error']}{Tcolors.ENDC}")
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
# create the socket
|
||||||
|
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||||
|
client_socket.bind(("0.0.0.0", 0))
|
||||||
|
|
||||||
|
# catch error message if server does not exist in those specifications
|
||||||
|
print(f"{Tcolors.WARNING}Connecting to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.WARNING}...{Tcolors.ENDC}")
|
||||||
|
try:
|
||||||
|
client_socket.connect((hostname, port))
|
||||||
|
except OSError:
|
||||||
|
print(f"{Tcolors.FAIL}Error: connection to server failed{Tcolors.ENDC}")
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
# send confirmation about the connection
|
||||||
|
print(f"{Tcolors.OKGREEN}Connected to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.OKGREEN} as client {Tcolors.UNDERLINE}{sys.argv[1]}\n{Tcolors.ENDC}")
|
||||||
|
|
||||||
|
# run the client
|
||||||
|
run_client(client_socket, sys.argv[1])
|
||||||
|
|
||||||
|
client_socket.close()
|
||||||
|
sys.exit(0)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
|
|
@ -0,0 +1,82 @@
|
||||||
|
import socket
|
||||||
|
import json
|
||||||
|
import base64
|
||||||
|
|
||||||
|
|
||||||
|
#
|
||||||
|
# Universal function to send a given amount of data to a TCP socket.
|
||||||
|
# It returns True or False, depending on the success of
|
||||||
|
# sending all the data to the socket.
|
||||||
|
#
|
||||||
|
def exact_send(dst, data):
|
||||||
|
try:
|
||||||
|
while len(data) != 0:
|
||||||
|
bytes_sent = dst.send(data)
|
||||||
|
data = data[bytes_sent:]
|
||||||
|
return True
|
||||||
|
except OSError:
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
#
|
||||||
|
# Universal function to receive a given amount of data from a TCP socket.
|
||||||
|
# It returns None or data, depending on the success of
|
||||||
|
# receiving all the required data from the socket.
|
||||||
|
#
|
||||||
|
def exact_recv(src, count):
|
||||||
|
data = bytearray(0)
|
||||||
|
while count != 0:
|
||||||
|
new_data = src.recv(count)
|
||||||
|
|
||||||
|
if len(new_data) == 0: return None
|
||||||
|
|
||||||
|
data += new_data
|
||||||
|
count -= len(new_data)
|
||||||
|
|
||||||
|
return data
|
||||||
|
|
||||||
|
|
||||||
|
#
|
||||||
|
# Universal function to send a dictionary message to a TCP socket.
|
||||||
|
# It actually transmits a JSON object, prefixed by its length (in network byte order).
|
||||||
|
#
|
||||||
|
# The JSON object is created from the dictionary.
|
||||||
|
# It returns True or False, depending on the success of sending the
|
||||||
|
# message to the socket.
|
||||||
|
#
|
||||||
|
def send_dict(dst, msg):
|
||||||
|
# DEBUG print ("Send: %s" % (msg))
|
||||||
|
data = bytes(json.dumps(msg), "utf8")
|
||||||
|
prefixed_data = len(data).to_bytes(4, "big") + data
|
||||||
|
return exact_send(dst, prefixed_data)
|
||||||
|
|
||||||
|
|
||||||
|
#
|
||||||
|
# Universal function to receive a dictionary message from a TCP socket.
|
||||||
|
# It actually receives a JSON object, prefixed by its length (in network byte order).
|
||||||
|
# The dictionary is created from that JSON object.
|
||||||
|
#
|
||||||
|
def recv_dict(src):
|
||||||
|
prefix = exact_recv(src, 4)
|
||||||
|
|
||||||
|
if prefix == None: return None
|
||||||
|
|
||||||
|
length = int.from_bytes(prefix, "big")
|
||||||
|
data = exact_recv(src, length)
|
||||||
|
|
||||||
|
if data == None: return None
|
||||||
|
|
||||||
|
msg = json.loads(str(data, "utf8"))
|
||||||
|
# DEBUG print ("Recv: %s" % (msg))
|
||||||
|
return msg
|
||||||
|
|
||||||
|
|
||||||
|
#
|
||||||
|
# Universal function to send and receive a dictionary to/from a TCP socket peer.
|
||||||
|
# It returns None upon an error.
|
||||||
|
#
|
||||||
|
def sendrecv_dict(peer, msg):
|
||||||
|
if send_dict(peer, msg):
|
||||||
|
return recv_dict(peer)
|
||||||
|
else:
|
||||||
|
return None
|
||||||
|
|
@ -0,0 +1,384 @@
|
||||||
|
#!/usr/bin/python3
|
||||||
|
import sys
|
||||||
|
import socket
|
||||||
|
import select
|
||||||
|
import json
|
||||||
|
import base64
|
||||||
|
import csv
|
||||||
|
import random
|
||||||
|
from common_comm import send_dict, recv_dict, sendrecv_dict
|
||||||
|
|
||||||
|
from Crypto.Cipher import AES
|
||||||
|
from Crypto.Hash import SHA256
|
||||||
|
|
||||||
|
# Dicionário com a informação relativa aos clientes
|
||||||
|
users = {}
|
||||||
|
|
||||||
|
|
||||||
|
# return the client_id of a socket or None
|
||||||
|
def find_client_id(client_sock):
|
||||||
|
for client_id in users:
|
||||||
|
if users[client_id]["socket"] == client_sock:
|
||||||
|
return client_id
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
# Função para encriptar valores a enviar em formato json com codificação base64
|
||||||
|
# return int data encrypted in a 16 bytes binary string and coded base64
|
||||||
|
def encrypt_intvalue(client_id, data_arg):
|
||||||
|
key = base64.b64decode(users[client_id]["cipher"])
|
||||||
|
cipher = AES.new(key, AES.MODE_ECB)
|
||||||
|
data = cipher.encrypt(bytes("%16d" % data_arg, "utf8"))
|
||||||
|
return str(base64.b64encode(data), "utf8")
|
||||||
|
|
||||||
|
|
||||||
|
# Função para desencriptar valores recebidos em formato json com codificação base64
|
||||||
|
# return int data decrypted from a 16 bytes binary string and coded base64
|
||||||
|
def decrypt_intvalue(client_id, data_arg):
|
||||||
|
key = base64.b64decode(users[client_id]["cipher"])
|
||||||
|
cipher = AES.new(key, AES.MODE_ECB)
|
||||||
|
data = base64.b64decode(data_arg)
|
||||||
|
data = cipher.decrypt(data)
|
||||||
|
return int(str(data, "utf8"))
|
||||||
|
|
||||||
|
|
||||||
|
# Função auxiliar para gerar o resultado - já está implementada
|
||||||
|
# return int value and list of description strings identifying the characteristic of the value
|
||||||
|
def generate_result(list_values):
|
||||||
|
if len(list_values) % 2 == 1:
|
||||||
|
test = 4
|
||||||
|
else:
|
||||||
|
test = 3
|
||||||
|
|
||||||
|
minimal = min(list_values)
|
||||||
|
maximal = max(list_values)
|
||||||
|
first = list_values[0]
|
||||||
|
last = list_values[-1]
|
||||||
|
|
||||||
|
choice = random.randint(0, test)
|
||||||
|
if choice == 0:
|
||||||
|
if minimal == first:
|
||||||
|
return first, ["min", "first"]
|
||||||
|
elif maximal == first:
|
||||||
|
return first, ["max", "first"]
|
||||||
|
else:
|
||||||
|
return first, ["first"]
|
||||||
|
elif choice == 1:
|
||||||
|
if minimal == last:
|
||||||
|
return last, ["min", "last"]
|
||||||
|
elif maximal == last:
|
||||||
|
return last, ["max", "last"]
|
||||||
|
else:
|
||||||
|
return last, ["last"]
|
||||||
|
elif choice == 2:
|
||||||
|
if minimal == first:
|
||||||
|
return first, ["min", "first"]
|
||||||
|
elif minimal == last:
|
||||||
|
return last, ["min", "last"]
|
||||||
|
else:
|
||||||
|
return minimal, ["min"]
|
||||||
|
elif choice == 3:
|
||||||
|
if maximal == first:
|
||||||
|
return first, ["max", "first"]
|
||||||
|
elif maximal == last:
|
||||||
|
return last, ["max", "last"]
|
||||||
|
else:
|
||||||
|
return maximal, ["max"]
|
||||||
|
elif choice == 4:
|
||||||
|
list_values.sort()
|
||||||
|
median = list_values[len(list_values) // 2]
|
||||||
|
if median == first:
|
||||||
|
return first, ["median", "first"]
|
||||||
|
elif median == last:
|
||||||
|
return last, ["median", "last"]
|
||||||
|
else:
|
||||||
|
return median, ["median"]
|
||||||
|
else:
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
# Incomming message structure:
|
||||||
|
# { op = "START", client_id, [cipher] }
|
||||||
|
# { op = "QUIT" }
|
||||||
|
# { op = "NUMBER", number }
|
||||||
|
# { op = "STOP", [shasum] }
|
||||||
|
# { op = "GUESS", choice }
|
||||||
|
#
|
||||||
|
# Outcomming message structure:
|
||||||
|
# { op = "START", status }
|
||||||
|
# { op = "QUIT" , status }
|
||||||
|
# { op = "NUMBER", status }
|
||||||
|
# { op = "STOP", status, value }
|
||||||
|
# { op = "GUESS", status, result }
|
||||||
|
|
||||||
|
|
||||||
|
#
|
||||||
|
# Suporte de descodificação da operação pretendida pelo cliente - já está implementada
|
||||||
|
#
|
||||||
|
def new_msg(client_sock):
|
||||||
|
request = recv_dict(client_sock)
|
||||||
|
# print( "Command: %s" % (str(request)) )
|
||||||
|
|
||||||
|
op = request["op"]
|
||||||
|
if op == "START":
|
||||||
|
response = new_client(client_sock, request)
|
||||||
|
elif op == "QUIT": #
|
||||||
|
response = quit_client(client_sock, request)
|
||||||
|
elif op == "NUMBER": #
|
||||||
|
response = number_client(client_sock, request)
|
||||||
|
elif op == "STOP": #
|
||||||
|
response = stop_client(client_sock, request)
|
||||||
|
elif op == "GUESS": #
|
||||||
|
response = guess_client(client_sock, request)
|
||||||
|
else:
|
||||||
|
response = {"op": op, "status": False, "error": "Invalid operation"}
|
||||||
|
|
||||||
|
# print (response)
|
||||||
|
send_dict(client_sock, response)
|
||||||
|
|
||||||
|
|
||||||
|
#
|
||||||
|
# Suporte da criação de um novo cliente - operação START
|
||||||
|
#
|
||||||
|
# detect the client in the request
|
||||||
|
# verify the appropriate conditions for executing this operation
|
||||||
|
# process the client in the dictionary
|
||||||
|
# return response message with or without error message
|
||||||
|
def new_client(client_sock, request):
|
||||||
|
client_id = request["client_id"]
|
||||||
|
|
||||||
|
# check if the client_id is in users
|
||||||
|
if client_id in users:
|
||||||
|
response = {"op": "START", "status": False, "error": "Client already exists"}
|
||||||
|
print("Failed to add client %s\nReason: %s" % (client_id, response["error"]))
|
||||||
|
else:
|
||||||
|
cipher = None
|
||||||
|
# verify if client wants to use cipher
|
||||||
|
if request["cipher"] is not None:
|
||||||
|
cipher = request["cipher"]
|
||||||
|
|
||||||
|
users[client_id] = {"socket": client_sock, "cipher": cipher, "numbers": [], "has_stopped": False}
|
||||||
|
response = {"op": "START", "status": True}
|
||||||
|
print("Client %s added\n" % client_id)
|
||||||
|
return response
|
||||||
|
|
||||||
|
|
||||||
|
#
|
||||||
|
# Suporte da eliminação de um cliente - já está implementada
|
||||||
|
#
|
||||||
|
# obtain the client_id from his socket and delete from the dictionary
|
||||||
|
def clean_client(client_sock):
|
||||||
|
client_id = find_client_id(client_sock)
|
||||||
|
# check if the client_id is in users
|
||||||
|
if client_id is not None:
|
||||||
|
print("Client %s removed\n" % client_id)
|
||||||
|
del users[client_id]
|
||||||
|
|
||||||
|
|
||||||
|
#
|
||||||
|
# Suporte do pedido de desistência de um cliente - operação QUIT
|
||||||
|
#
|
||||||
|
# obtain the client_id from his socket
|
||||||
|
# verify the appropriate conditions for executing this operation
|
||||||
|
# process the report file with the QUIT result
|
||||||
|
# eliminate client from dictionary using the function clean_client
|
||||||
|
# return response message with or without error message
|
||||||
|
def quit_client(client_sock, request):
|
||||||
|
client_id = find_client_id(client_sock)
|
||||||
|
# check if the client_id is in users
|
||||||
|
if client_id is None:
|
||||||
|
response = {"op": "QUIT", "status": False, "error": "Client does not exist"}
|
||||||
|
print("Failed to remove client %s\nReason: %s" % (client_id, response["error"]))
|
||||||
|
else:
|
||||||
|
# remove client
|
||||||
|
clean_client(client_sock)
|
||||||
|
response = {"op": "QUIT", "status": True}
|
||||||
|
return response
|
||||||
|
|
||||||
|
|
||||||
|
#
|
||||||
|
# Suporte da criação de um ficheiro csv com o respetivo cabeçalho - já está implementada
|
||||||
|
#
|
||||||
|
def create_file():
|
||||||
|
with open("result.csv", "w", newline="") as csvfile:
|
||||||
|
columns = ["client_id", "number_of_numbers", "guess"]
|
||||||
|
|
||||||
|
fw = csv.DictWriter(csvfile, delimiter=",", fieldnames=columns)
|
||||||
|
fw.writeheader()
|
||||||
|
|
||||||
|
|
||||||
|
#
|
||||||
|
# Suporte da actualização de um ficheiro csv com a informação do cliente
|
||||||
|
#
|
||||||
|
# update report csv file with the simulation of the client
|
||||||
|
def update_file(client_id, size, guess):
|
||||||
|
with open("result.csv", "a", newline="") as csvfile:
|
||||||
|
writer = csv.DictWriter(csvfile, delimiter=',', fieldnames=["client_id", "number_of_numbers", "guess"])
|
||||||
|
writer.writerow({"client_id": client_id, "number_of_numbers": size, "guess": guess})
|
||||||
|
|
||||||
|
|
||||||
|
#
|
||||||
|
# Suporte do processamento do número de um cliente - operação NUMBER
|
||||||
|
#
|
||||||
|
# obtain the client_id from his socket
|
||||||
|
# verify the appropriate conditions for executing this operation
|
||||||
|
# return response message with or without error message
|
||||||
|
def number_client(client_sock, request):
|
||||||
|
client_id = find_client_id(client_sock)
|
||||||
|
# check if the client_id is in users
|
||||||
|
if client_id is None:
|
||||||
|
response = {"op": "NUMBER", "status": False, "error": "Client does not exist"}
|
||||||
|
print("Failed to add number to client %s\nReason: %s" % (client_id, response["error"]))
|
||||||
|
# check if client has stopped adding numbers
|
||||||
|
elif users[client_id]["has_stopped"]:
|
||||||
|
response = {"op": "NUMBER", "status": False, "error": "Client has stopped"}
|
||||||
|
print("Failed to add number to client %s\nReason: %s" % (client_id, response["error"]))
|
||||||
|
else:
|
||||||
|
num = request["number"]
|
||||||
|
# decrypt the number if a cipher is being used
|
||||||
|
if users[client_id]["cipher"] is not None:
|
||||||
|
num = decrypt_intvalue(client_id, num)
|
||||||
|
|
||||||
|
users[client_id]["numbers"].append(num)
|
||||||
|
response = {"op": "NUMBER", "status": True}
|
||||||
|
print("Number %d added to client %s\n" % (num, client_id))
|
||||||
|
return response
|
||||||
|
|
||||||
|
|
||||||
|
#
|
||||||
|
# Suporte do pedido de terminação de um cliente - operação STOP
|
||||||
|
#
|
||||||
|
# obtain the client_id from his socket
|
||||||
|
# verify the appropriate conditions for executing this operation
|
||||||
|
# randomly generate a value to return using the function generate_result
|
||||||
|
# process the report file with the result
|
||||||
|
# return response message with result or error message
|
||||||
|
def stop_client(client_sock, request):
|
||||||
|
client_id = find_client_id(client_sock)
|
||||||
|
if client_id is None:
|
||||||
|
response = {"op": "STOP", "status": False, "error": "Client does not exist"}
|
||||||
|
print("Failed to stop client %s\nReason: %s" % (client_id, response["error"]))
|
||||||
|
elif len(users[client_id]["numbers"]) < 1:
|
||||||
|
response = {"op": "STOP", "status": False, "error": "Client has not yet sent any number"}
|
||||||
|
print("Failed to stop client %s\nReason: %s" % (client_id, response["error"]))
|
||||||
|
else:
|
||||||
|
# creates the synthesis for the list
|
||||||
|
hasher = SHA256.new()
|
||||||
|
for number in users[client_id]["numbers"]:
|
||||||
|
hasher.update(bytes(str(number), "utf8"))
|
||||||
|
|
||||||
|
# compares the synthesis of the server and the client to see if they match
|
||||||
|
if hasher.hexdigest() != request["shasum"]:
|
||||||
|
response = {"op": "STOP", "status": False, "error": "Server numbers list synthesis doesn't match with client list"}
|
||||||
|
print("Failed to stop client %s\nReason: %s" % (client_id, response["error"]))
|
||||||
|
else:
|
||||||
|
# generates the result
|
||||||
|
value, solution = generate_result(users[client_id]["numbers"])
|
||||||
|
# encrypts the value if a cipher is being used
|
||||||
|
if users[client_id]["cipher"] is not None:
|
||||||
|
encripted_value = encrypt_intvalue(client_id, value)
|
||||||
|
else:
|
||||||
|
encripted_value = value
|
||||||
|
|
||||||
|
update_file(client_id, len(users[client_id]["numbers"]), solution)
|
||||||
|
response = {"op": "STOP", "status": True, "value": encripted_value}
|
||||||
|
users[client_id]["solution"] = solution
|
||||||
|
users[client_id]["has_stopped"] = True
|
||||||
|
print("Client %s stopped\nChosen number: %d\nSolution: %s" % (client_id, value, solution))
|
||||||
|
return response
|
||||||
|
|
||||||
|
|
||||||
|
#
|
||||||
|
# Suporte da adivinha de um cliente - operação GUESS
|
||||||
|
#
|
||||||
|
# obtain the client_id from his socket
|
||||||
|
# verify the appropriate conditions for executing this operation
|
||||||
|
# eliminate client from dictionary
|
||||||
|
# return response message with result or error message
|
||||||
|
def guess_client(client_sock, request):
|
||||||
|
client_id = find_client_id(client_sock)
|
||||||
|
if client_id is None:
|
||||||
|
response = {"op": "GUESS", "status": False, "error": "Client does not exist"}
|
||||||
|
print("Failed to guess client %s\nReason: %s" % (client_id, response["error"]))
|
||||||
|
elif not users[client_id]["has_stopped"]:
|
||||||
|
response = {"op": "GUESS", "status": False, "error": "Client has not yet stopped"}
|
||||||
|
print("Failed to guess client %s\nReason: %s" % (client_id, response["error"]))
|
||||||
|
else:
|
||||||
|
choice = request["choice"]
|
||||||
|
response = {"op": "GUESS", "status": True, "result": choice == users[client_id]["solution"]}
|
||||||
|
print("Client %s guessed %s\n" % (client_id, choice))
|
||||||
|
return response
|
||||||
|
|
||||||
|
|
||||||
|
def verify_port(port):
|
||||||
|
# verify if port is a number
|
||||||
|
if not port.isdigit():
|
||||||
|
return {"status": False, "error": "Port must be an integer"}
|
||||||
|
# verify if port is between 1024 and 65535
|
||||||
|
if not (1024 <= int(port) <= 65535):
|
||||||
|
return {"status": False, "error": "Port number must be between 1024 and 65535"}
|
||||||
|
return {"status": True, "port": int(port)}
|
||||||
|
|
||||||
|
|
||||||
|
def main():
|
||||||
|
# validate the number of arguments and eventually print error message and exit with error
|
||||||
|
# verify type of arguments and eventually print error message and exit with error
|
||||||
|
if len(sys.argv) != 2:
|
||||||
|
print("Usage: python3 %s <port>" % sys.argv[0])
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
# obtain the port number
|
||||||
|
port = sys.argv[1]
|
||||||
|
verified = verify_port(port)
|
||||||
|
if not verified["status"]:
|
||||||
|
print(f"{verified['error']}")
|
||||||
|
sys.exit(1)
|
||||||
|
port = verified["port"]
|
||||||
|
|
||||||
|
# create the server socket
|
||||||
|
try:
|
||||||
|
hostname = socket.gethostbyname(socket.gethostname())
|
||||||
|
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||||
|
server_socket.bind((hostname, port))
|
||||||
|
print(f"Server started.\nHostname: {hostname}\nPort: {port}\n-----------------------------------------------------\n")
|
||||||
|
# handle errors while creating server socket and exit with error
|
||||||
|
except OSError:
|
||||||
|
print(f"Failed to create server socket, maybe the port is already in use?\n")
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
server_socket.listen()
|
||||||
|
|
||||||
|
clients = []
|
||||||
|
create_file()
|
||||||
|
|
||||||
|
while True:
|
||||||
|
try:
|
||||||
|
available = select.select([server_socket] + clients, [], [])[0]
|
||||||
|
except ValueError:
|
||||||
|
# Sockets may have been closed, check for that
|
||||||
|
for client_sock in clients:
|
||||||
|
if client_sock.fileno() == -1:
|
||||||
|
clients.remove(client_sock) # closed
|
||||||
|
continue # Reiterate select
|
||||||
|
|
||||||
|
for client_sock in available:
|
||||||
|
# New client?
|
||||||
|
if client_sock is server_socket:
|
||||||
|
newclient, addr = server_socket.accept()
|
||||||
|
clients.append(newclient)
|
||||||
|
# Or an existing client
|
||||||
|
else:
|
||||||
|
# See if client sent a message
|
||||||
|
if len(client_sock.recv(1, socket.MSG_PEEK)) != 0:
|
||||||
|
# client socket has a message
|
||||||
|
# print ("server" + str (client_sock))
|
||||||
|
new_msg(client_sock)
|
||||||
|
else: # Or just disconnected
|
||||||
|
clients.remove(client_sock)
|
||||||
|
clean_client(client_sock)
|
||||||
|
client_sock.close()
|
||||||
|
break # Reiterate select
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
|
|
@ -0,0 +1,344 @@
|
||||||
|
import socket
|
||||||
|
from subprocess import Popen
|
||||||
|
from subprocess import PIPE
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
|
||||||
|
# class for colors in terminal
|
||||||
|
class Tcolors:
|
||||||
|
HEADER = '\033[95m'
|
||||||
|
OKBLUE = '\033[94m'
|
||||||
|
OKCYAN = '\033[96m'
|
||||||
|
OKGREEN = '\033[92m'
|
||||||
|
WARNING = '\033[93m'
|
||||||
|
FAIL = '\033[91m'
|
||||||
|
ENDC = '\033[0m'
|
||||||
|
BOLD = '\033[1m'
|
||||||
|
UNDERLINE = '\033[4m'
|
||||||
|
|
||||||
|
|
||||||
|
def test_arguments():
|
||||||
|
# start of testing lack of or too many sys.argv
|
||||||
|
proc = Popen("python3 client.py", stdout=PIPE, shell=True)
|
||||||
|
output = proc.stdout.read()
|
||||||
|
|
||||||
|
assert output == f"{Tcolors.WARNING}Usage: python3 client.py client_id port DNS{Tcolors.ENDC}\n".encode("utf-8")
|
||||||
|
|
||||||
|
proc = Popen("python3 client.py test", stdout=PIPE, shell=True)
|
||||||
|
output = proc.stdout.read().decode("utf-8")
|
||||||
|
|
||||||
|
assert output == f"{Tcolors.WARNING}Usage: python3 client.py client_id port DNS{Tcolors.ENDC}\n"
|
||||||
|
|
||||||
|
proc = Popen("python3 client.py test 2000 123.245.14.25 123", stdout=PIPE, shell=True)
|
||||||
|
output = proc.stdout.read().decode("utf-8")
|
||||||
|
|
||||||
|
assert output == f"{Tcolors.WARNING}Usage: python3 client.py client_id port DNS{Tcolors.ENDC}\n"
|
||||||
|
# end of testing lack of or too many sys.argv
|
||||||
|
|
||||||
|
|
||||||
|
def test_invalid_port():
|
||||||
|
# start of testing invalid port
|
||||||
|
proc = Popen("python3 client.py test 1000", stdout=PIPE, shell=True)
|
||||||
|
output = proc.stdout.read().decode("utf-8")
|
||||||
|
|
||||||
|
assert output == f"{Tcolors.WARNING}Port number must be between 1024 and 65535{Tcolors.ENDC}\n"
|
||||||
|
|
||||||
|
proc = Popen("python3 client.py test 1000000", stdout=PIPE, shell=True)
|
||||||
|
output = proc.stdout.read().decode("utf-8")
|
||||||
|
|
||||||
|
assert output == f"{Tcolors.WARNING}Port number must be between 1024 and 65535{Tcolors.ENDC}\n"
|
||||||
|
|
||||||
|
proc = Popen("python3 client.py test test", stdout=PIPE, shell=True)
|
||||||
|
output = proc.stdout.read().decode("utf-8")
|
||||||
|
|
||||||
|
assert output == f"{Tcolors.WARNING}Port must be an integer{Tcolors.ENDC}\n"
|
||||||
|
# end of testing invalid port
|
||||||
|
|
||||||
|
|
||||||
|
def test_invalid_ip():
|
||||||
|
# start of testing invalid IP address
|
||||||
|
proc = Popen("python3 client.py test 2000 2154", stdout=PIPE, shell=True)
|
||||||
|
output = proc.stdout.read().decode("utf-8")
|
||||||
|
assert output == f"{Tcolors.WARNING}Invalid DNS address{Tcolors.ENDC}\n"
|
||||||
|
|
||||||
|
proc = Popen("python3 client.py test 2000 256.256.256.256", stdout=PIPE, shell=True)
|
||||||
|
output = proc.stdout.read().decode("utf-8")
|
||||||
|
assert output == f"{Tcolors.WARNING}Invalid DNS address{Tcolors.ENDC}\n"
|
||||||
|
|
||||||
|
proc = Popen("python3 client.py test 2000 255.255.str.255", stdout=PIPE, shell=True)
|
||||||
|
output = proc.stdout.read().decode("utf-8")
|
||||||
|
|
||||||
|
assert output == f"{Tcolors.WARNING}Invalid DNS address{Tcolors.ENDC}\n"
|
||||||
|
# end of testing invalid IP address
|
||||||
|
|
||||||
|
|
||||||
|
def test_invalid_connection():
|
||||||
|
# start of testing an invalid connection to server
|
||||||
|
proc = Popen("python3 client.py test 2040", stdout=PIPE, shell=True)
|
||||||
|
output = proc.stdout.read().decode("utf-8")
|
||||||
|
hostname = socket.gethostbyname(socket.gethostname())
|
||||||
|
port = proc.args.split(" ")[3]
|
||||||
|
|
||||||
|
assert output == f"""{Tcolors.WARNING}Connecting to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.WARNING}...{Tcolors.ENDC}
|
||||||
|
{Tcolors.FAIL}Error: connection to server failed{Tcolors.ENDC}\n"""
|
||||||
|
# end of testing an invalid connection to server
|
||||||
|
|
||||||
|
|
||||||
|
def test_valid_connection():
|
||||||
|
# start of testing a valid connection to server
|
||||||
|
server = Popen("python3 server.py 2000", stdout=PIPE, shell=True)
|
||||||
|
client_test = Popen("python3 client.py test 2000", stdout=PIPE, shell=True)
|
||||||
|
output = client_test.stdout.read().decode("utf-8")
|
||||||
|
port = client_test.args.split(" ")[3]
|
||||||
|
hostname = socket.gethostbyname(socket.gethostname())
|
||||||
|
|
||||||
|
client_test.terminate()
|
||||||
|
server.terminate()
|
||||||
|
|
||||||
|
assert output == f"""{Tcolors.WARNING}Connecting to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.WARNING}...{Tcolors.ENDC}
|
||||||
|
{Tcolors.OKGREEN}Connected to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.OKGREEN} as client {Tcolors.UNDERLINE}{client_test.args.split(" ")[2]}\n{Tcolors.ENDC}
|
||||||
|
{Tcolors.OKCYAN}{Tcolors.BOLD}{Tcolors.UNDERLINE}Number characteristics guesser game!{Tcolors.ENDC}
|
||||||
|
|
||||||
|
{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}"""
|
||||||
|
# end of testing a valid connection to server
|
||||||
|
|
||||||
|
|
||||||
|
def test_start_option():
|
||||||
|
# start of testing the START option
|
||||||
|
server = Popen("python3 server.py 3000", stdout=PIPE, shell=True, close_fds=True)
|
||||||
|
client_test = Popen("python3 client.py test 3000", stdout=PIPE, stdin=PIPE, shell=True)
|
||||||
|
port = client_test.args.split(" ")[3]
|
||||||
|
hostname = socket.gethostbyname(socket.gethostname())
|
||||||
|
|
||||||
|
output = client_test.communicate(input=b"start\ny")[0]
|
||||||
|
|
||||||
|
client_test.terminate()
|
||||||
|
server.terminate()
|
||||||
|
|
||||||
|
assert output == f"""{Tcolors.WARNING}Connecting to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.WARNING}...{Tcolors.ENDC}
|
||||||
|
{Tcolors.OKGREEN}Connected to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.OKGREEN} as client {Tcolors.UNDERLINE}{client_test.args.split(" ")[2]}\n{Tcolors.ENDC}
|
||||||
|
{Tcolors.OKCYAN}{Tcolors.BOLD}{Tcolors.UNDERLINE}Number characteristics guesser game!{Tcolors.ENDC}
|
||||||
|
|
||||||
|
{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}
|
||||||
|
{Tcolors.ENDC}Do you wish to use a cipher? {Tcolors.BOLD}(Y/N)\n> {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.OKGREEN}\nClient added with success{Tcolors.ENDC}
|
||||||
|
|
||||||
|
{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}""".encode("utf-8")
|
||||||
|
# end of testing the START option
|
||||||
|
|
||||||
|
|
||||||
|
def test_quit_option():
|
||||||
|
# start of testing the QUIT option
|
||||||
|
server = Popen("python3 server.py 4000", stdout=PIPE, shell=True, close_fds=True)
|
||||||
|
client_test = Popen("python3 client.py test 4000", stdout=PIPE, stdin=PIPE, shell=True)
|
||||||
|
port = client_test.args.split(" ")[3]
|
||||||
|
hostname = socket.gethostbyname(socket.gethostname())
|
||||||
|
|
||||||
|
output = client_test.communicate(input=b"quit")[0]
|
||||||
|
|
||||||
|
client_test.terminate()
|
||||||
|
server.terminate()
|
||||||
|
|
||||||
|
assert output == f"""{Tcolors.WARNING}Connecting to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.WARNING}...{Tcolors.ENDC}
|
||||||
|
{Tcolors.OKGREEN}Connected to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.OKGREEN} as client {Tcolors.UNDERLINE}{client_test.args.split(" ")[2]}\n{Tcolors.ENDC}
|
||||||
|
{Tcolors.OKCYAN}{Tcolors.BOLD}{Tcolors.UNDERLINE}Number characteristics guesser game!{Tcolors.ENDC}
|
||||||
|
|
||||||
|
{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}{Tcolors.ENDC}Quitting...
|
||||||
|
{Tcolors.OKGREEN}Client quit with success\n""".encode("utf-8")
|
||||||
|
# end of testing the QUIT option
|
||||||
|
|
||||||
|
|
||||||
|
def test_number_option():
|
||||||
|
# start of testing the NUMBER option
|
||||||
|
server = Popen("python3 server.py 5000", stdout=PIPE, shell=True, close_fds=True)
|
||||||
|
client_test = Popen("python3 client.py test 5000", stdout=PIPE, stdin=PIPE, shell=True)
|
||||||
|
port = client_test.args.split(" ")[3]
|
||||||
|
hostname = socket.gethostbyname(socket.gethostname())
|
||||||
|
|
||||||
|
output = client_test.communicate(input=b"start\ny\nnumber\n200")[0]
|
||||||
|
|
||||||
|
client_test.terminate()
|
||||||
|
server.terminate()
|
||||||
|
|
||||||
|
assert output == f"""{Tcolors.WARNING}Connecting to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.WARNING}...{Tcolors.ENDC}
|
||||||
|
{Tcolors.OKGREEN}Connected to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.OKGREEN} as client {Tcolors.UNDERLINE}{client_test.args.split(" ")[2]}\n{Tcolors.ENDC}
|
||||||
|
{Tcolors.OKCYAN}{Tcolors.BOLD}{Tcolors.UNDERLINE}Number characteristics guesser game!{Tcolors.ENDC}
|
||||||
|
|
||||||
|
{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}
|
||||||
|
{Tcolors.ENDC}Do you wish to use a cipher? {Tcolors.BOLD}(Y/N)\n> {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.OKGREEN}\nClient added with success{Tcolors.ENDC}
|
||||||
|
|
||||||
|
{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}
|
||||||
|
{Tcolors.ENDC}{Tcolors.BOLD}Number > {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.OKGREEN}Number added with success{Tcolors.ENDC}
|
||||||
|
|
||||||
|
{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}""".encode("utf-8")
|
||||||
|
# end of testing the NUMBER option
|
||||||
|
|
||||||
|
|
||||||
|
def test_guess_without_stopping():
|
||||||
|
# start of testing the GUESS option without a number
|
||||||
|
server = Popen("python3 server.py 6000", stdout=PIPE, shell=True, close_fds=True)
|
||||||
|
client_test = Popen("python3 client.py test 6000", stdout=PIPE, stdin=PIPE, shell=True)
|
||||||
|
port = client_test.args.split(" ")[3]
|
||||||
|
hostname = socket.gethostbyname(socket.gethostname())
|
||||||
|
|
||||||
|
output = client_test.communicate(input=b"start\ny\nguess")[0]
|
||||||
|
|
||||||
|
client_test.terminate()
|
||||||
|
server.terminate()
|
||||||
|
|
||||||
|
assert output == f"""{Tcolors.WARNING}Connecting to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.WARNING}...{Tcolors.ENDC}
|
||||||
|
{Tcolors.OKGREEN}Connected to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.OKGREEN} as client {Tcolors.UNDERLINE}{client_test.args.split(" ")[2]}\n{Tcolors.ENDC}
|
||||||
|
{Tcolors.OKCYAN}{Tcolors.BOLD}{Tcolors.UNDERLINE}Number characteristics guesser game!{Tcolors.ENDC}
|
||||||
|
|
||||||
|
{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}
|
||||||
|
{Tcolors.ENDC}Do you wish to use a cipher? {Tcolors.BOLD}(Y/N)\n> {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.OKGREEN}\nClient added with success{Tcolors.ENDC}
|
||||||
|
|
||||||
|
{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.WARNING}You can't guess before stopping the game\n{Tcolors.ENDC}
|
||||||
|
{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}""".encode("utf-8")
|
||||||
|
# end of testing the GUESS option without a number
|
||||||
|
|
||||||
|
|
||||||
|
def test_stop_option():
|
||||||
|
# start of testing the STOP option
|
||||||
|
server = Popen("python3 server.py 7000", stdout=PIPE, shell=True, close_fds=True)
|
||||||
|
client_test = Popen("python3 client.py test 7000", stdout=PIPE, stdin=PIPE, shell=True)
|
||||||
|
port = client_test.args.split(" ")[3]
|
||||||
|
hostname = socket.gethostbyname(socket.gethostname())
|
||||||
|
|
||||||
|
output = client_test.communicate(input=b"start\ny\nnumber\n200\nstop")[0]
|
||||||
|
|
||||||
|
data = 200
|
||||||
|
|
||||||
|
client_test.terminate()
|
||||||
|
server.terminate()
|
||||||
|
|
||||||
|
assert output == f"""{Tcolors.WARNING}Connecting to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.WARNING}...{Tcolors.ENDC}
|
||||||
|
{Tcolors.OKGREEN}Connected to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.OKGREEN} as client {Tcolors.UNDERLINE}{client_test.args.split(" ")[2]}\n{Tcolors.ENDC}
|
||||||
|
{Tcolors.OKCYAN}{Tcolors.BOLD}{Tcolors.UNDERLINE}Number characteristics guesser game!{Tcolors.ENDC}
|
||||||
|
|
||||||
|
{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}
|
||||||
|
{Tcolors.ENDC}Do you wish to use a cipher? {Tcolors.BOLD}(Y/N)\n> {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.OKGREEN}\nClient added with success{Tcolors.ENDC}
|
||||||
|
|
||||||
|
{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}
|
||||||
|
{Tcolors.ENDC}{Tcolors.BOLD}Number > {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.OKGREEN}Number added with success{Tcolors.ENDC}
|
||||||
|
|
||||||
|
{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.OKGREEN}\nChosen number: {Tcolors.UNDERLINE}{data}{Tcolors.ENDC}
|
||||||
|
|
||||||
|
{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}""".encode("utf-8")
|
||||||
|
# end of testing the STOP option
|
||||||
|
|
||||||
|
|
||||||
|
def test_guess_without_starting():
|
||||||
|
# start of testing the GUESS option without starting the game
|
||||||
|
server = Popen("python3 server.py 8000", stdout=PIPE, shell=True, close_fds=True)
|
||||||
|
client_test = Popen("python3 client.py test 8000", stdout=PIPE, stdin=PIPE, shell=True)
|
||||||
|
port = client_test.args.split(" ")[3]
|
||||||
|
hostname = socket.gethostbyname(socket.gethostname())
|
||||||
|
|
||||||
|
output = client_test.communicate(input=b"guess")[0]
|
||||||
|
|
||||||
|
client_test.terminate()
|
||||||
|
server.terminate()
|
||||||
|
|
||||||
|
assert output == f"""{Tcolors.WARNING}Connecting to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.WARNING}...{Tcolors.ENDC}
|
||||||
|
{Tcolors.OKGREEN}Connected to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.OKGREEN} as client {Tcolors.UNDERLINE}{client_test.args.split(" ")[2]}\n{Tcolors.ENDC}
|
||||||
|
{Tcolors.OKCYAN}{Tcolors.BOLD}{Tcolors.UNDERLINE}Number characteristics guesser game!{Tcolors.ENDC}
|
||||||
|
|
||||||
|
{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.WARNING}You must start the game first\n{Tcolors.ENDC}
|
||||||
|
{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}""".encode("utf-8")
|
||||||
|
# end of testing the GUESS option without starting the game
|
||||||
|
|
||||||
|
|
||||||
|
def test_number_without_starting():
|
||||||
|
# start of testing the NUMBER option without starting the game
|
||||||
|
server = Popen("python3 server.py 9000", stdout=PIPE, shell=True, close_fds=True)
|
||||||
|
client_test = Popen("python3 client.py test 9000", stdout=PIPE, stdin=PIPE, shell=True)
|
||||||
|
port = client_test.args.split(" ")[3]
|
||||||
|
hostname = socket.gethostbyname(socket.gethostname())
|
||||||
|
|
||||||
|
output = client_test.communicate(input=b"number")[0]
|
||||||
|
|
||||||
|
client_test.terminate()
|
||||||
|
server.terminate()
|
||||||
|
|
||||||
|
assert output == f"""{Tcolors.WARNING}Connecting to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.WARNING}...{Tcolors.ENDC}
|
||||||
|
{Tcolors.OKGREEN}Connected to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.OKGREEN} as client {Tcolors.UNDERLINE}{client_test.args.split(" ")[2]}\n{Tcolors.ENDC}
|
||||||
|
{Tcolors.OKCYAN}{Tcolors.BOLD}{Tcolors.UNDERLINE}Number characteristics guesser game!{Tcolors.ENDC}
|
||||||
|
|
||||||
|
{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.WARNING}You must start the game first\n{Tcolors.ENDC}
|
||||||
|
{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}""".encode("utf-8")
|
||||||
|
# end of testing the NUMBER option without starting the game
|
||||||
|
|
||||||
|
|
||||||
|
def test_stop_without_starting():
|
||||||
|
# start of testing the STOP option without starting the game
|
||||||
|
server = Popen("python3 server.py 10000", stdout=PIPE, shell=True, close_fds=True)
|
||||||
|
client_test = Popen("python3 client.py test 10000", stdout=PIPE, stdin=PIPE, shell=True)
|
||||||
|
port = client_test.args.split(" ")[3]
|
||||||
|
hostname = socket.gethostbyname(socket.gethostname())
|
||||||
|
|
||||||
|
output = client_test.communicate(input=b"stop")[0]
|
||||||
|
|
||||||
|
client_test.terminate()
|
||||||
|
server.terminate()
|
||||||
|
|
||||||
|
assert output == f"""{Tcolors.WARNING}Connecting to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.WARNING}...{Tcolors.ENDC}
|
||||||
|
{Tcolors.OKGREEN}Connected to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.OKGREEN} as client {Tcolors.UNDERLINE}{client_test.args.split(" ")[2]}\n{Tcolors.ENDC}
|
||||||
|
{Tcolors.OKCYAN}{Tcolors.BOLD}{Tcolors.UNDERLINE}Number characteristics guesser game!{Tcolors.ENDC}
|
||||||
|
|
||||||
|
{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.WARNING}You must start the game first\n{Tcolors.ENDC}
|
||||||
|
{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}""".encode("utf-8")
|
||||||
|
# end of testing the STOP option without starting the game
|
||||||
|
|
||||||
|
|
||||||
|
def test_stop_without_numbers():
|
||||||
|
# start of testing the STOP option after stopping the game
|
||||||
|
server = Popen("python3 server.py 11000", stdout=PIPE, shell=True, close_fds=True)
|
||||||
|
client_test = Popen("python3 client.py test 11000", stdout=PIPE, stdin=PIPE, shell=True)
|
||||||
|
port = client_test.args.split(" ")[3]
|
||||||
|
hostname = socket.gethostbyname(socket.gethostname())
|
||||||
|
|
||||||
|
output = client_test.communicate(input=b"start\ny\nstop")[0]
|
||||||
|
|
||||||
|
client_test.terminate()
|
||||||
|
server.terminate()
|
||||||
|
|
||||||
|
assert output == f"""{Tcolors.WARNING}Connecting to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.WARNING}...{Tcolors.ENDC}
|
||||||
|
{Tcolors.OKGREEN}Connected to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.OKGREEN} as client {Tcolors.UNDERLINE}{client_test.args.split(" ")[2]}\n{Tcolors.ENDC}
|
||||||
|
{Tcolors.OKCYAN}{Tcolors.BOLD}{Tcolors.UNDERLINE}Number characteristics guesser game!{Tcolors.ENDC}
|
||||||
|
|
||||||
|
{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}
|
||||||
|
{Tcolors.ENDC}Do you wish to use a cipher? {Tcolors.BOLD}(Y/N)\n> {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.OKGREEN}\nClient added with success{Tcolors.ENDC}
|
||||||
|
|
||||||
|
{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.FAIL}Error: Client has not yet sent any number{Tcolors.ENDC}
|
||||||
|
{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}""".encode("utf-8")
|
||||||
|
# end of testing the STOP option after stopping the game
|
||||||
|
|
||||||
|
|
||||||
|
def test_stop_after_stopping():
|
||||||
|
# start of testing the STOP option after stopping the game
|
||||||
|
server = Popen("python3 server.py 12000", stdout=PIPE, shell=True, close_fds=True)
|
||||||
|
client_test = Popen("python3 client.py test 12000", stdout=PIPE, stdin=PIPE, shell=True)
|
||||||
|
port = client_test.args.split(" ")[3]
|
||||||
|
hostname = socket.gethostbyname(socket.gethostname())
|
||||||
|
|
||||||
|
output = client_test.communicate(input=b"start\ny\nnumber\n123\nstop\nstop")[0]
|
||||||
|
|
||||||
|
client_test.terminate()
|
||||||
|
server.terminate()
|
||||||
|
|
||||||
|
data = 123
|
||||||
|
|
||||||
|
assert output == f"""{Tcolors.WARNING}Connecting to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.WARNING}...{Tcolors.ENDC}
|
||||||
|
{Tcolors.OKGREEN}Connected to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.OKGREEN} as client {Tcolors.UNDERLINE}{client_test.args.split(" ")[2]}\n{Tcolors.ENDC}
|
||||||
|
{Tcolors.OKCYAN}{Tcolors.BOLD}{Tcolors.UNDERLINE}Number characteristics guesser game!{Tcolors.ENDC}
|
||||||
|
|
||||||
|
{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}
|
||||||
|
{Tcolors.ENDC}Do you wish to use a cipher? {Tcolors.BOLD}(Y/N)\n> {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.OKGREEN}\nClient added with success{Tcolors.ENDC}
|
||||||
|
|
||||||
|
{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}
|
||||||
|
{Tcolors.ENDC}{Tcolors.BOLD}Number > {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.OKGREEN}Number added with success{Tcolors.ENDC}
|
||||||
|
|
||||||
|
{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.OKGREEN}\nChosen number: {Tcolors.UNDERLINE}{data}{Tcolors.ENDC}
|
||||||
|
|
||||||
|
{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.WARNING}You can't stop the game again\n{Tcolors.ENDC}
|
||||||
|
{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}""".encode("utf-8")
|
||||||
|
|
@ -0,0 +1,39 @@
|
||||||
|
import socket
|
||||||
|
from subprocess import Popen
|
||||||
|
from subprocess import PIPE
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
|
||||||
|
def test_lack_args():
|
||||||
|
# no args
|
||||||
|
proc = Popen("python3 server.py", stdout=PIPE, shell=True, close_fds=True)
|
||||||
|
output = proc.communicate()[0]
|
||||||
|
|
||||||
|
assert output == "Usage: python3 server.py <port>\n".encode("utf-8")
|
||||||
|
|
||||||
|
|
||||||
|
def test_too_many_args():
|
||||||
|
# too many args
|
||||||
|
proc = Popen("python3 server.py 1234 1234", stdout=PIPE, shell=True, close_fds=True)
|
||||||
|
output = proc.communicate()[0]
|
||||||
|
|
||||||
|
assert output == "Usage: python3 server.py <port>\n".encode("utf-8")
|
||||||
|
|
||||||
|
|
||||||
|
def test_str_port():
|
||||||
|
proc = Popen("python3 server.py abc", stdout=PIPE, shell=True, close_fds=True)
|
||||||
|
output = proc.communicate()[0]
|
||||||
|
|
||||||
|
proc.terminate()
|
||||||
|
assert output == "Port must be an integer\n".encode("utf-8")
|
||||||
|
|
||||||
|
|
||||||
|
def test_port_in_use():
|
||||||
|
Popen("python3 server.py 50000", stdout=PIPE, shell=True, close_fds=True)
|
||||||
|
proc2 = Popen("python3 server.py 50000", stdout=PIPE, shell=True, close_fds=True)
|
||||||
|
output = proc2.communicate()[0]
|
||||||
|
|
||||||
|
proc2.terminate()
|
||||||
|
hostname = socket.gethostbyname(socket.gethostname())
|
||||||
|
port = 50000
|
||||||
|
assert output == f"Failed to create server socket, maybe the port is already in use?\n\n".encode("utf-8")
|
||||||
|
|
@ -0,0 +1,50 @@
|
||||||
|
import random
|
||||||
|
import socket
|
||||||
|
import subprocess
|
||||||
|
import sys
|
||||||
|
from subprocess import Popen
|
||||||
|
from subprocess import PIPE
|
||||||
|
import pytest
|
||||||
|
import client
|
||||||
|
import server
|
||||||
|
|
||||||
|
|
||||||
|
def test_client_verify_port():
|
||||||
|
assert client.verify_port("0") == {'status': False, 'error': 'Port number must be between 1024 and 65535'}
|
||||||
|
assert client.verify_port("1023") == {'status': False, 'error': 'Port number must be between 1024 and 65535'}
|
||||||
|
assert client.verify_port("1024") == {'status': True, "port": 1024}
|
||||||
|
assert client.verify_port("65535") == {'status': True, "port": 65535}
|
||||||
|
assert client.verify_port("65536") == {'status': False, 'error': 'Port number must be between 1024 and 65535'}
|
||||||
|
assert client.verify_port("100000") == {'status': False, 'error': 'Port number must be between 1024 and 65535'}
|
||||||
|
assert client.verify_port("example") == {'status': False, 'error': 'Port must be an integer'}
|
||||||
|
assert client.verify_port("test") == {'status': False, 'error': 'Port must be an integer'}
|
||||||
|
|
||||||
|
|
||||||
|
def test_verify_hostname():
|
||||||
|
assert client.verify_hostname("localhost") == {'status': True}
|
||||||
|
assert client.verify_hostname("123.123.123.123") == {'status': True}
|
||||||
|
assert client.verify_hostname("example.com") == {'status': False, 'error': 'Invalid DNS address'}
|
||||||
|
assert client.verify_hostname("test") == {'status': False, 'error': 'Invalid DNS address'}
|
||||||
|
assert client.verify_hostname("123.123.123") == {'status': False, 'error': 'Invalid DNS address'}
|
||||||
|
assert client.verify_hostname("413.123.123.123") == {'status': False, 'error': 'Invalid DNS address'}
|
||||||
|
assert client.verify_hostname("123.413.123.123") == {'status': False, 'error': 'Invalid DNS address'}
|
||||||
|
assert client.verify_hostname("123.123.413.123") == {'status': False, 'error': 'Invalid DNS address'}
|
||||||
|
assert client.verify_hostname("123.123.123.413") == {'status': False, 'error': 'Invalid DNS address'}
|
||||||
|
assert client.verify_hostname("-1.-1.-1.-1") == {'status': False, 'error': 'Invalid DNS address'}
|
||||||
|
assert client.verify_hostname("256.256.256.256") == {'status': False, 'error': 'Invalid DNS address'}
|
||||||
|
assert client.verify_hostname("0.0.0.0") == {'status': True}
|
||||||
|
assert client.verify_hostname("255.255.255.255") == {'status': True}
|
||||||
|
|
||||||
|
|
||||||
|
def test_server_verify_port():
|
||||||
|
assert server.verify_port("0") == {'status': False, 'error': 'Port number must be between 1024 and 65535'}
|
||||||
|
assert server.verify_port("1023") == {'status': False, 'error': 'Port number must be between 1024 and 65535'}
|
||||||
|
assert server.verify_port("1024") == {'status': True, "port": 1024}
|
||||||
|
assert server.verify_port("65535") == {'status': True, "port": 65535}
|
||||||
|
assert server.verify_port("65536") == {'status': False, 'error': 'Port number must be between 1024 and 65535'}
|
||||||
|
assert server.verify_port("100000") == {'status': False, 'error': 'Port number must be between 1024 and 65535'}
|
||||||
|
assert server.verify_port("example") == {'status': False, 'error': 'Port must be an integer'}
|
||||||
|
assert server.verify_port("test") == {'status': False, 'error': 'Port must be an integer'}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -0,0 +1,2 @@
|
||||||
|
Nome: Rúben Gomes, NMec: 113435, Email: rlcg@ua.pt, Trabalho: 50%
|
||||||
|
Nome: Tiago Garcia, NMec: 114184, Email: tiago.rgarcia@ua.pt, Trabalho: 50%
|
||||||
Loading…
Reference in New Issue