CI/CD Pipeline Guide¶

# ci/validate_contract.py
import yaml
import sys
from pathlib import Path

def validate_yaml_syntax(file_path: Path) -> bool:
    """Validate YAML syntax"""
    try:
        with open(file_path) as f:
            yaml.safe_load(f)
        return True
    except yaml.YAMLError as e:
        print(f"❌ {file_path}: {e}")
        return False

def main():
    contracts = list(Path("domains").rglob("contract.yaml"))

    errors = []
    for contract in contracts:
        if not validate_yaml_syntax(contract):
            errors.append(contract)

    if errors:
        print(f"\n❌ {len(errors)} contracts have syntax errors")
        sys.exit(1)

    print(f"✓ All {len(contracts)} contracts have valid syntax")

if __name__ == "__main__":
    main()

# ci/validate_against_schema.py
import json
import yaml
from jsonschema import validate, ValidationError
from pathlib import Path

def load_contract_schema() -> dict:
    """Load JSON Schema for contracts"""
    with open("schemas/contract-schema.json") as f:
        return json.load(f)

def validate_contract(contract_path: Path, schema: dict) -> list:
    """Validate contract against schema"""
    errors = []

    with open(contract_path) as f:
        contract = yaml.safe_load(f)

    try:
        validate(instance=contract, schema=schema)
    except ValidationError as e:
        errors.append(f"{contract_path}: {e.message}")

    return errors

def main():
    schema = load_contract_schema()
    contracts = list(Path("domains").rglob("contract.yaml"))

    all_errors = []
    for contract in contracts:
        errors = validate_contract(contract, schema)
        all_errors.extend(errors)

    if all_errors:
        for error in all_errors:
            print(f"❌ {error}")
        sys.exit(1)

    print(f"✓ All {len(contracts)} contracts conform to schema")

# ci/detect_breaking_changes.py
import yaml
import subprocess
from enum import Enum
from pathlib import Path
from pydantic import BaseModel

class ChangeType(str, Enum):
    BREAKING = "breaking"
    NON_BREAKING = "non_breaking"
    PATCH = "patch"

class Change(BaseModel):
    """Represents a change detected in contract."""
    contract: str
    field: str
    change_type: ChangeType
    description: str

    model_config = {
        "use_enum_values": True,
    }

def get_file_at_ref(file_path: str, ref: str) -> str:
    """Get file content at specific git ref"""
    result = subprocess.run(
        ["git", "show", f"{ref}:{file_path}"],
        capture_output=True,
        text=True
    )
    return result.stdout if result.returncode == 0 else None

def detect_breaking_changes(old_contract: dict, new_contract: dict) -> list[Change]:
    """Detect breaking changes between contract versions"""
    changes = []

    old_fields = {f['name']: f for f in old_contract.get('schema', {}).get('fields', [])}
    new_fields = {f['name']: f for f in new_contract.get('schema', {}).get('fields', [])}

    # Removed fields = BREAKING
    for field_name in old_fields:
        if field_name not in new_fields:
            changes.append(Change(
                contract=old_contract['metadata']['name'],
                field=field_name,
                change_type=ChangeType.BREAKING,
                description=f"Field '{field_name}' was removed"
            ))

    # Type changes = BREAKING
    for field_name, new_field in new_fields.items():
        if field_name in old_fields:
            old_field = old_fields[field_name]
            if old_field.get('type') != new_field.get('type'):
                changes.append(Change(
                    contract=old_contract['metadata']['name'],
                    field=field_name,
                    change_type=ChangeType.BREAKING,
                    description=f"Field '{field_name}' type changed from {old_field.get('type')} to {new_field.get('type')}"
                ))

            # Required: false -> true = BREAKING
            if not old_field.get('required') and new_field.get('required'):
                changes.append(Change(
                    contract=old_contract['metadata']['name'],
                    field=field_name,
                    change_type=ChangeType.BREAKING,
                    description=f"Field '{field_name}' became required"
                ))

    # New nullable field = NON_BREAKING
    for field_name, new_field in new_fields.items():
        if field_name not in old_fields:
            if new_field.get('required', False):
                changes.append(Change(
                    contract=old_contract['metadata']['name'],
                    field=field_name,
                    change_type=ChangeType.BREAKING,
                    description=f"New required field '{field_name}' added"
                ))
            else:
                changes.append(Change(
                    contract=old_contract['metadata']['name'],
                    field=field_name,
                    change_type=ChangeType.NON_BREAKING,
                    description=f"New optional field '{field_name}' added"
                ))

    return changes

def main():
    # Parse arguments and run detection
    # Output results as JSON
    pass

# ci/check_version_bump.py
import yaml
from packaging import version
from pathlib import Path

def check_version_bump(old_contract: dict, new_contract: dict, changes: list) -> tuple[bool, str]:
    """Check if version bump is correct for changes"""

    old_version = version.parse(old_contract.get('contract_version', '0.0.0'))
    new_version = version.parse(new_contract.get('contract_version', '0.0.0'))

    has_breaking = any(c.change_type == ChangeType.BREAKING for c in changes)
    has_non_breaking = any(c.change_type == ChangeType.NON_BREAKING for c in changes)

    if has_breaking:
        # Must bump MAJOR
        if new_version.major <= old_version.major:
            return False, f"Breaking changes require MAJOR version bump: {old_version} -> {new_version.major + 1}.0.0"
    elif has_non_breaking:
        # Must bump at least MINOR
        if new_version.minor <= old_version.minor and new_version.major == old_version.major:
            return False, f"Non-breaking changes require MINOR version bump: {old_version} -> {old_version.major}.{old_version.minor + 1}.0"
    else:
        # PATCH for description changes
        if new_version <= old_version:
            return False, f"Changes require at least PATCH version bump"

    return True, f"Version bump OK: {old_version} -> {new_version}"

# ci/deploy_to_registry.py
import requests
import yaml
import json
from pathlib import Path

SCHEMA_REGISTRY_URL = os.environ.get('SCHEMA_REGISTRY_URL', 'http://localhost:8081')

def contract_to_avro_schema(contract: dict) -> dict:
    """Convert contract schema to Avro schema"""
    fields = []
    for field in contract['schema']['fields']:
        avro_field = {
            "name": field['name'],
            "type": map_type_to_avro(field['type'], field.get('required', True))
        }
        if field.get('default') is not None:
            avro_field['default'] = field['default']
        fields.append(avro_field)

    return {
        "type": "record",
        "name": contract['metadata']['name'],
        "namespace": contract['metadata']['namespace'],
        "fields": fields
    }

def deploy_schema(contract_path: Path):
    """Deploy contract schema to Schema Registry"""
    with open(contract_path) as f:
        contract = yaml.safe_load(f)

    avro_schema = contract_to_avro_schema(contract)
    subject = f"{contract['metadata']['namespace']}.{contract['metadata']['name']}-value"

    response = requests.post(
        f"{SCHEMA_REGISTRY_URL}/subjects/{subject}/versions",
        headers={"Content-Type": "application/vnd.schemaregistry.v1+json"},
        json={"schema": json.dumps(avro_schema)}
    )

    if response.status_code in [200, 201]:
        print(f"✓ Deployed {subject} version {response.json().get('id')}")
    else:
        print(f"❌ Failed to deploy {subject}: {response.text}")
        return False

    return True