import json import logging import os import requests import jsonschema from jsonschema import validate from typing import Dict, Any, List # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') class APIContractTester: """ A class to test API endpoints against their contracts. """ def __init__(self, api_catalog_path: str): """ Initializes the APIContractTester with the path to the API catalog. Args: api_catalog_path (str): The path to the API catalog file (JSON). """ self.api_catalog_path = api_catalog_path self.api_catalog: Dict[str, Any] = self._load_api_catalog() def _load_api_catalog(self) -> Dict[str, Any]: """ Loads the API catalog from the specified JSON file. Returns: Dict[str, Any]: The API catalog as a dictionary. Raises: FileNotFoundError: If the API catalog file does not exist. json.JSONDecodeError: If the API catalog file is not valid JSON. """ try: with open(self.api_catalog_path, 'r') as f: return json.load(f) except FileNotFoundError as e: logging.error(f"API catalog file not found: {e}") raise except json.JSONDecodeError as e: logging.error(f"Error decoding API catalog JSON: {e}") raise def validate_contract(self, endpoint_name: str) -> Dict[str, List[str]]: """ Validates a specific API endpoint against its contract. Args: endpoint_name (str): The name of the endpoint to validate. Returns: Dict[str, List[str]]: A dictionary containing validation results, with keys like 'request_format', 'response_format', 'status_codes', etc. Each value is a list of error messages (empty list if no errors). """ endpoint = self.api_catalog.get(endpoint_name) if not endpoint: logging.error(f"Endpoint '{endpoint_name}' not found in API catalog.") return {"error": [f"Endpoint '{endpoint_name}' not found in API catalog."]} base_url = endpoint.get("base_url") path = endpoint.get("path") method = endpoint.get("method", "GET").upper() # Default to GET request_schema = endpoint.get("request_schema") response_schema = endpoint.get("response_schema") expected_status_codes = endpoint.get("expected_status_codes", [200]) # Default to 200 if not all([base_url, path, request_schema, response_schema, expected_status_codes]): logging.error(f"Missing required fields for endpoint '{endpoint_name}'.") return {"error": ["Missing required fields in API catalog."]} url = f"{base_url}{path}" validation_results: Dict[str, List[str]] = { "request_format": [], "response_format": [], "status_codes": [] } try: # Prepare request data (if any) based on request_schema if request_schema and method in ["POST", "PUT", "PATCH"]: # Generate dummy data based on schema request_data = self._generate_dummy_data(request_schema) if not isinstance(request_data, dict): validation_results["request_format"].append("Failed to create request data. Request data must be a dict.") return validation_results response = requests.request(method, url, json=request_data) else: response = requests.request(method, url) # Validate status code if response.status_code not in expected_status_codes: validation_results["status_codes"].append( f"Expected status code(s) {expected_status_codes}, but got {response.status_code}." ) # Validate response format try: response_json = response.json() validate(instance=response_json, schema=response_schema) except json.JSONDecodeError: validation_results["response_format"].append("Response is not valid JSON.") except jsonschema.exceptions.ValidationError as e: validation_results["response_format"].append(f"Response validation error: {e.message}") except requests.exceptions.RequestException as e: logging.error(f"Request failed for endpoint '{endpoint_name}': {e}") validation_results["error"] = [f"Request failed: {str(e)}"] except Exception as e: logging.exception(f"An unexpected error occurred while validating endpoint '{endpoint_name}'.") validation_results["error"] = [f"Unexpected error: {str(e)}"] return validation_results def _generate_dummy_data(self, schema: Dict[str, Any]) -> Dict[str, Any]: """ Generates dummy data based on the provided JSON schema. This is a simplified implementation and can be expanded to handle more complex schema types. Args: schema (Dict[str, Any]): The JSON schema. Returns: Dict[str, Any]: The generated dummy data. Returns an empty dict if the schema is invalid. """ if not isinstance(schema, dict): logging.error("Invalid schema: Schema must be a dictionary.") return {} # Or raise an exception if more appropriate data: Dict[str, Any] = {} properties = schema.get("properties", {}) for property_name, property_schema in properties.items(): property_type = property_schema.get("type") if property_type == "string": data[property_name] = "string_value" elif property_type == "integer": data[property_name] = 123 elif property_type == "number": data[property_name] = 123.45 elif property_type == "boolean": data[property_name] = True elif property_type == "array": items_schema = property_schema.get("items") if items_schema and isinstance(items_schema, dict): data[property_name] = [self._generate_dummy_data(items_schema)] else: data[property_name] = [] elif property_type == "object": data[property_name] = self._generate_dummy_data(property_schema) else: data[property_name] = None # Unknown type return data def generate_report(self, validation_results: Dict[str, Dict[str, List[str]]]) -> str: """ Generates a contract compliance report based on the validation results. Args: validation_results (Dict[str, Dict[str, List[str]]]): A dictionary containing validation results for each endpoint. Returns: str: The contract compliance report as a string. """ report = "API Contract Compliance Report:\n\n" for endpoint_name, results in validation_results.items(): report += f"Endpoint: {endpoint_name}\n" has_errors = False for category, errors in results.items(): if errors: has_errors = True report += f" {category.capitalize()}:\n" for error in errors: report += f" - {error}\n" if not has_errors: report += " No errors found.\n" report += "\n" return report def main(): """ Main function to run the API contract tester. """ api_catalog_path = "/mnt/e/genesis-system/api_catalog.json" # Replace with your actual path # Create a dummy api_catalog.json file for testing purposes dummy_api_catalog = { "example_endpoint": { "base_url": "https://httpbin.org", "path": "/get", "method": "GET", "request_schema": {}, "response_schema": { "type": "object", "properties": { "url": {"type": "string"} }, "required": ["url"] }, "expected_status_codes": [200] }, "post_endpoint": { "base_url": "https://httpbin.org", "path": "/post", "method": "POST", "request_schema": { "type": "object", "properties": { "key1": {"type": "string"}, "key2": {"type": "integer"} }, "required": ["key1"] }, "response_schema": { "type": "object", "properties": { "json": {"type": "object"} }, "required": ["json"] }, "expected_status_codes": [200] } } # Ensure the directory exists os.makedirs(os.path.dirname(api_catalog_path), exist_ok=True) try: with open(api_catalog_path, 'w') as f: json.dump(dummy_api_catalog, f, indent=4) # Create a dummy catalog except Exception as e: logging.error(f"Failed to create dummy API catalog: {e}") return try: tester = APIContractTester(api_catalog_path) validation_results: Dict[str, Dict[str, List[str]]] = {} for endpoint_name in tester.api_catalog.keys(): validation_results[endpoint_name] = tester.validate_contract(endpoint_name) report = tester.generate_report(validation_results) print(report) except Exception as e: logging.exception("An error occurred during the testing process.") if __name__ == "__main__": main()