Files

134 lines
4.4 KiB
Python

from datetime import UTC, datetime, timedelta
import pytest
from pydantic import ValidationError
from modelforge_api.domain.contracts import (
ArtifactProvenance,
BenchmarkEnvironmentFingerprint,
CapabilityContractManifest,
MigrationContract,
ProjectBindingManifest,
RuntimeProfileContract,
)
from modelforge_api.domain.enums import MigrationStatus, UpgradeClass
def minimal_embedding_contract(**overrides):
payload = {
"capability": "rag.embedding",
"version": 1,
"description": "test",
"input_schema": {"type": "object"},
"output_schema": {"type": "object"},
"modalities": {"input": ["text"], "output": ["vector"]},
"vector": {"cross_deployment_compatible": False},
"quality_metrics": ["recall_at_5"],
"upgrade_class": "requires_reindex",
"production_priority": "production",
"default_residency": "keep_warm",
"estate": {
"category": "RAG",
"purpose": "test retrieval",
"stability": "experimental",
"resource_class": "LIGHT",
"evaluation_type": "retrieval",
"consumers": [],
"payload_limits": {"max_bytes": 1024},
},
}
payload.update(overrides)
return payload
def test_embedding_contract_requires_reindex_when_spaces_are_not_compatible() -> None:
CapabilityContractManifest.model_validate(minimal_embedding_contract())
with pytest.raises(ValidationError, match="requires_reindex"):
CapabilityContractManifest.model_validate(
minimal_embedding_contract(upgrade_class="transparent")
)
def test_project_binding_rejects_concrete_model_fields() -> None:
with pytest.raises(ValidationError):
ProjectBindingManifest.model_validate(
{
"contract_version": 1,
"channel": "stable",
"priority": "production",
"model_id": "vendor/model",
}
)
def test_artifact_provenance_validates_digest_and_derived_lineage() -> None:
digest = "a" * 64
provenance = ArtifactProvenance(
upstream_repository="org/model",
resolved_commit_sha="b" * 40,
filename="model.safetensors",
artifact_type="weights",
sha256=digest,
)
assert provenance.sha256 == digest
with pytest.raises(ValidationError, match="derivation tool"):
ArtifactProvenance(
upstream_repository="org/model",
resolved_commit_sha="b" * 40,
filename="model.gguf",
artifact_type="weights",
sha256="c" * 64,
source_artifact_sha256=digest,
)
def test_runtime_profile_security_defaults_and_stable_fingerprint() -> None:
profile = RuntimeProfileContract(
runtime_type="llama_cpp",
runtime_version="1.0",
artifact_sha256="a" * 64,
)
assert profile.trust_remote_code is False
assert profile.network_egress is False
assert profile.fingerprint == profile.fingerprint
def test_benchmark_comparability_detects_runtime_or_hardware_change() -> None:
base = BenchmarkEnvironmentFingerprint(
runtime_type="vllm",
runtime_version="1",
runtime_image_digest=None,
launch_arguments={},
cuda_version="13",
driver_version="600",
accelerator_name="GPU A",
accelerator_uuid="gpu-1",
context_length=4096,
concurrency=1,
seed=42,
operating_system="linux",
)
changed = base.model_copy(update={"driver_version": "601"})
assert base.comparable_with(base)
assert not base.comparable_with(changed)
def test_embedding_migration_requires_distinct_shadow_and_reindex() -> None:
now = datetime.now(UTC)
valid = MigrationContract(
project_id="examplerag",
capability="rag.embedding",
source_deployment_id="old",
target_deployment_id="new",
upgrade_class=UpgradeClass.REQUIRES_REINDEX,
current_index_ref="index-v1",
shadow_index_ref="index-v2-shadow",
status=MigrationStatus.PLANNED,
rollback_retain_until=now + timedelta(days=7),
)
assert valid.shadow_index_ref != valid.current_index_ref
with pytest.raises(ValidationError, match="requires_reindex"):
valid.model_copy(
update={"upgrade_class": UpgradeClass.TRANSPARENT}, deep=True
).model_validate({**valid.model_dump(), "upgrade_class": "transparent"})