File size: 5,796 Bytes
2aa8b3a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 | """Inert fixture tools for exercising the broker without shipping a real one.
The production registry is empty (see :func:`distinct_tools.default_registry`).
That is a deliberate product decision, not an oversight, and it would normally
take the framework's test coverage down with it: a policy boundary with nothing
behind it is hard to prove anything about.
These fixtures exist so the interesting machinery — the per-job broker, the
allowlist, quotas, timeouts, byte caps, host permissions and the fail-closed
default — stays genuinely exercised. They perform **no** I/O of any kind:
* ``demo.echo`` is a pure function of its arguments.
* ``demo.hostcheck`` calls :meth:`ToolContext.require_url` and returns the
validated hostname. It never opens a socket, and the host it validates
against is under ``.invalid``, a reserved TLD guaranteed never to resolve.
Nothing here is registered automatically. A caller must import this module and
call ``register_*`` explicitly, which is why an accidental import cannot put a
tool into a running agent's registry.
"""
from __future__ import annotations
import time
from collections.abc import Mapping
from .core import Registry, ToolContext, ToolInputError, ToolSpec
#: A reserved, non-resolvable TLD (RFC 6761). Used so host-permission tests
#: can never accidentally reach a real service.
FIXTURE_HOST = "tools.example.invalid"
FIXTURE_URL = f"https://{FIXTURE_HOST}/probe"
ECHO_SPEC = ToolSpec(
tool_id="demo.echo",
version="1",
description="Return the supplied text unchanged. A local, inert test fixture.",
required_hosts=frozenset(),
input_schema={
"type": "object",
"additionalProperties": False,
"required": ["text"],
"properties": {
"text": {"type": "string", "minLength": 1, "maxLength": 4_000},
"repeat": {"type": "integer", "minimum": 1, "maximum": 8, "default": 1},
"delay_seconds": {"type": "number", "minimum": 0, "maximum": 60},
},
},
)
HOSTCHECK_SPEC = ToolSpec(
tool_id="demo.hostcheck",
version="1",
description="Validate a fixed URL against the job host allowlist without contacting it.",
required_hosts=frozenset((FIXTURE_HOST,)),
input_schema={
"type": "object",
"additionalProperties": False,
"properties": {},
},
)
def echo_handler(arguments: Mapping[str, object], context: ToolContext) -> dict[str, object]:
"""Return the supplied text. Pure; touches nothing outside its arguments."""
text = arguments.get("text")
if not isinstance(text, str) or not text or len(text) > 4_000:
raise ToolInputError("text must be a non-empty string of at most 4,000 characters")
repeat = arguments.get("repeat", 1)
if isinstance(repeat, bool) or not isinstance(repeat, int) or not 1 <= repeat <= 8:
raise ToolInputError("repeat must be an integer between 1 and 8")
delay = arguments.get("delay_seconds", 0)
if isinstance(delay, bool) or not isinstance(delay, int | float) or not 0 <= delay <= 60:
raise ToolInputError("delay_seconds must be a number between 0 and 60")
if delay:
# Only used to prove the broker's deadline actually fires.
time.sleep(float(delay))
echoed = text * repeat
return {"text": echoed, "length": len(echoed), "repeat": repeat}
def hostcheck_handler(arguments: Mapping[str, object], context: ToolContext) -> dict[str, object]:
"""Prove the host allowlist is enforced, without performing a request."""
host = context.require_url(FIXTURE_URL)
return {"host": host, "contacted": False}
def register_echo(registry: Registry) -> None:
"""Explicitly add the inert ``demo.echo`` fixture to ``registry``."""
registry.register(ECHO_SPEC, echo_handler)
def register_hostcheck(registry: Registry) -> None:
"""Explicitly add the inert ``demo.hostcheck`` fixture to ``registry``."""
registry.register(HOSTCHECK_SPEC, hostcheck_handler)
def fixture_registry(*, hostcheck: bool = False) -> Registry:
"""Build a registry containing only the inert fixtures a test asked for."""
registry = Registry()
register_echo(registry)
if hostcheck:
register_hostcheck(registry)
return registry
def fixture_operator_policy(*, hostcheck: bool = False, egress: bool = False):
"""The approval that matches :func:`fixture_registry`, for the same test.
A fixture registry stands for "the operator approved these", because in a
real worker `default_registry()` only ever contains what the policy
permitted. `distinct_agent.tools.load_tool_broker` now hands the operator's
policy to the broker, which is where `ToolBroker` documents the check as
happening, so a test that substitutes one of the two must substitute both
or it is testing a worker whose registry and policy disagree, which is a
state a real worker cannot be in.
``egress`` is separate and defaults to off, exactly as it is for a real
operator: approving a tool and clearing it to reach the network are two
decisions, and a fixture that collapsed them would let a test pass that
the product would refuse.
"""
from .approval import OperatorPolicy
approved = {ECHO_SPEC.ref}
if hostcheck:
approved.add(HOSTCHECK_SPEC.ref)
cleared = set()
if egress and hostcheck:
cleared.add(HOSTCHECK_SPEC.ref)
return OperatorPolicy(
approved=frozenset(approved),
egress_approved=frozenset(cleared),
source="test fixture",
)
__all__ = [
"ECHO_SPEC",
"FIXTURE_HOST",
"FIXTURE_URL",
"HOSTCHECK_SPEC",
"echo_handler",
"fixture_operator_policy",
"fixture_registry",
"hostcheck_handler",
"register_echo",
"register_hostcheck",
]
|