864 lines
32 KiB
Python
864 lines
32 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import uuid
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
from sqlalchemy import create_engine, update
|
|
from sqlalchemy.orm import Session
|
|
|
|
from modelforge_api.domain.runtime import (
|
|
AgentRuntimeProbeComplete,
|
|
AgentRuntimeProbeFailure,
|
|
AgentRuntimeProbeProgress,
|
|
CompatibilityAssessmentCreate,
|
|
ExecutionApprovalCreate,
|
|
RuntimeEnvironmentCreate,
|
|
RuntimeProbeCreate,
|
|
RuntimeProfileCreate,
|
|
)
|
|
from modelforge_api.persistence.models import (
|
|
Accelerator,
|
|
ArtifactLocation,
|
|
ArtifactSet,
|
|
ArtifactSetMember,
|
|
Base,
|
|
ComputeNode,
|
|
Model,
|
|
ModelArtifact,
|
|
ModelRevision,
|
|
StorageRoot,
|
|
UpstreamSnapshot,
|
|
)
|
|
from modelforge_api.services.registry import RegistryConflict
|
|
from modelforge_api.services.runtime import RuntimeService
|
|
from modelforge_api.settings import Settings
|
|
|
|
|
|
@pytest.fixture
|
|
def session() -> Session:
|
|
engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as value:
|
|
yield value
|
|
|
|
|
|
def setup_runtime(session: Session):
|
|
now = datetime.now(UTC)
|
|
model = Model(
|
|
key="qwen-runtime",
|
|
display_name="Qwen3 Embedding",
|
|
upstream_provider="Qwen",
|
|
upstream_source="Qwen/Qwen3-Embedding-0.6B",
|
|
upstream_metadata={
|
|
"tags": ["qwen3", "safetensors", "sentence-transformers"],
|
|
"pipeline_tag": "feature-extraction",
|
|
"library_name": "sentence-transformers",
|
|
},
|
|
local_metadata={},
|
|
interpretation_metadata={},
|
|
modalities=[],
|
|
parameter_metadata={},
|
|
license_metadata={"status": "captured_unreviewed"},
|
|
lifecycle="candidate",
|
|
)
|
|
node = ComputeNode(
|
|
key="gpu_node",
|
|
hostname="GPU Node",
|
|
enabled=True,
|
|
liveness_state="online",
|
|
lab_eligible=True,
|
|
agent_capabilities=["runtime.probe.v1", "runtime.health.v1", "runtime.unload.v1"],
|
|
hardware_fingerprint="h" * 64,
|
|
)
|
|
session.add_all([model, node])
|
|
session.flush()
|
|
accelerator = Accelerator(
|
|
compute_node_id=node.id,
|
|
device_index=0,
|
|
device_uuid="GPU-test",
|
|
name="NVIDIA RTX",
|
|
compute_capability_major=8,
|
|
compute_capability_minor=9,
|
|
total_vram_bytes=17_171_480_576,
|
|
driver_version="575.64.05",
|
|
cuda_version="12.9",
|
|
status="active",
|
|
)
|
|
root = StorageRoot(
|
|
compute_node_id=node.id,
|
|
name="models",
|
|
path="/host/models",
|
|
agent_path="/data/artifacts/model-registry",
|
|
status="ready",
|
|
writable=True,
|
|
)
|
|
revision = ModelRevision(
|
|
model_id=model.id,
|
|
upstream_revision="main",
|
|
resolved_commit_sha="a" * 40,
|
|
metadata_snapshot={},
|
|
immutable_at=now,
|
|
)
|
|
session.add_all([accelerator, root, revision])
|
|
session.flush()
|
|
snapshot = UpstreamSnapshot(
|
|
model_id=model.id,
|
|
repository_id=model.upstream_source,
|
|
requested_revision="main",
|
|
resolved_commit_sha=revision.resolved_commit_sha,
|
|
access_state="public",
|
|
metadata_snapshot={},
|
|
card_metadata={},
|
|
security_metadata={},
|
|
stale_after=now + timedelta(hours=1),
|
|
)
|
|
session.add(snapshot)
|
|
session.flush()
|
|
names = [
|
|
("config.json", "configuration", "json", "1" * 64, 100),
|
|
("tokenizer.json", "tokenizer", "json", "2" * 64, 100),
|
|
("modules.json", "configuration", "json", "3" * 64, 100),
|
|
("1_Pooling/config.json", "configuration", "json", "4" * 64, 100),
|
|
("model.safetensors", "weights", "safetensors", "5" * 64, 1000),
|
|
]
|
|
artifacts = []
|
|
for name, role, file_format, digest, size in names:
|
|
artifact = ModelArtifact(
|
|
revision_id=revision.id,
|
|
filename=name,
|
|
artifact_type=role,
|
|
serialization_format=file_format,
|
|
sha256=digest,
|
|
size_bytes=size,
|
|
status="verified",
|
|
security_status="static_checks_passed_unapproved",
|
|
license_status="captured_unreviewed",
|
|
quarantined=False,
|
|
verification_details={"risk_flags": []},
|
|
verified_at=now,
|
|
immutable_at=now,
|
|
)
|
|
session.add(artifact)
|
|
session.flush()
|
|
artifacts.append(artifact)
|
|
artifact_set = ArtifactSet(
|
|
revision_id=revision.id,
|
|
snapshot_id=snapshot.id,
|
|
variant_key="safetensors-default",
|
|
label="Safetensors",
|
|
selection_reason="test",
|
|
selected_paths=[name for name, *_ in names],
|
|
total_size_bytes=1400,
|
|
file_count=5,
|
|
availability="local",
|
|
status="verified",
|
|
completeness="complete",
|
|
security_status="static_checks_passed_unapproved",
|
|
license_status="captured_unreviewed",
|
|
immutable_at=now,
|
|
)
|
|
session.add(artifact_set)
|
|
session.flush()
|
|
parent = "repositories/Qwen--Qwen3-Embedding/" + revision.resolved_commit_sha
|
|
for ordinal, artifact in enumerate(artifacts):
|
|
session.add_all(
|
|
[
|
|
ArtifactSetMember(
|
|
artifact_set_id=artifact_set.id,
|
|
artifact_id=artifact.id,
|
|
ordinal=ordinal,
|
|
required=True,
|
|
),
|
|
ArtifactLocation(
|
|
artifact_id=artifact.id,
|
|
storage_root_id=root.id,
|
|
relative_path=f"{parent}/{artifact.filename}",
|
|
status="verified",
|
|
size_bytes=artifact.size_bytes,
|
|
observed_sha256=artifact.sha256,
|
|
last_checked_at=now,
|
|
),
|
|
]
|
|
)
|
|
session.commit()
|
|
service = RuntimeService(
|
|
session,
|
|
Settings(database_url="sqlite+pysqlite:///:memory:", runtime_artifact_root="/models"),
|
|
)
|
|
return service, artifact_set, node, artifacts
|
|
|
|
|
|
def create_environment(service: RuntimeService, adapter: str = "sentence_transformers"):
|
|
supported = (
|
|
["qwen3"]
|
|
if adapter in {"sentence_transformers", "qwen3_reranker", "transformers", "vllm"}
|
|
else []
|
|
)
|
|
return service.create_environment(
|
|
RuntimeEnvironmentCreate(
|
|
name=f"{adapter} pinned",
|
|
adapter=adapter,
|
|
runtime_version="4.1.0",
|
|
image_repository="modelforge-runtime-worker",
|
|
image_digest="sha256:" + "a" * 64,
|
|
python_version="3.12.11",
|
|
cuda_runtime_version="12.8",
|
|
package_versions={
|
|
"torch": "2.7.1",
|
|
"transformers": "4.51.3",
|
|
"sentence-transformers": "4.1.0",
|
|
adapter: "4.1.0",
|
|
"llama-cpp-python": "0.3.16",
|
|
"diffusers": "0.35.1",
|
|
},
|
|
supported_model_types=supported,
|
|
supported_formats=["safetensors"],
|
|
supported_modalities=["reranking" if adapter == "qwen3_reranker" else "embedding"],
|
|
)
|
|
)
|
|
|
|
|
|
def create_profile(
|
|
service: RuntimeService, artifact_set_id: uuid.UUID, adapter: str = "sentence_transformers"
|
|
):
|
|
environment = create_environment(service, adapter)
|
|
return service.create_profile(
|
|
RuntimeProfileCreate(
|
|
name=f"Qwen {adapter}",
|
|
runtime_environment_id=environment.id,
|
|
artifact_set_id=artifact_set_id,
|
|
modality="reranking" if adapter == "qwen3_reranker" else "embedding",
|
|
)
|
|
)
|
|
|
|
|
|
def prepare_probe(service: RuntimeService, artifact_set, node):
|
|
profile = create_profile(service, artifact_set.id)
|
|
assessment = service.assess(
|
|
CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id)
|
|
)
|
|
approval = service.approve(
|
|
artifact_set.id,
|
|
ExecutionApprovalCreate(
|
|
reason="Reviewed exact verified set for isolated lab execution",
|
|
approved_by="test-operator",
|
|
),
|
|
)
|
|
probe = service.create_probe(
|
|
RuntimeProbeCreate(
|
|
compatibility_assessment_id=assessment.id,
|
|
execution_approval_id=approval.id,
|
|
)
|
|
)
|
|
return profile, assessment, approval, probe
|
|
|
|
|
|
def test_static_sentence_transformers_compatibility_is_reproducible(session: Session) -> None:
|
|
service, artifact_set, node, _artifacts = setup_runtime(session)
|
|
profile = create_profile(service, artifact_set.id)
|
|
result = service.assess(
|
|
CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id)
|
|
)
|
|
assert result.status == "compatible"
|
|
assert result.artifact_facts["required_complete"] is True
|
|
assert result.evidence["locally_proven_support"] is False
|
|
assert result.evidence["resource_estimate"]["kind"] == "estimated"
|
|
revision = service.repo.revision(artifact_set.revision_id)
|
|
assert revision is not None
|
|
filtered = service.assessments(
|
|
model_id=revision.model_id,
|
|
compute_node_id=node.id,
|
|
runtime_profile_id=profile.id,
|
|
status="compatible",
|
|
)
|
|
assert [item.id for item in filtered] == [result.id]
|
|
assert service.assessments(status="blocked") == []
|
|
|
|
|
|
def test_artifact_manifest_allows_distinct_paths_with_the_same_content_digest(
|
|
session: Session,
|
|
) -> None:
|
|
service, artifact_set, node, artifacts = setup_runtime(session)
|
|
tokenizer = artifacts[1]
|
|
existing_location = service.repo.artifact_locations([tokenizer.id])[0]
|
|
duplicate_path = "tokenizer_config.json"
|
|
selected_paths = [*artifact_set.selected_paths, duplicate_path]
|
|
session.execute(
|
|
update(ArtifactSet)
|
|
.where(ArtifactSet.id == artifact_set.id)
|
|
.values(selected_paths=selected_paths, file_count=len(selected_paths))
|
|
)
|
|
session.add(
|
|
ArtifactLocation(
|
|
artifact_id=tokenizer.id,
|
|
storage_root_id=existing_location.storage_root_id,
|
|
relative_path=existing_location.relative_path.rsplit("/", 1)[0]
|
|
+ f"/{duplicate_path}",
|
|
status="verified",
|
|
size_bytes=tokenizer.size_bytes,
|
|
observed_sha256=tokenizer.sha256,
|
|
last_checked_at=datetime.now(UTC),
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
profile = create_profile(service, artifact_set.id)
|
|
result = service.assess(
|
|
CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id)
|
|
)
|
|
|
|
assert result.status == "compatible"
|
|
assert result.artifact_facts["required_complete"] is True
|
|
assert result.artifact_facts["file_count"] == 6
|
|
assert result.artifact_facts["content_blob_count"] == 5
|
|
assert duplicate_path in result.artifact_facts["verified_location_paths"]
|
|
|
|
approval = service.approve(
|
|
artifact_set.id,
|
|
ExecutionApprovalCreate(reason="Reviewed duplicate-content manifest", approved_by="test"),
|
|
)
|
|
probe = service.create_probe(
|
|
RuntimeProbeCreate(
|
|
compatibility_assessment_id=result.id,
|
|
execution_approval_id=approval.id,
|
|
)
|
|
)
|
|
lease = service.claim_next(node)
|
|
assert lease is not None
|
|
assert lease.probe_id == probe.id
|
|
manifest_paths = [item["path"] for item in lease.expected_manifest["files"]]
|
|
assert manifest_paths == selected_paths
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("adapter", "blocker"),
|
|
[("llama_cpp", "NO_GGUF_ARTIFACT_VARIANT"), ("diffusers", "NOT_A_DIFFUSION_PIPELINE")],
|
|
)
|
|
def test_wrong_runtime_is_incompatible_by_artifact_evidence(
|
|
session: Session, adapter: str, blocker: str
|
|
) -> None:
|
|
service, artifact_set, node, _artifacts = setup_runtime(session)
|
|
profile = create_profile(service, artifact_set.id, adapter)
|
|
result = service.assess(
|
|
CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id)
|
|
)
|
|
assert result.status == "blocked"
|
|
assert blocker in result.blockers
|
|
|
|
|
|
@pytest.mark.parametrize("mutation", ["incomplete", "corrupt", "remote-code", "pickle", "no-gpu"])
|
|
def test_static_assessment_fails_closed(session: Session, mutation: str) -> None:
|
|
service, artifact_set, node, artifacts = setup_runtime(session)
|
|
if mutation == "incomplete":
|
|
artifact_set.completeness = "partial"
|
|
elif mutation == "corrupt":
|
|
artifacts[0].status = "corrupt"
|
|
elif mutation == "remote-code":
|
|
artifacts[0].verification_details = {"risk_flags": ["remote_code"]}
|
|
elif mutation == "pickle":
|
|
artifacts[-1].verification_details = {"risk_flags": ["pickle_or_executable_serialization"]}
|
|
else:
|
|
session.query(Accelerator).delete()
|
|
session.commit()
|
|
profile = create_profile(service, artifact_set.id)
|
|
result = service.assess(
|
|
CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id)
|
|
)
|
|
assert result.status == "blocked"
|
|
|
|
|
|
def test_runtime_profile_rejects_secrets_and_is_immutable(session: Session) -> None:
|
|
service, artifact_set, _node, _artifacts = setup_runtime(session)
|
|
environment = create_environment(service)
|
|
with pytest.raises(ValidationError, match="secrets"):
|
|
RuntimeProfileCreate(
|
|
name="unsafe",
|
|
runtime_environment_id=environment.id,
|
|
artifact_set_id=artifact_set.id,
|
|
modality="embedding",
|
|
environment_variables={"HF_TOKEN": "secret"},
|
|
)
|
|
safe = RuntimeProfileCreate(
|
|
name="offline tokenizer",
|
|
runtime_environment_id=environment.id,
|
|
artifact_set_id=artifact_set.id,
|
|
modality="embedding",
|
|
environment_variables={"TOKENIZERS_PARALLELISM": "false"},
|
|
)
|
|
assert safe.environment_variables == {"TOKENIZERS_PARALLELISM": "false"}
|
|
profile = create_profile(service, artifact_set.id)
|
|
stored = service.repo.profile(profile.id)
|
|
assert stored is not None
|
|
stored.dtype = "float16"
|
|
with pytest.raises(ValueError, match="immutable approved fields"):
|
|
session.commit()
|
|
session.rollback()
|
|
|
|
|
|
def test_probe_requires_exact_valid_lab_approval(session: Session) -> None:
|
|
service, artifact_set, node, artifacts = setup_runtime(session)
|
|
profile = create_profile(service, artifact_set.id)
|
|
assessment = service.assess(
|
|
CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id)
|
|
)
|
|
other_set = ArtifactSet(
|
|
revision_id=artifact_set.revision_id,
|
|
snapshot_id=artifact_set.snapshot_id,
|
|
variant_key="safetensors-second",
|
|
label="Second set",
|
|
selection_reason="approval scoping test",
|
|
selected_paths=artifact_set.selected_paths,
|
|
total_size_bytes=artifact_set.total_size_bytes,
|
|
file_count=artifact_set.file_count,
|
|
availability="local",
|
|
status="verified",
|
|
completeness="complete",
|
|
security_status="static_checks_passed_unapproved",
|
|
license_status="captured_unreviewed",
|
|
immutable_at=datetime.now(UTC),
|
|
)
|
|
session.add(other_set)
|
|
session.flush()
|
|
for ordinal, artifact in enumerate(artifacts):
|
|
session.add(
|
|
ArtifactSetMember(
|
|
artifact_set_id=other_set.id,
|
|
artifact_id=artifact.id,
|
|
ordinal=ordinal,
|
|
required=True,
|
|
)
|
|
)
|
|
session.commit()
|
|
wrong = service.approve(
|
|
other_set.id,
|
|
ExecutionApprovalCreate(reason="Separate reviewed artifact approval", approved_by="test"),
|
|
)
|
|
with pytest.raises(RegistryConflict, match="another artifact set"):
|
|
service.create_probe(
|
|
RuntimeProbeCreate(
|
|
compatibility_assessment_id=assessment.id,
|
|
execution_approval_id=wrong.id,
|
|
)
|
|
)
|
|
|
|
|
|
def test_probe_is_node_owned_cancelable_and_creates_no_candidate_early(session: Session) -> None:
|
|
service, artifact_set, node, _artifacts = setup_runtime(session)
|
|
_profile, _assessment, _approval, probe = prepare_probe(service, artifact_set, node)
|
|
other = ComputeNode(
|
|
key="workstation",
|
|
hostname="workstation",
|
|
enabled=True,
|
|
liveness_state="online",
|
|
lab_eligible=True,
|
|
agent_capabilities=["runtime.probe.v1"],
|
|
)
|
|
session.add(other)
|
|
session.commit()
|
|
assert service.claim_next(other) is None
|
|
assert service.candidates() == []
|
|
cancelled = service.cancel(probe.id)
|
|
assert cancelled.status == "cancelled"
|
|
assert service.claim_next(node) is None
|
|
|
|
|
|
def test_successful_probe_persists_measured_evidence_and_lab_candidate(session: Session) -> None:
|
|
service, artifact_set, node, _artifacts = setup_runtime(session)
|
|
_profile, _assessment, _approval, probe = prepare_probe(service, artifact_set, node)
|
|
lease = service.claim_next(node)
|
|
assert lease is not None and lease.probe_id == probe.id
|
|
claimed = service.probe(probe.id)
|
|
assert claimed.logs_reference == f"runtime-worker://{node.id}/{probe.id}/attempt/1"
|
|
completed = service.complete(
|
|
probe.id,
|
|
node,
|
|
AgentRuntimeProbeComplete(
|
|
lease_token=lease.lease_token,
|
|
load_result={"status": "passed", "load_time_ms": 1000.0},
|
|
health_result={
|
|
"process": "healthy",
|
|
"runtime": "healthy",
|
|
"model": "healthy",
|
|
"capability": "not_routed_in_m4",
|
|
},
|
|
inference_result={
|
|
"status": "passed",
|
|
"shape": [1, 1024],
|
|
"dimension": 1024,
|
|
"finite": True,
|
|
"latency_ms": 10.0,
|
|
},
|
|
unload_result={"status": "passed", "reclaimed": True},
|
|
measured_resources={
|
|
"kind": "measured",
|
|
"samples": {"before_load": {"used_vram_bytes": 1}},
|
|
},
|
|
runtime_facts={"offline_local_only": True, "trust_remote_code": False},
|
|
environment_fingerprint=hashlib.sha256(b"runtime").hexdigest(),
|
|
),
|
|
)
|
|
assert completed.status == "completed"
|
|
candidate = service.candidates()[0]
|
|
assert candidate.status == "lab_ready"
|
|
assert candidate.production is False
|
|
assert len(service.repo.metrics(probe.id)) == 1
|
|
|
|
|
|
def test_reranking_probe_accepts_finite_ranked_score_evidence(session: Session) -> None:
|
|
service, artifact_set, node, _artifacts = setup_runtime(session)
|
|
profile = create_profile(service, artifact_set.id, "qwen3_reranker")
|
|
assert profile.health_contract == {
|
|
"process": "required",
|
|
"runtime": "required",
|
|
"model": "functional_ranking_required",
|
|
"capability": "rag.reranking@1",
|
|
}
|
|
assessment = service.assess(
|
|
CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id)
|
|
)
|
|
approval = service.approve(
|
|
artifact_set.id,
|
|
ExecutionApprovalCreate(reason="Reviewed exact reranker artifact set", approved_by="test"),
|
|
)
|
|
probe = service.create_probe(
|
|
RuntimeProbeCreate(
|
|
compatibility_assessment_id=assessment.id,
|
|
execution_approval_id=approval.id,
|
|
)
|
|
)
|
|
lease = service.claim_next(node)
|
|
assert lease is not None
|
|
completed = service.complete(
|
|
probe.id,
|
|
node,
|
|
AgentRuntimeProbeComplete(
|
|
lease_token=lease.lease_token,
|
|
load_result={"status": "passed", "load_time_ms": 500.0},
|
|
health_result={
|
|
"process": "healthy",
|
|
"runtime": "healthy",
|
|
"model": "healthy",
|
|
"capability": "healthy",
|
|
},
|
|
inference_result={
|
|
"status": "passed",
|
|
"output_type": "ranked_scores",
|
|
"count": 1,
|
|
"finite": True,
|
|
"score": 0.75,
|
|
"latency_ms": 20.0,
|
|
},
|
|
unload_result={"status": "passed", "reclaimed": True},
|
|
measured_resources={"kind": "measured", "samples": {}},
|
|
runtime_facts={"offline_local_only": True, "trust_remote_code": False},
|
|
environment_fingerprint="e" * 64,
|
|
),
|
|
)
|
|
assert completed.status == "completed"
|
|
assert completed.inference_result["output_type"] == "ranked_scores"
|
|
assert service.candidates()[0].runtime_probe_id == probe.id
|
|
|
|
|
|
def test_completion_rejects_unhealthy_or_online_dependent_evidence(session: Session) -> None:
|
|
service, artifact_set, node, _artifacts = setup_runtime(session)
|
|
_profile, _assessment, _approval, probe = prepare_probe(service, artifact_set, node)
|
|
lease = service.claim_next(node)
|
|
assert lease is not None
|
|
failed = service.complete(
|
|
probe.id,
|
|
node,
|
|
AgentRuntimeProbeComplete(
|
|
lease_token=lease.lease_token,
|
|
load_result={"status": "passed"},
|
|
health_result={"process": "healthy", "runtime": "healthy", "model": "unhealthy"},
|
|
inference_result={"finite": False, "dimension": 0},
|
|
unload_result={"reclaimed": True},
|
|
measured_resources={"samples": {}},
|
|
runtime_facts={"offline_local_only": True},
|
|
environment_fingerprint="f" * 64,
|
|
),
|
|
)
|
|
assert failed.status == "failed"
|
|
assert failed.failure_code == "HEALTHCHECK_FAILED"
|
|
assert service.candidates() == []
|
|
|
|
|
|
def test_runtime_environment_and_profile_creation_are_idempotent(session: Session) -> None:
|
|
service, artifact_set, _node, _artifacts = setup_runtime(session)
|
|
first_environment = create_environment(service)
|
|
second_environment = create_environment(service)
|
|
first_profile = create_profile(service, artifact_set.id)
|
|
second_profile = create_profile(service, artifact_set.id)
|
|
assert first_environment.id == second_environment.id
|
|
assert first_profile.id == second_profile.id
|
|
assert first_profile.image_digest == first_environment.image_digest
|
|
assert first_profile.version == 1
|
|
|
|
|
|
def test_runtime_environment_identity_is_immutable(session: Session) -> None:
|
|
service, _artifact_set, _node, _artifacts = setup_runtime(session)
|
|
environment = create_environment(service)
|
|
stored = service.repo.environment(environment.id)
|
|
assert stored is not None
|
|
stored.image_digest = "sha256:" + "b" * 64
|
|
with pytest.raises(ValueError, match="immutable approved fields"):
|
|
session.commit()
|
|
session.rollback()
|
|
|
|
|
|
def test_missing_runtime_dependency_blocks_static_compatibility(session: Session) -> None:
|
|
service, artifact_set, node, _artifacts = setup_runtime(session)
|
|
environment = service.create_environment(
|
|
RuntimeEnvironmentCreate(
|
|
name="missing sentence transformers",
|
|
adapter="sentence_transformers",
|
|
runtime_version="4.1.0",
|
|
image_repository="worker",
|
|
image_digest="sha256:" + "c" * 64,
|
|
python_version="3.11",
|
|
cuda_runtime_version="12.8",
|
|
package_versions={"torch": "2.7.1", "transformers": "4.51.3"},
|
|
supported_model_types=["qwen3"],
|
|
supported_formats=["safetensors"],
|
|
supported_modalities=["embedding"],
|
|
)
|
|
)
|
|
profile = service.create_profile(
|
|
RuntimeProfileCreate(
|
|
name="missing dependency",
|
|
runtime_environment_id=environment.id,
|
|
artifact_set_id=artifact_set.id,
|
|
modality="embedding",
|
|
)
|
|
)
|
|
result = service.assess(
|
|
CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id)
|
|
)
|
|
assert "RUNTIME_DEPENDENCY_MISSING:sentence-transformers" in result.blockers
|
|
|
|
|
|
def test_modality_and_compute_capability_are_enforced(session: Session) -> None:
|
|
service, artifact_set, node, _artifacts = setup_runtime(session)
|
|
environment = service.create_environment(
|
|
RuntimeEnvironmentCreate(
|
|
name="text-only sentence runtime",
|
|
adapter="sentence_transformers",
|
|
runtime_version="4.1.0",
|
|
image_repository="worker",
|
|
image_digest="sha256:" + "d" * 64,
|
|
python_version="3.11",
|
|
cuda_runtime_version="12.8",
|
|
package_versions={"sentence-transformers": "4.1.0"},
|
|
supported_model_types=["qwen3"],
|
|
supported_formats=["safetensors"],
|
|
supported_modalities=["text_generation"],
|
|
)
|
|
)
|
|
profile = service.create_profile(
|
|
RuntimeProfileCreate(
|
|
name="incompatible GPU and modality",
|
|
runtime_environment_id=environment.id,
|
|
artifact_set_id=artifact_set.id,
|
|
modality="embedding",
|
|
gpu_memory_policy={"minimum_compute_capability": "9.0"},
|
|
)
|
|
)
|
|
result = service.assess(
|
|
CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id)
|
|
)
|
|
assert "RUNTIME_MODALITY_UNSUPPORTED" in result.blockers
|
|
assert "GPU_COMPUTE_CAPABILITY_INCOMPATIBLE" in result.blockers
|
|
|
|
|
|
def test_insufficient_vram_estimate_blocks_without_loading(session: Session) -> None:
|
|
service, artifact_set, node, _artifacts = setup_runtime(session)
|
|
service.repo.accelerators(node.id)[0].total_vram_bytes = 100
|
|
session.commit()
|
|
profile = create_profile(service, artifact_set.id)
|
|
result = service.assess(
|
|
CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id)
|
|
)
|
|
assert "INSUFFICIENT_VRAM_ESTIMATE" in result.blockers
|
|
assert result.evidence["resource_estimate"]["kind"] == "estimated"
|
|
|
|
|
|
def test_unknown_architecture_is_requires_probe_not_claimed_compatible(session: Session) -> None:
|
|
service, artifact_set, node, _artifacts = setup_runtime(session)
|
|
model = service.repo.model(service.repo.revision(artifact_set.revision_id).model_id) # type: ignore[union-attr]
|
|
assert model is not None
|
|
model.upstream_metadata = {"pipeline_tag": "feature-extraction"}
|
|
session.commit()
|
|
profile = create_profile(service, artifact_set.id)
|
|
result = service.assess(
|
|
CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id)
|
|
)
|
|
assert result.status == "requires_probe"
|
|
assert "MODEL_ARCHITECTURE_EVIDENCE_UNKNOWN" in result.warnings
|
|
|
|
|
|
def test_assessment_becomes_stale_when_hardware_facts_change(session: Session) -> None:
|
|
service, artifact_set, node, _artifacts = setup_runtime(session)
|
|
profile = create_profile(service, artifact_set.id)
|
|
assessment = service.assess(
|
|
CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id)
|
|
)
|
|
accelerator = service.repo.accelerators(node.id)[0]
|
|
accelerator.driver_version = "999.1"
|
|
session.commit()
|
|
refreshed = service.assessment(assessment.id)
|
|
assert refreshed.stale is True
|
|
assert "evidence changed" in str(refreshed.stale_reason)
|
|
|
|
|
|
def test_expired_and_changed_approvals_cannot_authorize_probe(session: Session) -> None:
|
|
service, artifact_set, node, artifacts = setup_runtime(session)
|
|
profile = create_profile(service, artifact_set.id)
|
|
assessment = service.assess(
|
|
CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id)
|
|
)
|
|
expired = service.approve(
|
|
artifact_set.id,
|
|
ExecutionApprovalCreate(
|
|
reason="Time bounded exact artifact review",
|
|
approved_by="operator",
|
|
expires_at=datetime.now(UTC) - timedelta(seconds=1),
|
|
),
|
|
)
|
|
assert expired.stale is True
|
|
with pytest.raises(RegistryConflict, match="valid LAB"):
|
|
service.create_probe(
|
|
RuntimeProbeCreate(
|
|
compatibility_assessment_id=assessment.id,
|
|
execution_approval_id=expired.id,
|
|
)
|
|
)
|
|
current = service.approve(
|
|
artifact_set.id,
|
|
ExecutionApprovalCreate(reason="Second exact artifact review", approved_by="operator"),
|
|
)
|
|
artifacts[0].status = "corrupt"
|
|
session.commit()
|
|
assert service.approval_response(service.repo.approval(current.id)).stale is True # type: ignore[arg-type]
|
|
|
|
|
|
def test_disabled_node_and_wrong_artifact_location_are_blocked(session: Session) -> None:
|
|
service, artifact_set, node, _artifacts = setup_runtime(session)
|
|
node.enabled = False
|
|
session.commit()
|
|
profile = create_profile(service, artifact_set.id)
|
|
disabled = service.assess(
|
|
CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id)
|
|
)
|
|
assert "NODE_NOT_LAB_READY" in disabled.blockers
|
|
node.enabled = True
|
|
other = ComputeNode(key="other", hostname="other", enabled=True, liveness_state="online")
|
|
session.add(other)
|
|
session.flush()
|
|
root = session.query(StorageRoot).one()
|
|
root.compute_node_id = other.id
|
|
session.commit()
|
|
misplaced = service.assess(
|
|
CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id)
|
|
)
|
|
assert "VERIFIED_ARTIFACT_LOCATION_NOT_ON_TARGET_NODE" in misplaced.blockers
|
|
|
|
|
|
def test_expired_worker_lease_is_recovered_after_restart(session: Session) -> None:
|
|
service, artifact_set, node, _artifacts = setup_runtime(session)
|
|
_profile, _assessment, _approval, probe = prepare_probe(service, artifact_set, node)
|
|
first = service.claim_next(node)
|
|
assert first is not None
|
|
stored = service.repo.probe(probe.id)
|
|
assert stored is not None
|
|
stored.lease_expires_at = datetime.now(UTC) - timedelta(seconds=1)
|
|
session.commit()
|
|
second = service.claim_next(node)
|
|
assert second is not None
|
|
assert second.lease_token != first.lease_token
|
|
assert service.probe(probe.id).attempt_count == 2
|
|
|
|
|
|
def test_running_probe_cancellation_is_acknowledged_and_terminal(session: Session) -> None:
|
|
service, artifact_set, node, _artifacts = setup_runtime(session)
|
|
_profile, _assessment, _approval, probe = prepare_probe(service, artifact_set, node)
|
|
lease = service.claim_next(node)
|
|
assert lease is not None
|
|
service.cancel(probe.id)
|
|
control = service.progress(
|
|
probe.id,
|
|
node,
|
|
AgentRuntimeProbeProgress(lease_token=lease.lease_token, status="loading"),
|
|
)
|
|
assert control.cancel_requested is True
|
|
cancelled = service.fail(
|
|
probe.id,
|
|
node,
|
|
AgentRuntimeProbeFailure(
|
|
lease_token=lease.lease_token,
|
|
failure_code="CANCELLED",
|
|
failure_message="operator cancelled",
|
|
),
|
|
)
|
|
assert cancelled.status == "cancelled"
|
|
assert service.candidates() == []
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("health", "inference", "unload", "facts", "code"),
|
|
[
|
|
(
|
|
{"process": "healthy", "runtime": "healthy", "model": "healthy"},
|
|
{"finite": False, "dimension": 0},
|
|
{"reclaimed": True},
|
|
{"offline_local_only": True},
|
|
"INVALID_OUTPUT",
|
|
),
|
|
(
|
|
{"process": "healthy", "runtime": "healthy", "model": "healthy"},
|
|
{"finite": True, "dimension": 1024},
|
|
{"reclaimed": False},
|
|
{"offline_local_only": True},
|
|
"GPU_MEMORY_NOT_RECLAIMED",
|
|
),
|
|
(
|
|
{"process": "healthy", "runtime": "healthy", "model": "healthy"},
|
|
{"finite": True, "dimension": 1024},
|
|
{"reclaimed": True},
|
|
{"offline_local_only": False},
|
|
"OFFLINE_LOAD_VIOLATION",
|
|
),
|
|
],
|
|
)
|
|
def test_completion_failures_are_typed_and_never_create_candidate(
|
|
session: Session,
|
|
health: dict,
|
|
inference: dict,
|
|
unload: dict,
|
|
facts: dict,
|
|
code: str,
|
|
) -> None:
|
|
service, artifact_set, node, _artifacts = setup_runtime(session)
|
|
_profile, _assessment, _approval, probe = prepare_probe(service, artifact_set, node)
|
|
lease = service.claim_next(node)
|
|
assert lease is not None
|
|
result = service.complete(
|
|
probe.id,
|
|
node,
|
|
AgentRuntimeProbeComplete(
|
|
lease_token=lease.lease_token,
|
|
load_result={"status": "passed"},
|
|
health_result=health,
|
|
inference_result=inference,
|
|
unload_result=unload,
|
|
measured_resources={"samples": {}},
|
|
runtime_facts=facts,
|
|
environment_fingerprint="e" * 64,
|
|
),
|
|
)
|
|
assert result.status == "failed"
|
|
assert result.failure_code == code
|
|
assert service.candidates() == []
|