From 229d63a63f41d0f3caa367e32f21fd4d2a2f344a Mon Sep 17 00:00:00 2001 From: Harshit Pathak Date: Mon, 9 Feb 2026 17:37:27 +0000 Subject: [PATCH] feat: Add http-schema sample for schema-match testing --- http-schema/app-test.py | 170 +++++++++++++++++++++++++++++++++ http-schema/app.py | 148 ++++++++++++++++++++++++++++ http-schema/check_endpoints.py | 60 ++++++++++++ http-schema/keploy.yml | 87 +++++++++++++++++ 4 files changed, 465 insertions(+) create mode 100644 http-schema/app-test.py create mode 100644 http-schema/app.py create mode 100644 http-schema/check_endpoints.py create mode 100755 http-schema/keploy.yml diff --git a/http-schema/app-test.py b/http-schema/app-test.py new file mode 100644 index 0000000..9610222 --- /dev/null +++ b/http-schema/app-test.py @@ -0,0 +1,170 @@ +#!/usr/bin/env python3 +""" +TEST VERSION of the HTTP server with MODIFIED responses. +Use this with: sudo ../../keploy-bin test -c "python3 app-test.py" --schema-match --delay 5 + +MODIFICATIONS FOR SCHEMA-MATCH TESTING: +- Tests 1-2: Different VALUES but same TYPES → Should PASS +- Tests 3-4: EXTRA FIELDS (superset) → Should PASS +- Tests 5-6: TYPE MISMATCHES → Should FAIL +- Tests 7-10: Same as original (control tests) → Should PASS +""" +import socket +import json +import random +import string + +PORT = 5000 + +RESPONSES = { + # TEST 1: Different values, same types → PASS + '/user/profile': { + "id": 999, # Changed from 101 + "username": "different_user", # Changed + "active": False, # Changed from True + "profile": { + "age": 99, # Changed from 25 + "city": "New York", # Changed + "preferences": {"theme": "light", "notifications": False} # Changed + }, + "roles": ["viewer"] # Changed from ["admin", "editor"] + }, + + # TEST 2: Different values, same types → PASS + '/user/history': { + "user_id": 999, # Changed + "login_history": [ + {"ip": "1.1.1.1", "timestamp": 9999999999} # Different values, fewer items + ] + }, + + # TEST 3: EXTRA FIELDS (superset) → PASS + '/product/search': { + "query": "laptop", + "total_results": 1500, + "page": 1, + "items": [ + {"id": "p1", "name": "Laptop Pro", "price": 1299.99, "stock": 50}, + {"id": "p2", "name": "Laptop Air", "price": 999.99, "stock": 0} + ], + "extra_field": "This field was not in original", # EXTRA + "metadata": {"source": "api", "cache": True} # EXTRA nested + }, + + # TEST 4: EXTRA FIELDS in nested object → PASS + '/admin/config': { + "maintenance_mode": False, + "feature_flags": { + "beta_access": True, + "legacy_support": False, + "new_feature": True # EXTRA field in nested object + }, + "deprecated_since": None, + "retry_limit": 3, + "added_config": "extra" # EXTRA field at root + }, + + # TEST 5: TYPE MISMATCH (int → string) → FAIL + '/data/matrix': { + "matrix": [["1", "0", "0"], ["0", "1", "0"], ["0", "0", "1"]], # strings instead of ints! + "dimension": "3x3" + }, + + # TEST 6: TYPE MISMATCH (mixed types changed) → FAIL + '/data/mixed_array': { + "mixed": ["1", 2, "true", {"obj": "val"}, "null", [1, 2]] # int->str, bool->str, null->str + }, + + # TEST 7-10: Same as original (control tests) → PASS + '/edge/empty_response': {}, + '/edge/null_root': None, + '/edge/special_chars': { + "text": "Hello Hello", + "emoji": "🚀 🔥 🐛", + "symbols": "!@#$%^&*()_+-=[]{}|;':\",./<>?", + "path": "C:\\Program Files\\Keploy" + } +} + +def get_large_payload(): + return { + "payload_size": "5KB", + "content": "".join(random.choices(string.ascii_letters, k=5000)) + } + +def handle_request(client_socket): + try: + client_socket.settimeout(5.0) + request = client_socket.recv(4096).decode('utf-8', errors='ignore') + + if not request: + return + + lines = request.split('\r\n') + if not lines: + return + + request_line = lines[0] + parts = request_line.split(' ') + if len(parts) < 2: + return + + method = parts[0] + path = parts[1] + + print(f"Request: {method} {path}") + + if path == '/edge/large_payload': + body_data = get_large_payload() + elif path in RESPONSES: + body_data = RESPONSES[path] + else: + response = "HTTP/1.0 404 Not Found\r\nConnection: close\r\n\r\nNot Found" + client_socket.sendall(response.encode('utf-8')) + return + + if body_data is None: + body = "null" + else: + body = json.dumps(body_data) + + response = f"HTTP/1.0 200 OK\r\n" + response += f"Content-Type: application/json\r\n" + response += f"Content-Length: {len(body)}\r\n" + response += f"Connection: close\r\n" + response += f"\r\n" + response += body + + client_socket.sendall(response.encode('utf-8')) + + except socket.timeout: + pass + except Exception as e: + print(f"Error: {e}") + finally: + client_socket.close() + +def main(): + server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + server_socket.bind(('0.0.0.0', PORT)) + server_socket.listen(10) + + print(f"Starting TEST Server on port {PORT}...") + print("⚠️ This server has MODIFIED responses for schema-match testing!") + print("Expected: 8 PASS (tests 1-4, 7-10), 2 FAIL (tests 5-6)") + + while True: + try: + client_socket, addr = server_socket.accept() + handle_request(client_socket) + except KeyboardInterrupt: + print("\nShutting down...") + break + except Exception as e: + print(f"Accept error: {e}") + + server_socket.close() + +if __name__ == "__main__": + main() diff --git a/http-schema/app.py b/http-schema/app.py new file mode 100644 index 0000000..860297f --- /dev/null +++ b/http-schema/app.py @@ -0,0 +1,148 @@ +#!/usr/bin/env python3 +""" +Simple socket-based HTTP server for Keploy compatibility. +Uses raw sockets without Python's http.server module. +""" +import socket +import json +import random +import string + +PORT = 5000 + +# Pre-compute responses for each endpoint +RESPONSES = { + '/user/profile': { + "id": 101, + "username": "keploy_user", + "active": True, + "profile": { + "age": 25, + "city": "San Francisco", + "preferences": {"theme": "dark", "notifications": True} + }, + "roles": ["admin", "editor"] + }, + '/user/history': { + "user_id": 101, + "login_history": [ + {"ip": "192.168.1.1", "timestamp": 1700000001}, + {"ip": "10.0.0.1", "timestamp": 1700000050}, + {"ip": "172.16.0.5", "timestamp": 1700000100} + ] + }, + '/product/search': { + "query": "laptop", + "total_results": 1500, + "page": 1, + "items": [ + {"id": "p1", "name": "Laptop Pro", "price": 1299.99, "stock": 50}, + {"id": "p2", "name": "Laptop Air", "price": 999.99, "stock": 0} + ] + }, + '/admin/config': { + "maintenance_mode": False, + "feature_flags": {"beta_access": True, "legacy_support": False}, + "deprecated_since": None, + "retry_limit": 3 + }, + '/data/matrix': { + "matrix": [[1, 0, 0], [0, 1, 0], [0, 0, 1]], + "dimension": "3x3" + }, + '/data/mixed_array': { + "mixed": [1, "string", True, {"obj": "val"}, None, [1, 2]] + }, + '/edge/empty_response': {}, + '/edge/null_root': None, + '/edge/special_chars': { + "text": "Hello Hello", + "emoji": "🚀 🔥 🐛", + "symbols": "!@#$%^&*()_+-=[]{}|;':\",./<>?", + "path": "C:\\Program Files\\Keploy" + } +} + +def get_large_payload(): + return { + "payload_size": "5KB", + "content": "".join(random.choices(string.ascii_letters, k=5000)) + } + +def handle_request(client_socket): + try: + client_socket.settimeout(5.0) + request = client_socket.recv(4096).decode('utf-8', errors='ignore') + + if not request: + return + + lines = request.split('\r\n') + if not lines: + return + + request_line = lines[0] + parts = request_line.split(' ') + if len(parts) < 2: + return + + method = parts[0] + path = parts[1] + + print(f"Request: {method} {path}") + + if path == '/edge/large_payload': + body_data = get_large_payload() + elif path in RESPONSES: + body_data = RESPONSES[path] + else: + response = "HTTP/1.0 404 Not Found\r\nConnection: close\r\n\r\nNot Found" + client_socket.sendall(response.encode('utf-8')) + return + + if body_data is None: + body = "null" + else: + body = json.dumps(body_data) + + response = f"HTTP/1.0 200 OK\r\n" + response += f"Content-Type: application/json\r\n" + response += f"Content-Length: {len(body)}\r\n" + response += f"Connection: close\r\n" + response += f"\r\n" + response += body + + client_socket.sendall(response.encode('utf-8')) + + except socket.timeout: + pass + except Exception as e: + print(f"Error: {e}") + finally: + client_socket.close() + +def main(): + server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + server_socket.bind(('0.0.0.0', PORT)) + server_socket.listen(10) + + print(f"Starting Simple Socket Server on port {PORT}...") + print("Endpoints: /user/profile, /user/history, /product/search, /admin/config,") + print(" /data/matrix, /data/mixed_array, /edge/empty_response,") + print(" /edge/null_root, /edge/large_payload, /edge/special_chars") + + while True: + try: + client_socket, addr = server_socket.accept() + handle_request(client_socket) + except KeyboardInterrupt: + print("\nShutting down...") + break + except Exception as e: + print(f"Accept error: {e}") + + server_socket.close() + +if __name__ == "__main__": + main() diff --git a/http-schema/check_endpoints.py b/http-schema/check_endpoints.py new file mode 100644 index 0000000..3f67ade --- /dev/null +++ b/http-schema/check_endpoints.py @@ -0,0 +1,60 @@ + +import urllib.request +import time +import json +import threading +import sys +import os + +# Wait for server to start (manual or automated) +# For this script, we assume the server is running on localhost:5000 + +BASE_URL = "http://localhost:5000" +ENDPOINTS = [ + '/user/profile', + '/user/history', + '/product/search', + '/admin/config', + '/data/matrix', + '/data/mixed_array', + '/edge/empty_response', + '/edge/null_root', + '/edge/large_payload', + '/edge/special_chars' +] + +def check_endpoints(): + print(f"Checking {len(ENDPOINTS)} endpoints on {BASE_URL}...") + success_count = 0 + fail_count = 0 + + for ep in ENDPOINTS: + url = BASE_URL + ep + try: + with urllib.request.urlopen(url) as response: + status = response.status + body = response.read().decode('utf-8') + + # Basic validation + if status == 200: + print(f"✅ {ep} [200 OK]") + # Peek at body if short + if len(body) < 100: + print(f" Body: {body}") + else: + print(f" Body: {body[:100]}... (truncated)") + success_count += 1 + else: + print(f"❌ {ep} [{status}]") + fail_count += 1 + except Exception as e: + print(f"❌ {ep} Error: {e}") + fail_count += 1 + + print(f"\n--- Summary ---") + print(f"Total: {len(ENDPOINTS)}") + print(f"Passed: {success_count}") + print(f"Failed: {fail_count}") + +if __name__ == "__main__": + check_endpoints() diff --git a/http-schema/keploy.yml b/http-schema/keploy.yml new file mode 100755 index 0000000..a0d86ca --- /dev/null +++ b/http-schema/keploy.yml @@ -0,0 +1,87 @@ +# Generated by Keploy (3-dev) +path: "" +appName: http-schema +appId: 0 +command: python3 app.py +templatize: + testSets: [] +port: 0 +e2e: false +dnsPort: 26789 +proxyPort: 16789 +incomingProxyPort: 36789 +debug: false +disableTele: false +disableANSI: false +containerName: "" +networkName: "" +buildDelay: 30 +test: + selectedTests: {} + globalNoise: + global: {} + test-sets: {} + delay: 5 + host: "" + port: 0 + grpcPort: 0 + apiTimeout: 5 + skipCoverage: false + coverageReportPath: "" + ignoreOrdering: true + mongoPassword: default@123 + language: "" + removeUnusedMocks: false + fallBackOnMiss: false + jacocoAgentPath: "" + basePath: "" + mocking: true + ignoredTests: {} + disableLineCoverage: false + disableMockUpload: true + useLocalMock: false + updateTemplate: false + mustPass: false + maxFailAttempts: 5 + maxFlakyChecks: 1 + protoFile: "" + protoDir: "" + protoInclude: [] + compareAll: false + schemaMatch: false +record: + filters: [] + basePath: "" + recordTimer: 0s + metadata: "" + sync: false + globalPassthrough: false +report: + selectedTestSets: {} + showFullBody: false + reportPath: "" + summary: false + testCaseIDs: [] +disableMapping: false +configPath: "" +bypassRules: [] +generateGithubActions: false +keployContainer: keploy-v3 +keployNetwork: keploy-network +cmdType: native +contract: + services: [] + tests: [] + path: "" + download: false + generate: false + driven: consumer + mappings: + servicesMapping: {} + self: s1 +inCi: false +serverPort: 0 +mockDownload: + registryIds: [] + +# Visit [https://keploy.io/docs/running-keploy/configuration-file/] to learn about using keploy through configration file.