Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions hatch_validator/core/pkg_accessor_factory.py
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,12 @@ def _ensure_accessors_loaded(cls) -> None:
except ImportError as e:
logger.warning(f"Could not load v1.2.2 accessor: {e}")

try:
from hatch_validator.package.v2_0_0.accessor import HatchPkgAccessor as V200HatchPkgAccessor
cls.register_accessor("2.0.0", V200HatchPkgAccessor)
except ImportError as e:
logger.warning(f"Could not load v2.0.0 accessor: {e}")

@classmethod
def create_accessor_chain(cls, target_version: Optional[str] = None) -> HatchPkgAccessor:
"""Create appropriate accessor chain based on target version.
Expand Down
6 changes: 6 additions & 0 deletions hatch_validator/core/validator_factory.py
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,12 @@ def _ensure_validators_loaded(cls) -> None:
cls.register_validator("1.2.2", V122Validator)
except ImportError as e:
logger.warning(f"Could not load v1.2.2 validator: {e}")

try:
from hatch_validator.package.v2_0_0.validator import Validator as V200Validator
cls.register_validator("2.0.0", V200Validator)
except ImportError as e:
logger.warning(f"Could not load v2.0.0 validator: {e}")

@classmethod
def create_validator_chain(cls, target_version: Optional[str] = None) -> Validator:
Expand Down
4 changes: 2 additions & 2 deletions hatch_validator/package/package_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -42,10 +42,10 @@ def load_metadata(self, metadata: Dict[str, Any]) -> None:
ValueError: If no accessor can handle the package metadata.
"""
self._metadata = metadata
schema_version = metadata.get("package_schema_version")
schema_version = metadata.get("hatch_schema_version") or metadata.get("package_schema_version")

if not schema_version:
raise ValueError("Missing 'package_schema_version' in metadata.")
raise ValueError("Missing schema version in metadata. Expected 'hatch_schema_version' or 'package_schema_version'.")

self._accessor = HatchPkgAccessorFactory.create_accessor_chain(schema_version)
if not self._accessor:
Expand Down
6 changes: 6 additions & 0 deletions hatch_validator/package/v2_0_0/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
"""Schema validation package for v2.0.0.

This package contains the validator and strategies for schema version 2.0.0,
which integrates the package metadata format with the Official MCP Registry.
"""

92 changes: 92 additions & 0 deletions hatch_validator/package/v2_0_0/accessor.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,92 @@
"""Package metadata accessor for schema version 2.0.0.

This module provides the metadata accessor for schema version 2.0.0.
Schema version 2.0.0 introduces the new hatch_schema_version field,
uses an authors array instead of a singular author object, and renames
`tools[].description` to `tools[].desc`.
"""

import logging
from typing import Any, Dict

from hatch_validator.core.pkg_accessor_base import HatchPkgAccessor as HatchPkgAccessorBase

logger = logging.getLogger("hatch.package.v2_0_0.accessor")

class HatchPkgAccessor(HatchPkgAccessorBase):
"""Metadata accessor for Hatch package schema version 2.0.0."""

def can_handle(self, schema_version: str) -> bool:
"""Check if this accessor can handle schema version 2.0.0.

Args:
schema_version (str): Schema version to check

Returns:
bool: True if schema_version is '2.0.0'
"""
return schema_version == "2.0.0"

def get_package_schema_version(self, metadata: Dict[str, Any]) -> Any:
"""Get the package schema version value from metadata.

Args:
metadata (Dict[str, Any]): Package metadata

Returns:
Any: Schema version value from either hatch_schema_version or package_schema_version
"""
return metadata.get("hatch_schema_version") or metadata.get("package_schema_version")

def get_author(self, metadata: Dict[str, Any]) -> Any:
"""Get authors from metadata.

Schema v2.0.0 stores author information in an array under the `authors` key.

Args:
metadata (Dict[str, Any]): Package metadata

Returns:
Any: Author information, preferably the authors list or legacy author object
"""
authors = metadata.get("authors")
if isinstance(authors, list):
return authors
if authors is not None:
return [authors]
return metadata.get("author")

def get_tools(self, metadata: Dict[str, Any]) -> Any:
"""Get tools from metadata.

Tools entries in schema v2.0.0 use `desc` instead of `description`.

Args:
metadata (Dict[str, Any]): Package metadata

Returns:
Any: Tools list from metadata
"""
return metadata.get("tools", [])

def get_provenance(self, metadata: Dict[str, Any]) -> Any:
"""Get provenance metadata for v2.0.0.

Args:
metadata (Dict[str, Any]): Package metadata

Returns:
Any: Provenance metadata object or None if not present
"""
return metadata.get("provenance")

def get_citations(self, metadata: Dict[str, Any]) -> Any:
"""Get citations metadata for v2.0.0.

Args:
metadata (Dict[str, Any]): Package metadata

Returns:
Any: Citations list or an empty list if not present
"""
return metadata.get("citations", [])
101 changes: 101 additions & 0 deletions hatch_validator/package/v2_0_0/dependency_validation.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,101 @@
"""Dependency validation strategy for schema version v2.0.0.

This module implements dependency validation for schema version 2.0.0.
Only Docker-specific validation is owned here; Hatch, Python, and System
dependency validation is delegated to v1.2.2 via the chain.
"""

import logging
from typing import Dict, List, Tuple

from hatch_validator.core.validation_strategy import DependencyValidationStrategy
from hatch_validator.core.validation_context import ValidationContext
from hatch_validator.package.package_service import PackageService

logger = logging.getLogger("hatch.dependency_validation_v2_0_0")
logger.setLevel(logging.DEBUG)


class DependencyValidation(DependencyValidationStrategy):
"""Strategy for validating Docker dependencies according to v2.0.0 schema."""

def validate_dependencies(self, metadata: Dict, context: ValidationContext) -> Tuple[bool, List[str]]:
"""Validate Docker dependencies according to v2.0.0 schema.

Hatch, Python, and System dependency validation is handled by the
previous validator in the chain. This strategy validates only the
Docker subset, which requires a digest in v2.0.0.

Args:
metadata (Dict): Package metadata containing dependency information
context (ValidationContext): Validation context with resources

Returns:
Tuple[bool, List[str]]: Tuple containing:
- bool: Whether Docker dependency validation was successful
- List[str]: List of Docker dependency validation errors
"""
try:
package_service = context.get_data("package_service", None)
if package_service is None:
package_service = PackageService(metadata)

dependencies = package_service.get_dependencies()
docker_dependencies = dependencies.get('docker', [])

errors = []
is_valid = True

if docker_dependencies:
docker_valid, docker_errors = self._validate_docker_dependencies(docker_dependencies, context)
if not docker_valid:
errors.extend(docker_errors)
is_valid = False

except Exception as e:
logger.error(f"Error during Docker dependency validation: {e}")
return False, [f"Error during Docker dependency validation: {e}"]

logger.debug(f"Docker dependency validation result: {is_valid}, errors: {errors}")
return is_valid, errors

def _validate_docker_dependencies(self, docker_dependencies: List[Dict],
context: ValidationContext) -> Tuple[bool, List[str]]:
"""Validate Docker image dependencies."""
errors = []
is_valid = True

for dep in docker_dependencies:
dep_valid, dep_errors = self._validate_single_docker_dependency(dep, context)
if not dep_valid:
errors.extend(dep_errors)
is_valid = False

return is_valid, errors

def _validate_single_docker_dependency(self, dep: Dict,
context: ValidationContext) -> Tuple[bool, List[str]]:
"""Validate a single Docker dependency.

Structural checks (digest presence, digest pattern, version_constraint rejection)
are enforced by the JSON schema and are not repeated here.
"""
errors = []
is_valid = True

dep_name = dep.get('name')
if not dep_name:
errors.append("Docker dependency missing name")
return False, errors

tag = dep.get('tag')
if tag is not None and not isinstance(tag, str):
errors.append(f"Invalid Docker tag for '{dep_name}'. Must be a string")
is_valid = False

registry = dep.get('registry')
if registry is not None and not isinstance(registry, str):
errors.append(f"Invalid registry value for Docker dependency '{dep_name}'. Must be a string")
is_valid = False

return is_valid, errors
59 changes: 59 additions & 0 deletions hatch_validator/package/v2_0_0/schema_validation.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,59 @@
"""Schema validation strategy for v2.0.0.

This module provides the schema validation strategy for schema version 2.0.0.
"""

import logging
from typing import Dict, List, Tuple

from hatch_validator.core.validation_strategy import SchemaValidationStrategy
from hatch_validator.core.validation_context import ValidationContext
from hatch_validator.schemas.schemas_retriever import get_package_schema

# Configure logging
logger = logging.getLogger("hatch.schema.v2_0_0.schema_validation")


class SchemaValidation(SchemaValidationStrategy):
"""Strategy for validating metadata against v2.0.0 schema."""

def validate_schema(self, metadata: Dict, context: ValidationContext) -> Tuple[bool, List[str]]:
"""Validate metadata against v2.0.0 schema.

Args:
metadata (Dict): Package metadata to validate against schema
context (ValidationContext): Validation context with resources

Returns:
Tuple[bool, List[str]]: Tuple containing:
- bool: Whether schema validation was successful
- List[str]: List of schema validation errors
"""
try:
jsonschema = __import__("jsonschema")
except ImportError:
error_msg = "jsonschema is required for schema validation but is not installed"
logger.error(error_msg)
return False, [error_msg]

try:
schema = get_package_schema(version="2.0.0", force_update=context.force_schema_update)
if not schema:
error_msg = "Failed to load package schema version 2.0.0"
logger.error(error_msg)
return False, [error_msg]

jsonschema.validate(instance=metadata, schema=schema)
logger.debug("Package metadata successfully validated against v2.0.0 schema")
return True, []

except jsonschema.ValidationError as e:
error_msg = f"Schema validation failed: {e.message}"
if e.absolute_path:
error_msg += f" at path: {'.'.join(str(p) for p in e.absolute_path)}"
logger.error(error_msg)
return False, [error_msg]
except Exception as e:
error_msg = f"Unexpected error during schema validation: {str(e)}"
logger.error(error_msg)
return False, [error_msg]
Loading
Loading