71 lines
2.7 KiB
Python
71 lines
2.7 KiB
Python
"""Test-only seams. Nothing in the production code path may import this module.
|
|
|
|
Startup admits a node only on a capability report produced by a *real* forward
|
|
through the loaded shard (see :mod:`meshnet_node.admission`). Tests run against
|
|
fake or stub backends that cannot perform one, so they pass an explicit validator
|
|
from here instead — the honest statement being "this test asserts capability it
|
|
never proved", which is a thing a test may do and a node may not.
|
|
|
|
`capability_stub` builds the deliberately-wrong reports the fail-closed tests
|
|
need: a failed one, one for another model or shard, one that has aged out.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from typing import Any
|
|
|
|
from .admission import CapabilityContext, CapabilityValidator
|
|
from .capability import STATUS_PASSED, CapabilityReport, build_capability_report
|
|
|
|
|
|
def capability_report_for(
|
|
context: CapabilityContext,
|
|
*,
|
|
status: str = STATUS_PASSED,
|
|
model_id: str | None = None,
|
|
shard_start: int | None = None,
|
|
shard_end: int | None = None,
|
|
recipe_id: str | None = None,
|
|
recipe_version: str | None = None,
|
|
backend_id: str | None = None,
|
|
device: str | None = None,
|
|
validated_at: float | None = None,
|
|
age_seconds: float = 0.0,
|
|
diagnostics: Any = None,
|
|
duration_ms: int = 0,
|
|
) -> CapabilityReport:
|
|
"""A report describing `context`, with any field bent away from the truth."""
|
|
now = time.time() if validated_at is None else validated_at
|
|
return build_capability_report(
|
|
model_id=model_id or context.selection.model_id,
|
|
shard_start=(
|
|
context.selection.shard_start if shard_start is None else shard_start
|
|
),
|
|
shard_end=context.selection.shard_end if shard_end is None else shard_end,
|
|
recipe_id=recipe_id or context.recipe.id,
|
|
recipe_version=recipe_version or context.recipe.version,
|
|
catalogue_version=context.manifest.catalogue_version,
|
|
backend_id=backend_id or context.recipe.backend_id,
|
|
device=device or context.device,
|
|
quantization=context.selection.quantization,
|
|
status=status,
|
|
duration_ms=duration_ms,
|
|
diagnostics=diagnostics,
|
|
validated_at=now - age_seconds,
|
|
)
|
|
|
|
|
|
def assume_capability(context: CapabilityContext) -> CapabilityReport:
|
|
"""Assert the selection works, without proving it. Tests only."""
|
|
return capability_report_for(context)
|
|
|
|
|
|
def capability_stub(**overrides: Any) -> CapabilityValidator:
|
|
"""A validator producing a report that deviates from `context` as named."""
|
|
|
|
def validator(context: CapabilityContext) -> CapabilityReport:
|
|
return capability_report_for(context, **overrides)
|
|
|
|
return validator
|