mirror of
https://github.com/fosrl/newt.git
synced 2026-05-16 13:19:53 +00:00
Add some test scripts for ws and move to testing/
This commit is contained in:
Owen
49
testing/udp_client.py
Normal file
49
testing/udp_client.py
Normal file
@@ -0,0 +1,49 @@
|
||||
import socket
|
||||
import sys
|
||||
|
||||
# Argument parsing: Check if IP and Port are provided
|
||||
if len(sys.argv) != 3:
|
||||
print("Usage: python udp_client.py <HOST_IP> <HOST_PORT>")
|
||||
# Example: python udp_client.py 127.0.0.1 12000
|
||||
sys.exit(1)
|
||||
|
||||
HOST = sys.argv[1]
|
||||
try:
|
||||
PORT = int(sys.argv[2])
|
||||
except ValueError:
|
||||
print("Error: HOST_PORT must be an integer.")
|
||||
sys.exit(1)
|
||||
|
||||
# The message to send to the server
|
||||
MESSAGE = "Hello UDP Server! How are you?"
|
||||
|
||||
# Create a UDP socket
|
||||
try:
|
||||
client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
except socket.error as err:
|
||||
print(f"Failed to create socket: {err}")
|
||||
sys.exit()
|
||||
|
||||
try:
|
||||
print(f"Sending message to {HOST}:{PORT}...")
|
||||
|
||||
# Send the message (data must be encoded to bytes)
|
||||
client_socket.sendto(MESSAGE.encode('utf-8'), (HOST, PORT))
|
||||
|
||||
# Wait for the server's response (buffer size 1024 bytes)
|
||||
data, server_address = client_socket.recvfrom(1024)
|
||||
|
||||
# Decode and print the server's response
|
||||
response = data.decode('utf-8')
|
||||
print("-" * 30)
|
||||
print(f"Received response from server {server_address[0]}:{server_address[1]}:")
|
||||
print(f"-> Data: '{response}'")
|
||||
|
||||
except socket.error as err:
|
||||
print(f"Error during communication: {err}")
|
||||
|
||||
finally:
|
||||
# Close the socket
|
||||
client_socket.close()
|
||||
print("-" * 30)
|
||||
print("Client finished and socket closed.")
|
||||
58
testing/udp_server.py
Normal file
58
testing/udp_server.py
Normal file
@@ -0,0 +1,58 @@
|
||||
import socket
|
||||
import sys
|
||||
|
||||
# optionally take in some positional args for the port
|
||||
if len(sys.argv) > 1:
|
||||
try:
|
||||
PORT = int(sys.argv[1])
|
||||
except ValueError:
|
||||
print("Invalid port number. Using default port 12000.")
|
||||
PORT = 12000
|
||||
else:
|
||||
PORT = 12000
|
||||
|
||||
# Define the server host and port
|
||||
HOST = '0.0.0.0' # Standard loopback interface address (localhost)
|
||||
|
||||
# Create a UDP socket
|
||||
try:
|
||||
server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
except socket.error as err:
|
||||
print(f"Failed to create socket: {err}")
|
||||
sys.exit()
|
||||
|
||||
# Bind the socket to the address
|
||||
try:
|
||||
server_socket.bind((HOST, PORT))
|
||||
print(f"UDP Server listening on {HOST}:{PORT}")
|
||||
except socket.error as err:
|
||||
print(f"Bind failed: {err}")
|
||||
server_socket.close()
|
||||
sys.exit()
|
||||
|
||||
# Wait for and process incoming data
|
||||
while True:
|
||||
try:
|
||||
# Receive data and the client's address (buffer size 1024 bytes)
|
||||
data, client_address = server_socket.recvfrom(1024)
|
||||
|
||||
# Decode the data and print the message
|
||||
message = data.decode('utf-8')
|
||||
print("-" * 30)
|
||||
print(f"Received message from {client_address[0]}:{client_address[1]}:")
|
||||
print(f"-> Data: '{message}'")
|
||||
|
||||
# Prepare the response message
|
||||
response_message = f"Hello client! Server received: '{message.upper()}'"
|
||||
|
||||
# Send the response back to the client
|
||||
server_socket.sendto(response_message.encode('utf-8'), client_address)
|
||||
print(f"Sent response back to client.")
|
||||
|
||||
except Exception as e:
|
||||
print(f"An error occurred: {e}")
|
||||
break
|
||||
|
||||
# Clean up (though usually unreachable in an infinite server loop)
|
||||
server_socket.close()
|
||||
print("Server stopped.")
|
||||
60
testing/ws_client.py
Normal file
60
testing/ws_client.py
Normal file
@@ -0,0 +1,60 @@
|
||||
import asyncio
|
||||
import sys
|
||||
import websockets
|
||||
|
||||
# Argument parsing: Check if HOST and PORT are provided
|
||||
if len(sys.argv) < 3 or len(sys.argv) > 4:
|
||||
print("Usage: python ws_client.py <HOST_IP> <HOST_PORT> [ws|wss]")
|
||||
# Example: python ws_client.py 127.0.0.1 8765
|
||||
# Example: python ws_client.py 127.0.0.1 8765 wss
|
||||
sys.exit(1)
|
||||
|
||||
HOST = sys.argv[1]
|
||||
try:
|
||||
PORT = int(sys.argv[2])
|
||||
except ValueError:
|
||||
print("Error: HOST_PORT must be an integer.")
|
||||
sys.exit(1)
|
||||
|
||||
if len(sys.argv) == 4:
|
||||
SCHEME = sys.argv[3].lower()
|
||||
if SCHEME not in ("ws", "wss"):
|
||||
print("Error: scheme must be 'ws' or 'wss'.")
|
||||
sys.exit(1)
|
||||
else:
|
||||
SCHEME = "ws"
|
||||
|
||||
URI = f"{SCHEME}://{HOST}:{PORT}"
|
||||
|
||||
# The message to send to the server
|
||||
MESSAGE = "Hello WebSocket Server! How are you?"
|
||||
|
||||
|
||||
async def main():
|
||||
print(f"Connecting to {URI}...")
|
||||
|
||||
try:
|
||||
async with websockets.connect(URI) as websocket:
|
||||
print(f"Connected to server.")
|
||||
print(f"Sending message: '{MESSAGE}'")
|
||||
|
||||
await websocket.send(MESSAGE)
|
||||
|
||||
response = await websocket.recv()
|
||||
|
||||
print("-" * 30)
|
||||
print(f"Received response from server:")
|
||||
print(f"-> Data: '{response}'")
|
||||
|
||||
except ConnectionRefusedError:
|
||||
print(f"Error: Connection to {URI} was refused. Is the server running?")
|
||||
except websockets.exceptions.InvalidMessage as e:
|
||||
print(f"Error: Server did not respond with a valid WebSocket handshake: {e}")
|
||||
except Exception as e:
|
||||
print(f"Error during communication: {e}")
|
||||
|
||||
print("-" * 30)
|
||||
print("Client finished.")
|
||||
|
||||
|
||||
asyncio.run(main())
|
||||
49
testing/ws_server.py
Normal file
49
testing/ws_server.py
Normal file
@@ -0,0 +1,49 @@
|
||||
import asyncio
|
||||
import sys
|
||||
import websockets
|
||||
|
||||
# Optionally take in a positional arg for the port
|
||||
if len(sys.argv) > 1:
|
||||
try:
|
||||
PORT = int(sys.argv[1])
|
||||
except ValueError:
|
||||
print("Invalid port number. Using default port 8765.")
|
||||
PORT = 8765
|
||||
else:
|
||||
PORT = 8765
|
||||
|
||||
# Define the server host
|
||||
HOST = "0.0.0.0"
|
||||
|
||||
|
||||
async def handle_client(websocket):
|
||||
client_address = websocket.remote_address
|
||||
print(f"Client connected: {client_address[0]}:{client_address[1]}")
|
||||
|
||||
try:
|
||||
async for message in websocket:
|
||||
print("-" * 30)
|
||||
print(f"Received message from {client_address[0]}:{client_address[1]}:")
|
||||
print(f"-> Data: '{message}'")
|
||||
|
||||
response = f"Hello client! Server received: '{message.upper()}'"
|
||||
|
||||
await websocket.send(response)
|
||||
print(f"Sent response back to client.")
|
||||
|
||||
except websockets.exceptions.ConnectionClosedOK:
|
||||
print(f"Client {client_address[0]}:{client_address[1]} disconnected cleanly.")
|
||||
except websockets.exceptions.ConnectionClosedError as e:
|
||||
print(f"Client {client_address[0]}:{client_address[1]} disconnected with error: {e}")
|
||||
|
||||
|
||||
async def main():
|
||||
print(f"WebSocket Server listening on {HOST}:{PORT}")
|
||||
async with websockets.serve(handle_client, HOST, PORT):
|
||||
await asyncio.Future() # Run forever
|
||||
|
||||
|
||||
try:
|
||||
asyncio.run(main())
|
||||
except KeyboardInterrupt:
|
||||
print("\nServer stopped.")
|
||||
Reference in New Issue
Block a user