Updated tests.

magnetrwn · Jun 16, 2023 · f89f357 · f89f357
1 parent 886c245
commit f89f357
Show file tree

Hide file tree

Showing 7 changed files with 51 additions and 48 deletions.
diff --git a/Makefile b/Makefile
@@ -13,3 +13,7 @@ build: clean test
 .PHONY: install
 install: clean test build
 	pip3 install --force-reinstall dist/*.whl
+
+.PHONY: fast-install
+fast-install: clean build
+	pip3 install --force-reinstall dist/*.whl
diff --git a/NetHang/run.py b/NetHang/run.py
@@ -2,24 +2,19 @@
 
 
 import sys
-from multiprocessing import Process
 
 from NetHang.server import HangmanServer
 
 
 SETTINGS = {
     # Maximum number of concurrent users
     # "max_conn": 20,
-
     # Available ports to use, will only attach to one
     # "avail_ports": [29111, 29112, 29113],
-
     # Number of new user handlers, for concurrent user connection
     # "new_conn_processes": 10
-
     # Multiply all delays (mainly between queries from same user) by this value
     # "delay_factor": 1
-
     # Allow two or more users from the same source IP address
     "allow_same_source_ip": False
 }
@@ -43,19 +38,18 @@ def cli_run():
 
 def temp_run_start(port):
     """Run temporary server on localhost"""
-    local_server = HangmanServer("localhost", settings=
-    {
-        "avail_ports": [port],
-        "allow_same_source_ip": True,
-        "delay_factor": 0
-    })
-    temp_process = Process(target=local_server.run)
-    temp_process.start()
-    return temp_process
-
-
-def temp_run_stop(temp_process):
+    server = HangmanServer(
+        "localhost",
+        settings={
+            "avail_ports": [port],
+            "allow_same_source_ip": True,
+            "delay_factor": 0,
+        },
+    )
+    server.run()
+    return server
+
+
+def temp_run_stop(server):
     """Stop temporary server on localhost"""
-    if temp_process.is_alive():
-        temp_process.kill()
-        temp_process.join()
+    server.stop()
diff --git a/NetHang/server.py b/NetHang/server.py
@@ -61,7 +61,6 @@ def __init__(
             + " total clients."
         )
 
-
     def accept_clients_worker(
         self, server_socket, players_read_queue, players_write_queue
     ):
@@ -70,7 +69,7 @@ def accept_clients_worker(
         while True:
             try:
                 client_raw = server_socket.accept()
-                sleep(0.5*self.delay_factor)
+                sleep(0.5 * self.delay_factor)
                 client_raw[0].send(GRAPHICS["title"].encode("utf-8"))
                 worker_players = players_read_queue.get()
                 if not self.allow_same_source_ip:
@@ -84,7 +83,7 @@ def accept_clients_worker(
                     client_raw[0].settimeout(1)
                     try:
                         while True:
-                            sleep(0.25*self.delay_factor)
+                            sleep(0.25 * self.delay_factor)
                             dirty = client_raw[0].recv(512)
                             if not dirty:
                                 break
@@ -104,7 +103,7 @@ def accept_clients_worker(
                             continue
                     except UnicodeDecodeError:
                         client_raw[0].send(
-                            "Please only use UTF-8 characters!\n".encode("utf-8")
+                            "Please only use valid UTF-8 characters!\n".encode("utf-8")
                         )
                         continue
                     if (
@@ -135,11 +134,10 @@ def accept_clients_worker(
                 )
                 print("\x1B[36m" + client_nickname + " joined\x1B[0m")
             except BrokenPipeError:
-                sleep(0.5*self.delay_factor)
+                sleep(0.5 * self.delay_factor)
             except KeyboardInterrupt:
                 return
 
-
     def run_worker(self, running):
         """Run the hangman server."""
         players = PlayerList()
@@ -150,11 +148,7 @@ def run_worker(self, running):
         for _ in range(self.new_conn_processes):
             Process(
                 target=self.accept_clients_worker,
-                args=(
-                    self.server_socket,
-                    players_read_queue,
-                    players_write_queue
-                ),
+                args=(self.server_socket, players_read_queue, players_write_queue),
                 daemon=True,
             ).start()
 
@@ -200,14 +194,12 @@ def run_worker(self, running):
         self.server_socket.close()
         print("\r\x1B[91mServer stopped.\x1B[0m")
 
-
     def run(self):
         """Run the hangman server."""
         self.running.value = 1
         self.server_process = Process(target=self.run_worker, args=(self.running,))
         self.server_process.start()
 
-
     def stop(self, *args):
         """Stop the hangman server."""
         self.running.value = 0

diff --git a/README.md b/README.md
@@ -2,10 +2,9 @@
 [![Code style: black](https://img.shields.io/badge/code%20style-black-000000.svg)](https://github.com/psf/black)
 [![CodeFactor](https://www.codefactor.io/repository/github/magnetrwn/NetHang/badge)](https://www.codefactor.io/repository/github/magnetrwn/NetHang)
 
-Compact, client-server Python application to play hangman using netcat (nc) or any other TCP/UDP client, written for linux systems.
-
-Get the wheel (.whl) from `dist` or run `make install` yourself!
-
-... add asciinema demo here ...
+Compact, client-server Python application to play hangman using netcat (nc) or any other TCP client.
+Get the wheel (.whl) from `dist` or run `make install` yourself. When installed, just run `NetHang` or `nethang` from shell to start the server.
 
 **Currently work in progress...**
+
+... add asciinema demo here ...
diff --git a/dist/NetHang-0.1.10-py3-none-any.whl b/dist/NetHang-0.1.10-py3-none-any.whl
diff --git a/dist/NetHang-0.1.10.tar.gz b/dist/NetHang-0.1.10.tar.gz
diff --git a/tests/test_user_input.py b/tests/test_user_input.py
@@ -4,28 +4,42 @@
 from NetHang.run import temp_run_start, temp_run_stop
 
 
-UTF8_DATA = ''.join(tuple(chr(i) for i in range(32, 0x110000) if chr(i).isprintable()))
+UTF8_DATA = "".join(tuple(chr(i) for i in range(32, 0x110000) if chr(i).isprintable()))
+
 
 def generate_utf8_string(length):
+    """Generate random strings using all UTF-8 characters available"""
     string = ""
     for _ in range(length):
         string += random.choices(UTF8_DATA)[0]
     return string
 
 
-def test_user_joining():
+def test_bad_username_attempt():
+    """Tests logging in with too long, too short or bad encoding usernames"""
+    # TODO: this test does not make much sense, because the server accepts
+    #       most unicode chars for users and there are no charset limits yet.
+    #       Length checking is also problematic because of the buffer size,
+    #       making consecutive requests of progressively shorter usernames.
     port = random.randint(49152, 65535)
     temp_run = temp_run_start(port)
     username_length = 1
-    while username_length <= 4:
-        for _ in range(1):
-            conn = so.create_connection(("localhost", port))
-            conn.settimeout(2)
-            conn.recv(256)
+    conn = so.create_connection(("localhost", port))
+    conn.settimeout(2)
+    conn.recv(256)
+    two_line_response = False
+    for username_length in [1, 33]:
+        for _ in range(6):
             conn.send(generate_utf8_string(username_length).encode("utf-8"))
-            if b"Nickname" not in conn.recv(256):
+            if two_line_response:
+                conn.recv(256)
+            a=conn.recv(256)
+            print(username_length, _, a)
+            if b"Nickname" not in a:
+                # User was able to login with a bad username
                 assert False
-            conn.close()
-        username_length += username_length
+            two_line_response = True
+
+    conn.close()
     temp_run_stop(temp_run)
     assert True