Created
January 30, 2026 23:48
-
-
Save f5-rahm/59e63ffeb98b5e777363abfe03f99428 to your computer and use it in GitHub Desktop.
Test tool for passing dummy json payloads between a client and server
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/env python3 | |
| """ | |
| Simple HTTP Client/Server for JSON payload exchange | |
| Usage: | |
| Server mode: python cspayload.py server --host 0.0.0.0 --port 8088 | |
| Client mode: python cspayload.py client --host localhost --port 8088 | |
| """ | |
| import argparse | |
| import json | |
| import sys | |
| from http.server import HTTPServer, BaseHTTPRequestHandler | |
| from urllib.request import Request, urlopen | |
| from urllib.error import URLError | |
| # Default JSON payloads | |
| DEFAULT_CLIENT_PAYLOAD = { | |
| "my_string": "Hello World", | |
| "my_number": 42, | |
| "my_boolean": True, | |
| "my_null": None, | |
| "my_array": [1, 2, 3], | |
| "my_object": { | |
| "nested_string": "I'm nested", | |
| "nested_array": ["a", "b", "c"] | |
| } | |
| } | |
| DEFAULT_SERVER_PAYLOAD = { | |
| "message": "Hello from server", | |
| "type": "response", | |
| "status": "success", | |
| "data": { | |
| "processed": True, | |
| "timestamp": "2026-01-29" | |
| } | |
| } | |
| class JSONRequestHandler(BaseHTTPRequestHandler): | |
| """HTTP request handler for JSON payloads""" | |
| custom_response = None | |
| def do_POST(self): | |
| """Handle POST requests with JSON payload""" | |
| content_length = int(self.headers.get('Content-Length', 0)) | |
| if content_length > 0: | |
| body = self.rfile.read(content_length) | |
| try: | |
| received_data = json.loads(body.decode('utf-8')) | |
| print("\n[Server] Received JSON payload:") | |
| print(json.dumps(received_data, indent=2)) | |
| except json.JSONDecodeError as e: | |
| print(f"\n[Server] Error decoding JSON: {e}") | |
| self.send_error(400, "Invalid JSON") | |
| return | |
| else: | |
| print("\n[Server] Received empty POST request") | |
| # Send response | |
| response_payload = self.custom_response if self.custom_response else DEFAULT_SERVER_PAYLOAD | |
| response_json = json.dumps(response_payload, indent=2) | |
| self.send_response(200) | |
| self.send_header('Content-Type', 'application/json') | |
| self.send_header('Content-Length', str(len(response_json))) | |
| self.end_headers() | |
| self.wfile.write(response_json.encode('utf-8')) | |
| print("\n[Server] Sent JSON response:") | |
| print(response_json) | |
| def do_GET(self): | |
| """Handle GET requests""" | |
| print(f"\n[Server] Received GET request for path: {self.path}") | |
| response_payload = self.custom_response if self.custom_response else DEFAULT_SERVER_PAYLOAD | |
| response_json = json.dumps(response_payload, indent=2) | |
| self.send_response(200) | |
| self.send_header('Content-Type', 'application/json') | |
| self.send_header('Content-Length', str(len(response_json))) | |
| self.end_headers() | |
| self.wfile.write(response_json.encode('utf-8')) | |
| print("[Server] Sent JSON response to GET request:") | |
| print(response_json) | |
| def log_message(self, format, *args): | |
| """Suppress default logging""" | |
| pass | |
| def run_server(host, port, custom_payload=None): | |
| """Run HTTP server""" | |
| if custom_payload: | |
| JSONRequestHandler.custom_response = custom_payload | |
| server_address = (host, port) | |
| httpd = HTTPServer(server_address, JSONRequestHandler) | |
| print(f"[Server] Starting HTTP server on {host}:{port}") | |
| print(f"[Server] Press Ctrl+C to stop") | |
| try: | |
| httpd.serve_forever() | |
| except KeyboardInterrupt: | |
| print("\n[Server] Shutting down...") | |
| httpd.shutdown() | |
| def run_client(host, port, custom_payload=None, method='POST'): | |
| """Run HTTP client""" | |
| payload = custom_payload if custom_payload else DEFAULT_CLIENT_PAYLOAD | |
| url = f"http://{host}:{port}/" | |
| print(f"[Client] Connecting to {url}") | |
| print(f"[Client] Sending JSON payload ({method}):") | |
| print(json.dumps(payload, indent=2)) | |
| try: | |
| if method == 'POST': | |
| json_data = json.dumps(payload).encode('utf-8') | |
| request = Request(url, data=json_data, headers={'Content-Type': 'application/json'}) | |
| else: # GET | |
| request = Request(url) | |
| with urlopen(request, timeout=10) as response: | |
| response_data = response.read().decode('utf-8') | |
| print(f"\n[Client] Received response (Status: {response.status}):") | |
| try: | |
| response_json = json.loads(response_data) | |
| print(json.dumps(response_json, indent=2)) | |
| except json.JSONDecodeError: | |
| print(response_data) | |
| except URLError as e: | |
| print(f"\n[Client] Error: {e}") | |
| sys.exit(1) | |
| except Exception as e: | |
| print(f"\n[Client] Unexpected error: {e}") | |
| sys.exit(1) | |
| def main(): | |
| parser = argparse.ArgumentParser(description='Simple HTTP Client/Server for JSON payloads') | |
| subparsers = parser.add_subparsers(dest='mode', help='Mode: client or server') | |
| # Server arguments | |
| server_parser = subparsers.add_parser('server', help='Run as HTTP server') | |
| server_parser.add_argument('--host', default='0.0.0.0', help='Host to bind to (default: 0.0.0.0)') | |
| server_parser.add_argument('--port', type=int, default=8088, help='Port to listen on (default: 8088)') | |
| server_parser.add_argument('--payload', help='Custom JSON payload file for responses') | |
| # Client arguments | |
| client_parser = subparsers.add_parser('client', help='Run as HTTP client') | |
| client_parser.add_argument('--host', default='localhost', help='Server host (default: localhost)') | |
| client_parser.add_argument('--port', type=int, default=8088, help='Server port (default: 8088)') | |
| client_parser.add_argument('--payload', help='Custom JSON payload file to send') | |
| client_parser.add_argument('--method', choices=['POST', 'GET'], default='POST', | |
| help='HTTP method (default: POST)') | |
| args = parser.parse_args() | |
| if not args.mode: | |
| parser.print_help() | |
| sys.exit(1) | |
| # Load custom payload if specified | |
| custom_payload = None | |
| if hasattr(args, 'payload') and args.payload: | |
| try: | |
| with open(args.payload, 'r') as f: | |
| custom_payload = json.load(f) | |
| except Exception as e: | |
| print(f"Error loading custom payload: {e}") | |
| sys.exit(1) | |
| # Run in appropriate mode | |
| if args.mode == 'server': | |
| run_server(args.host, args.port, custom_payload) | |
| elif args.mode == 'client': | |
| run_client(args.host, args.port, custom_payload, args.method) | |
| if __name__ == '__main__': | |
| main() | |
| (.venv) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment