contract

contract

The consumer half of the matrix capability contract.

The engine exports TEROK_EXPECT (a comma-separated capability list from matrix.yml) inside every matrix test container — see inner. A consuming repo's integration conftest owns the other half: a probe per capability it knows how to verify, handed to check_capability_contract at session start. Inside the matrix the harness built the image, so a declared-but-missing capability means the slot is broken and must fail up front — not dissolve into skips that read as green.

Typical conftest wiring::

_CAPABILITY_PROBES = {"podman": lambda: binary_on_path("podman"), ...}

def pytest_sessionstart(session):
    if broken := check_capability_contract(_CAPABILITY_PROBES):
        pytest.exit(broken, returncode=3)

The probe maps rightly differ per repo (shield probes its own hook chain, clearance only D-Bus); the protocol — env var, separator, unknown-name rejection — lives here so it cannot drift apart again.

check_capability_contract(probes)

Verify TEROK_EXPECT against probes; return the failure, if any.

Parameters:

Name	Type	Description	Default
probes	Mapping[str, Callable[[], bool]]	Capability name to presence probe, the repo-specific half of the contract.	required

Returns:

Type	Description
str | None	A human-readable failure message when the contract names unknown
str | None	capabilities or a declared capability is missing; None when
str | None	the contract holds (or none is declared — dev-machine runs).

Source code in src/terok_util/matrix/contract.py

def check_capability_contract(probes: Mapping[str, Callable[[], bool]]) -> str | None:
    """Verify ``TEROK_EXPECT`` against *probes*; return the failure, if any.

    Args:
        probes: Capability name to presence probe, the repo-specific half
            of the contract.

    Returns:
        A human-readable failure message when the contract names unknown
        capabilities or a declared capability is missing; ``None`` when
        the contract holds (or none is declared — dev-machine runs).
    """
    expected = {c for c in os.environ.get(EXPECT_ENV, "").split(",") if c}
    if not expected:
        return None
    unknown = expected - probes.keys()
    if unknown:
        return f"{EXPECT_ENV} names unknown capabilities: {sorted(unknown)}"
    missing = sorted(cap for cap in expected if not probes[cap]())
    if missing:
        return "matrix capability contract broken - expected but missing: " + ", ".join(missing)
    return None

binary_on_path(name)

Whether name resolves on the sbin-extended PATH.

Source code in src/terok_util/matrix/contract.py

def binary_on_path(name: str) -> bool:
    """Whether *name* resolves on the sbin-extended ``PATH``."""
    search = os.pathsep.join([os.environ.get("PATH", ""), *_SBIN_DIRS])
    return shutil.which(name, path=search) is not None

tcp_reachable(ip, port, timeout=5.0)

Whether a TCP connection to ip:port succeeds within timeout.

Source code in src/terok_util/matrix/contract.py

def tcp_reachable(ip: str, port: int, timeout: float = 5.0) -> bool:
    """Whether a TCP connection to ``ip:port`` succeeds within *timeout*."""
    try:
        with socket.create_connection((ip, port), timeout=timeout):
            return True
    except OSError:
        return False