From 47507bd4849e5b490c852093e0069a6124c4c345 Mon Sep 17 00:00:00 2001 From: TiagoRG Date: Mon, 22 May 2023 20:29:58 +0100 Subject: [PATCH] Revert "[LABI] Added directory with mid-semester project" This reverts commit 6db8c56290c5b69868a3bf9d9bea724613c615a0. Making it's own repository --- 1ano/2semestre/labi/labi2023-ap-g6/README.md | 11 - .../labi2023-ap-g6/client_server/client.py | 394 ------------------ .../client_server/common_comm.py | 82 ---- .../labi2023-ap-g6/client_server/server.py | 384 ----------------- .../client_server/test_client.py | 344 --------------- .../client_server/test_server.py | 39 -- .../client_server/test_units.py | 50 --- 1ano/2semestre/labi/labi2023-ap-g6/readme.txt | 2 - 8 files changed, 1306 deletions(-) delete mode 100644 1ano/2semestre/labi/labi2023-ap-g6/README.md delete mode 100644 1ano/2semestre/labi/labi2023-ap-g6/client_server/client.py delete mode 100644 1ano/2semestre/labi/labi2023-ap-g6/client_server/common_comm.py delete mode 100644 1ano/2semestre/labi/labi2023-ap-g6/client_server/server.py delete mode 100644 1ano/2semestre/labi/labi2023-ap-g6/client_server/test_client.py delete mode 100644 1ano/2semestre/labi/labi2023-ap-g6/client_server/test_server.py delete mode 100644 1ano/2semestre/labi/labi2023-ap-g6/client_server/test_units.py delete mode 100644 1ano/2semestre/labi/labi2023-ap-g6/readme.txt diff --git a/1ano/2semestre/labi/labi2023-ap-g6/README.md b/1ano/2semestre/labi/labi2023-ap-g6/README.md deleted file mode 100644 index efb8c23..0000000 --- a/1ano/2semestre/labi/labi2023-ap-g6/README.md +++ /dev/null @@ -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 diff --git a/1ano/2semestre/labi/labi2023-ap-g6/client_server/client.py b/1ano/2semestre/labi/labi2023-ap-g6/client_server/client.py deleted file mode 100644 index f50c2f4..0000000 --- a/1ano/2semestre/labi/labi2023-ap-g6/client_server/client.py +++ /dev/null @@ -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() diff --git a/1ano/2semestre/labi/labi2023-ap-g6/client_server/common_comm.py b/1ano/2semestre/labi/labi2023-ap-g6/client_server/common_comm.py deleted file mode 100644 index 8862cf5..0000000 --- a/1ano/2semestre/labi/labi2023-ap-g6/client_server/common_comm.py +++ /dev/null @@ -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 diff --git a/1ano/2semestre/labi/labi2023-ap-g6/client_server/server.py b/1ano/2semestre/labi/labi2023-ap-g6/client_server/server.py deleted file mode 100644 index 39f0b82..0000000 --- a/1ano/2semestre/labi/labi2023-ap-g6/client_server/server.py +++ /dev/null @@ -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 " % 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() diff --git a/1ano/2semestre/labi/labi2023-ap-g6/client_server/test_client.py b/1ano/2semestre/labi/labi2023-ap-g6/client_server/test_client.py deleted file mode 100644 index d585307..0000000 --- a/1ano/2semestre/labi/labi2023-ap-g6/client_server/test_client.py +++ /dev/null @@ -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") diff --git a/1ano/2semestre/labi/labi2023-ap-g6/client_server/test_server.py b/1ano/2semestre/labi/labi2023-ap-g6/client_server/test_server.py deleted file mode 100644 index 48e7fde..0000000 --- a/1ano/2semestre/labi/labi2023-ap-g6/client_server/test_server.py +++ /dev/null @@ -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 \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 \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") \ No newline at end of file diff --git a/1ano/2semestre/labi/labi2023-ap-g6/client_server/test_units.py b/1ano/2semestre/labi/labi2023-ap-g6/client_server/test_units.py deleted file mode 100644 index 685b70b..0000000 --- a/1ano/2semestre/labi/labi2023-ap-g6/client_server/test_units.py +++ /dev/null @@ -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'} - - - diff --git a/1ano/2semestre/labi/labi2023-ap-g6/readme.txt b/1ano/2semestre/labi/labi2023-ap-g6/readme.txt deleted file mode 100644 index f6db3db..0000000 --- a/1ano/2semestre/labi/labi2023-ap-g6/readme.txt +++ /dev/null @@ -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%