{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/mcp-servers\/mcp-contracts\/tools\/contract_tools\/contract_validator.py",
"content": "\"\"\"Contract Validation Module\"\"\"\nfrom typing import Optional, Any\nfrom datetime import datetime\n\nfrom domain.contracts import ContractValidation\nfrom validators.contract_validator import run_contract_validation\n\nfrom .contract_parser import ContractParser\nfrom .constants import VALIDATION_PASSED, VALIDATION_FAILED\n\n\nclass ContractValidatorService:\n \"\"\"Service for validating contracts against their rules\"\"\"\n\n def __init__(self, repository):\n \"\"\"\n Initialize validator service.\n\n Args:\n repository: ContractRepository instance\n \"\"\"\n self.repository = repository\n self.parser = ContractParser()\n\n def validate_contract(\n self,\n contract_id: int,\n yaml_content: str,\n path: Optional[str] = None,\n triggered_by: str = \"mcp\",\n start_time: float = 0.0\n ) -> tuple[bool, dict[str, Any], Optional[int]]:\n \"\"\"\n Validate a contract and save the result.\n\n Args:\n contract_id: ID of the contract being validated\n yaml_content: YAML content of the contract\n path: Optional specific path to validate\n triggered_by: Who\/what triggered the validation\n start_time: Start time for duration calculation\n\n Returns:\n Tuple of (success, result_dict, validation_id)\n \"\"\"\n # Parse YAML\n try:\n contract_data = self.parser.parse_yaml(yaml_content)\n except Exception as e:\n return False, {\"error\": f\"Invalid contract YAML: {e}\"}, None\n\n # Run validation\n try:\n result = run_contract_validation(contract_data, path)\n except Exception as e:\n return False, {\"error\": f\"Validation failed: {e}\"}, None\n\n # Calculate duration\n duration_ms = 0\n if start_time > 0:\n import time\n duration_ms = int((time.time() - start_time) * 1000)\n\n # Save validation result\n validation = ContractValidation(\n contract_id=contract_id,\n validated_at=datetime.now(),\n result=VALIDATION_PASSED if result.outcome == \"passed\" else VALIDATION_FAILED,\n critical_count=result.critical,\n major_count=result.major,\n minor_count=result.minor,\n violations=result.findings,\n triggered_by=triggered_by,\n target_path=path,\n duration_ms=duration_ms,\n )\n\n try:\n validation_id = self.repository.save_validation(validation)\n except Exception as e:\n return False, {\"error\": f\"Failed to save validation: {e}\"}, None\n\n return True, result.to_dict(), validation_id\n\n def get_latest_validations(\n self,\n contract_id: int,\n limit: int = 10\n ) -> list[dict[str, Any]]:\n \"\"\"\n Get latest validation results for a contract.\n\n Args:\n contract_id: Contract ID\n limit: Maximum number of results\n\n Returns:\n List of validation dictionaries\n \"\"\"\n validations = self.repository.get_validations(contract_id, limit=limit)\n return [v.to_dict() for v in validations]\n"
}
}
{
"tool_response": {
"type": "create",
"filePath": "\/var\/www\/mcp-servers\/mcp-contracts\/tools\/contract_tools\/contract_validator.py",
"content": "\"\"\"Contract Validation Module\"\"\"\nfrom typing import Optional, Any\nfrom datetime import datetime\n\nfrom domain.contracts import ContractValidation\nfrom validators.contract_validator import run_contract_validation\n\nfrom .contract_parser import ContractParser\nfrom .constants import VALIDATION_PASSED, VALIDATION_FAILED\n\n\nclass ContractValidatorService:\n \"\"\"Service for validating contracts against their rules\"\"\"\n\n def __init__(self, repository):\n \"\"\"\n Initialize validator service.\n\n Args:\n repository: ContractRepository instance\n \"\"\"\n self.repository = repository\n self.parser = ContractParser()\n\n def validate_contract(\n self,\n contract_id: int,\n yaml_content: str,\n path: Optional[str] = None,\n triggered_by: str = \"mcp\",\n start_time: float = 0.0\n ) -> tuple[bool, dict[str, Any], Optional[int]]:\n \"\"\"\n Validate a contract and save the result.\n\n Args:\n contract_id: ID of the contract being validated\n yaml_content: YAML content of the contract\n path: Optional specific path to validate\n triggered_by: Who\/what triggered the validation\n start_time: Start time for duration calculation\n\n Returns:\n Tuple of (success, result_dict, validation_id)\n \"\"\"\n # Parse YAML\n try:\n contract_data = self.parser.parse_yaml(yaml_content)\n except Exception as e:\n return False, {\"error\": f\"Invalid contract YAML: {e}\"}, None\n\n # Run validation\n try:\n result = run_contract_validation(contract_data, path)\n except Exception as e:\n return False, {\"error\": f\"Validation failed: {e}\"}, None\n\n # Calculate duration\n duration_ms = 0\n if start_time > 0:\n import time\n duration_ms = int((time.time() - start_time) * 1000)\n\n # Save validation result\n validation = ContractValidation(\n contract_id=contract_id,\n validated_at=datetime.now(),\n result=VALIDATION_PASSED if result.outcome == \"passed\" else VALIDATION_FAILED,\n critical_count=result.critical,\n major_count=result.major,\n minor_count=result.minor,\n violations=result.findings,\n triggered_by=triggered_by,\n target_path=path,\n duration_ms=duration_ms,\n )\n\n try:\n validation_id = self.repository.save_validation(validation)\n except Exception as e:\n return False, {\"error\": f\"Failed to save validation: {e}\"}, None\n\n return True, result.to_dict(), validation_id\n\n def get_latest_validations(\n self,\n contract_id: int,\n limit: int = 10\n ) -> list[dict[str, Any]]:\n \"\"\"\n Get latest validation results for a contract.\n\n Args:\n contract_id: Contract ID\n limit: Maximum number of results\n\n Returns:\n List of validation dictionaries\n \"\"\"\n validations = self.repository.get_validations(contract_id, limit=limit)\n return [v.to_dict() for v in validations]\n",
"structuredPatch": [],
"originalFile": null
}
}