distinct_tools/testing.py

"""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",
]