From c266e9a43cea4a2608ead54af9de0159222cb6c0 Mon Sep 17 00:00:00 2001 From: Harshit Pathak Date: Tue, 10 Feb 2026 13:09:58 +0000 Subject: [PATCH 1/3] feat: Keploy schema-match sample app Signed-off-by: Harshit Pathak --- python-schema-match/app-test.py | 154 +++++++++++++++++++++++++ python-schema-match/app.py | 146 +++++++++++++++++++++++ python-schema-match/check-endpoints.py | 50 ++++++++ 3 files changed, 350 insertions(+) create mode 100644 python-schema-match/app-test.py create mode 100644 python-schema-match/app.py create mode 100644 python-schema-match/check-endpoints.py diff --git a/python-schema-match/app-test.py b/python-schema-match/app-test.py new file mode 100644 index 0000000..7442ebe --- /dev/null +++ b/python-schema-match/app-test.py @@ -0,0 +1,154 @@ +#!/usr/bin/env python3 +""" +TEST VERSION of the HTTP server with MODIFIED responses. +Use this for TESTING mode. + +EXPECTED RESULTS (Total 10): +PASS: 7 +FAIL: 3 (Missing Field, Type Mismatch, Hierarchy Mismatch) +""" +import socket +import json +import random +import string +import time + +PORT = 5000 + +RESPONSES = { + # 1. PASS: Different values, same types + '/user/profile': { + "id": 999, + "username": "different_user", + "active": False, + "profile": { + "age": 99, + "city": "New York", + "preferences": {"theme": "light", "notifications": False} + }, + "roles": ["viewer"] + }, + + # 2. PASS: Array length diff (Standard schema matching allows this) + '/user/history': { + "user_id": 999, + "login_history": [ + {"ip": "1.1.1.1", "timestamp": 9999999999} + ] + }, + + # 3. PASS: Extra fields (Superset is allowed) + '/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", + "metadata": {"source": "api", "cache": True} + }, + + # 4. FAIL: FIELD DELETION (Missing 'feature_flags') + '/admin/config': { + "maintenance_mode": False, + # "feature_flags": MISSING! -> Should Fail + "deprecated_since": None, + "retry_limit": 3, + "added_config": "extra" + }, + + # 5. FAIL: TYPE MISMATCH (Int -> String) + '/data/matrix': { + "matrix": [["1", "0", "0"], ["0", "1", "0"], ["0", "0", "1"]], + "dimension": "3x3" + }, + + # 6. FAIL: HIERARCHY/TUPLE MISMATCH + # Original: [int, string, bool...] + # Here: [string, int, bool...] (Swapped first two) + '/data/mixed_array': { + "mixed": ["string", 1, True, {"obj": "val"}, None, [1, 2]] + }, + + # 7. PASS: Exact match + '/edge/empty_response': {}, + + # 8. PASS: Exact match + '/edge/null_root': None, + + # 9. PASS: Exact match + '/edge/special_chars': { + "text": "Hello Hello", + "emoji": "🚀 🔥 🐛", + "symbols": "!@#$%^&*()_+-=[]{}|;':\",./<>?", + "path": "C:\\Program Files\\Keploy" + }, + + # 10. PASS: Exact match + '/edge/nested_null': { + "data": {"value": None} + } +} + +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 + path = parts[1] + + print(f"Test Request: {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\n\r\n" + 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\nContent-Type: application/json\r\nContent-Length: {len(body)}\r\n\r\n{body}" + client_socket.sendall(response.encode('utf-8')) + + 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}...") + + while True: + try: + client_socket, addr = server_socket.accept() + handle_request(client_socket) + except KeyboardInterrupt: + break + server_socket.close() + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/python-schema-match/app.py b/python-schema-match/app.py new file mode 100644 index 0000000..b160d94 --- /dev/null +++ b/python-schema-match/app.py @@ -0,0 +1,146 @@ +#!/usr/bin/env python3 +""" +Simple socket-based HTTP server for Keploy compatibility. +ORIGINAL STATE - Use this for RECORDING. +""" +import socket +import json +import random +import string +import time +import sys + +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} + ] + }, + '/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" + }, + # Added 10th endpoint for clean 7/3 split + '/edge/nested_null': { + "data": {"value": None} + } +} + +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\nContent-Type: application/json\r\nContent-Length: {len(body)}\r\nConnection: close\r\n\r\n{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) + + try: + server_socket.bind(('0.0.0.0', PORT)) + except OSError: + print("Port in use, waiting...") + time.sleep(2) + server_socket.bind(('0.0.0.0', PORT)) + + server_socket.listen(10) + print(f"Starting ORIGINAL Server on port {PORT}...") + + while True: + try: + client_socket, addr = server_socket.accept() + handle_request(client_socket) + except KeyboardInterrupt: + break + except Exception as e: + print(f"Accept error: {e}") + + server_socket.close() + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/python-schema-match/check-endpoints.py b/python-schema-match/check-endpoints.py new file mode 100644 index 0000000..d64180b --- /dev/null +++ b/python-schema-match/check-endpoints.py @@ -0,0 +1,50 @@ +import urllib.request +import time +import sys + +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/special_chars', + '/edge/nested_null' +] + +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: + # Short timeout to fail fast + with urllib.request.urlopen(url, timeout=2) as response: + status = response.status + body = response.read().decode('utf-8') + + if status == 200: + print(f"✅ {ep} [200 OK]") + 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__": + # Small delay to ensure server is ready + time.sleep(2) + check_endpoints() \ No newline at end of file From 34f963f2e9b80a541d3d5bc84e9e057ccb0b7faa Mon Sep 17 00:00:00 2001 From: Harshit Pathak Date: Tue, 10 Feb 2026 13:17:39 +0000 Subject: [PATCH 2/3] feat: added read me file Signed-off-by: Harshit Pathak --- python-schema-match/README.md | 269 ++++++++++++++++++++++++++++++++++ 1 file changed, 269 insertions(+) create mode 100644 python-schema-match/README.md diff --git a/python-schema-match/README.md b/python-schema-match/README.md new file mode 100644 index 0000000..8fa1497 --- /dev/null +++ b/python-schema-match/README.md @@ -0,0 +1,269 @@ +# Python Schema Match + +A sample HTTP server application designed to test and validate JSON schema matching capabilities with Keploy. This application serves multiple diverse endpoints with various response schemas to comprehensively test schema structure validation and compatibility. + +## Overview + +This application is a simple socket-based HTTP server written in Python that provides 10 endpoints covering different schema patterns including: + +- **Nested Objects**: User profile with complex nested data structures +- **Arrays**: Different array formats and mixed-type arrays +- **Edge Cases**: Null values, empty responses, special characters, and large payloads +- **Data Types**: Strings, numbers, booleans, objects, and null values + +The application supports schema validation testing to ensure API responses maintain consistent structure across different versions or implementations. + +### API Endpoints + +1. `GET /user/profile` - Returns user profile with nested preferences +2. `GET /user/history` - Returns user login history with array of objects +3. `GET /product/search` - Returns product search results with items array +4. `GET /admin/config` - Returns admin configuration with feature flags +5. `GET /data/matrix` - Returns a matrix data structure +6. `GET /data/mixed_array` - Returns an array with mixed data types +7. `GET /edge/empty_response` - Returns an empty JSON object +8. `GET /edge/null_root` - Returns a null value at root +9. `GET /edge/special_chars` - Returns special characters and escape sequences +10. `GET /edge/nested_null` - Returns an object with null values + +## Installation & Setup + +### Prerequisites + +- Python 3.7 or higher +- Keploy CLI installed on your system +- Linux or WSL (for Windows users) + +### Install Keploy + +```bash +curl -O -L https://keploy.io/install.sh && source install.sh +``` + +### Clone the Repository + +```bash +git clone https://github.com/keploy/samples-python.git && cd samples-python/python-schema-match +``` + +## Running the Application + +### Start the Server (Original Version - for Recording) + +This server returns standard responses and is used for recording test cases: + +```bash +python3 app.py +``` + +The server will start on `http://localhost:5000` + +### Start the Test Server (Modified Version - for Testing) + +This server returns modified responses with schema variations to test compatibility: + +```bash +python3 app-test.py +``` + +## Keploy Integration + +### Recording Test Cases + +Keploy will capture API calls and generate test cases with mocked responses. + +**1. Start Recording:** + +```bash +keploy record -c "python3 app.py" +``` + +The server will start and be ready to capture API interactions. + +**2. Generate Test Cases by Making API Calls:** + +Use curl, Postman, or Hoppscotch to make requests to the endpoints. Here are some examples: + +```bash +# Check endpoint 1: User Profile +curl http://localhost:5000/user/profile + +# Check endpoint 2: User History +curl http://localhost:5000/user/history + +# Check endpoint 3: Product Search +curl http://localhost:5000/product/search + +# Check endpoint 4: Admin Config +curl http://localhost:5000/admin/config + +# Check endpoint 5: Data Matrix +curl http://localhost:5000/data/matrix + +# Check endpoint 6: Mixed Array +curl http://localhost:5000/data/mixed_array + +# Check endpoint 7: Empty Response +curl http://localhost:5000/edge/empty_response + +# Check endpoint 8: Null Root +curl http://localhost:5000/edge/null_root + +# Check endpoint 9: Special Characters +curl http://localhost:5000/edge/special_chars + +# Check endpoint 10: Nested Null +curl http://localhost:5000/edge/nested_null +``` + +Or use the provided endpoint checker script: + +```bash +python3 check-endpoints.py +``` + +This script will automatically test all 10 endpoints and provide a summary. + +**3. Keploy Test Artifacts** + +After making API calls, Keploy will generate test cases in the `keploy/` directory: + +- **Test Files**: `keploy/tests/test-*.yml` - Contains HTTP request/response pairs +- **Mock Files**: `keploy/mocks/mocks.yml` - Contains any external service interactions + +Each test file captures: +- **Request**: HTTP method, headers, URL, and body +- **Response**: Status code, headers, and response body +- **Assertions**: Validation rules including noise fields (like timestamps) + +Example test output: + +```yaml +version: api.keploy.io/v1beta2 +kind: Http +name: test-1 +spec: + metadata: {} + req: + method: GET + url: http://localhost:5000/user/profile + header: + Accept: "*/*" + Host: localhost:5000 + User-Agent: curl/7.68.0 + resp: + status_code: 200 + header: + Content-Type: application/json + Content-Length: "245" + body: | + { + "id": 101, + "username": "keploy_user", + "active": true, + "profile": { + "age": 25, + "city": "San Francisco", + "preferences": {"theme": "dark", "notifications": true} + }, + "roles": ["admin", "editor"] + } + assertions: + noise: + - header.Date + created: 1694000000 +``` + +### Running Test Cases + +Test your application against the captured test cases using the test server: + +```bash +keploy test -c "python3 app-test.py" --delay 5 +``` + +The `--delay` flag gives your application time to start before tests begin (in seconds). + +**Expected Results:** + +- **Total Endpoints**: 10 +- **Expected PASS**: 7 (Same schema structure, different values) +- **Expected FAIL**: 3 (Schema modifications testing) + - Missing fields + - Type mismatches + - Hierarchy mismatches + +The application validates that: +1. Extra fields are allowed (superset schema) +2. Array length variations are acceptable +3. Different values with same types pass +4. Type changes are detected +5. Missing required fields are detected +6. Nested structure changes are caught + +### Output and Coverage + +After running tests, Keploy will display: +- Test summary (passed/failed count) +- Coverage metrics +- Detailed failure reports with diff information + +## Project Structure + +``` +python-schema-match/ +├── app.py # Original HTTP server (for recording) +├── app-test.py # Modified HTTP server (for testing) +├── check-endpoints.py # Utility script to test all endpoints +├── README.md # This file +└── keploy/ # Keploy artifacts (generated) + ├── tests/ + │ ├── test-1.yml + │ ├── test-2.yml + │ └── ... + └── mocks/ + └── mocks.yml +``` + +## Use Cases + +This sample application demonstrates: + +1. **Schema Validation**: Ensure API responses maintain consistent JSON structure +2. **API Contract Testing**: Validate that responses match expected schemas +3. **Backward Compatibility**: Test if new versions break schema contracts +4. **Data Type Verification**: Ensure fields have correct data types +5. **Response Structure Integrity**: Validate nested objects and arrays + +## Troubleshooting + +### Port Already in Use +If port 5000 is already in use, modify the `PORT` variable in `app.py` or `app-test.py`. + +### Connection Refused +Ensure the server is running before making API calls or running tests. + +### Test Failures +Expected failures are intentional to demonstrate schema mismatch detection. Review the diff information in Keploy output to understand the schema differences. + +## Additional Resources + +- [Keploy Documentation](https://docs.keploy.io) +- [JSON Schema Guide](https://json-schema.org/) +- [Python HTTP Servers](https://docs.python.org/3/library/http.server.html) + +## Contributing + +Feel free to extend this sample by: +- Adding more diverse schema patterns +- Testing additional data types +- Creating more complex nested structures +- Adding custom validation rules + +## License + +MIT License - See LICENSE.md for details + +--- + +**Happy Testing! 🚀** From 62bfa40a5846ab33fb79f3746f50d3f9e70d4ded Mon Sep 17 00:00:00 2001 From: Harshit Pathak Date: Wed, 11 Feb 2026 09:07:57 +0000 Subject: [PATCH 3/3] fix(sample-app): add connection header to response in python-schema-match --- python-schema-match/app-test.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/python-schema-match/app-test.py b/python-schema-match/app-test.py index 7442ebe..e1c0ce9 100644 --- a/python-schema-match/app-test.py +++ b/python-schema-match/app-test.py @@ -127,7 +127,7 @@ def handle_request(client_socket): else: body = json.dumps(body_data) - response = f"HTTP/1.0 200 OK\r\nContent-Type: application/json\r\nContent-Length: {len(body)}\r\n\r\n{body}" + response = f"HTTP/1.0 200 OK\r\nContent-Type: application/json\r\nContent-Length: {len(body)}\r\nConnection: close\r\n\r\n{body}" client_socket.sendall(response.encode('utf-8')) except Exception as e: