contract_validator.py

• Pfad: /var/www/mcp-servers/mcp-contracts/tools/contract_tools_components/contract_validator.py
• Namespace: -
• Zeilen: 102 | Größe: 3,200 Bytes
• Geändert: 2025-12-28 12:21:49 | Gescannt: 2025-12-31 10:22:15

Code Hygiene Score: 89

• Dependencies: 60 (25%)
• LOC: 100 (20%)
• Methods: 100 (20%)
• Secrets: 100 (15%)
• Classes: 100 (10%)
• Magic Numbers: 90 (10%)

Issues 1

Dependencies 9

• use typing.Optional
• use typing.Any
• use datetime.datetime
• use domain.contracts.ContractValidation
• use validators.contract_validator.run_contract_validation
• use contract_parser.ContractParser
• use constants.VALIDATION_PASSED
• use constants.VALIDATION_FAILED
• use time

Klassen 1

• ContractValidatorService class Zeile 12

Code

"""Contract Validation Module"""
from typing import Optional, Any
from datetime import datetime

from domain.contracts import ContractValidation
from validators.contract_validator import run_contract_validation

from .contract_parser import ContractParser
from .constants import VALIDATION_PASSED, VALIDATION_FAILED


class ContractValidatorService:
    """Service for validating contracts against their rules"""

    def __init__(self, repository):
        """
        Initialize validator service.

        Args:
            repository: ContractRepository instance
        """
        self.repository = repository
        self.parser = ContractParser()

    def validate_contract(
        self,
        contract_id: int,
        yaml_content: str,
        path: Optional[str] = None,
        triggered_by: str = "mcp",
        start_time: float = 0.0
    ) -> tuple[bool, dict[str, Any], Optional[int]]:
        """
        Validate a contract and save the result.

        Args:
            contract_id: ID of the contract being validated
            yaml_content: YAML content of the contract
            path: Optional specific path to validate
            triggered_by: Who/what triggered the validation
            start_time: Start time for duration calculation

        Returns:
            Tuple of (success, result_dict, validation_id)
        """
        # Parse YAML
        try:
            contract_data = self.parser.parse_yaml(yaml_content)
        except Exception as e:
            return False, {"error": f"Invalid contract YAML: {e}"}, None

        # Run validation
        try:
            result = run_contract_validation(contract_data, path)
        except Exception as e:
            return False, {"error": f"Validation failed: {e}"}, None

        # Calculate duration
        duration_ms = 0
        if start_time > 0:
            import time
            duration_ms = int((time.time() - start_time) * 1000)

        # Save validation result
        validation = ContractValidation(
            contract_id=contract_id,
            validated_at=datetime.now(),
            result=VALIDATION_PASSED if result.outcome == "passed" else VALIDATION_FAILED,
            critical_count=result.critical,
            major_count=result.major,
            minor_count=result.minor,
            violations=result.findings,
            triggered_by=triggered_by,
            target_path=path,
            duration_ms=duration_ms,
        )

        try:
            validation_id = self.repository.save_validation(validation)
        except Exception as e:
            return False, {"error": f"Failed to save validation: {e}"}, None

        return True, result.to_dict(), validation_id

    def get_latest_validations(
        self,
        contract_id: int,
        limit: int = 10
    ) -> list[dict[str, Any]]:
        """
        Get latest validation results for a contract.

        Args:
            contract_id: Contract ID
            limit: Maximum number of results

        Returns:
            List of validation dictionaries
        """
        validations = self.repository.get_validations(contract_id, limit=limit)
        return [v.to_dict() for v in validations]