Revert "[LABI] Added directory with mid-semester project"
This reverts commit 6db8c56290.
Making it's own repository
This commit is contained in:
parent
7b212ca3e2
commit
47507bd484
|
|
@ -1,11 +0,0 @@
|
|||
# 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
|
||||
|
|
@ -1,394 +0,0 @@
|
|||
#!/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()
|
||||
|
|
@ -1,82 +0,0 @@
|
|||
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
|
||||
|
|
@ -1,384 +0,0 @@
|
|||
#!/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()
|
||||
|
|
@ -1,344 +0,0 @@
|
|||
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")
|
||||
|
|
@ -1,39 +0,0 @@
|
|||
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")
|
||||
|
|
@ -1,50 +0,0 @@
|
|||
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'}
|
||||
|
||||
|
||||
|
||||
|
|
@ -1,2 +0,0 @@
|
|||
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