keploy · pathakharshit · Feb 9, 2026
diff --git a/http-schema/app-test.py 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
@@ -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
@@ -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()