wehub-resource-sync
134 lines
5.0 KiB
Python
134 lines
5.0 KiB
Python
"""Guard tests for the declarative harness capability model.
|
|
|
|
Capabilities are declared per harness in ``_BUILTIN_CONTRIBUTION.capabilities``.
|
|
These tests assert the declarations stay complete (every built-in harness is
|
|
covered) and that the two *derivable* axes (model_family, subagents) do not
|
|
contradict the code that actually enforces them, so the table cannot silently
|
|
drift.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from omnigent import harness_plugins as hp
|
|
from omnigent.harness_capabilities import (
|
|
AuthModel,
|
|
EffortFamily,
|
|
Elicitation,
|
|
HarnessCapabilities,
|
|
IntegrationMode,
|
|
ModelFamily,
|
|
Resume,
|
|
)
|
|
from omnigent.harness_plugins import (
|
|
HarnessContribution,
|
|
harness_capabilities,
|
|
harness_catalog,
|
|
native_agents,
|
|
valid_harnesses,
|
|
)
|
|
from omnigent.model_override import (
|
|
_ANTIGRAVITY_FAMILY_HARNESSES,
|
|
_CLAUDE_FAMILY_HARNESSES,
|
|
_CODEX_FAMILY_HARNESSES,
|
|
)
|
|
|
|
_NATIVE_MODES = frozenset({IntegrationMode.NATIVE_TUI, IntegrationMode.NATIVE_SERVER})
|
|
|
|
|
|
def test_every_builtin_harness_declares_capabilities() -> None:
|
|
caps = harness_capabilities()
|
|
missing = sorted(valid_harnesses() - set(caps))
|
|
assert not missing, f"harnesses missing a capability declaration: {missing}"
|
|
|
|
|
|
def test_capability_keys_are_valid_harnesses() -> None:
|
|
# No stray capability entry for a non-existent harness id.
|
|
stray = sorted(set(harness_capabilities()) - valid_harnesses())
|
|
assert not stray, f"capability entries for unknown harnesses: {stray}"
|
|
|
|
|
|
def test_native_agents_have_native_integration_mode() -> None:
|
|
caps = harness_capabilities()
|
|
native_harness_ids = {agent.harness for agent in native_agents()}
|
|
for harness in native_harness_ids:
|
|
assert caps[harness].integration_mode in _NATIVE_MODES, harness
|
|
|
|
|
|
def test_model_family_matches_model_override_sets() -> None:
|
|
# model_family is derivable from model_override's family frozensets, so the
|
|
# declaration must not contradict the code that enforces routing.
|
|
for harness, capability in harness_capabilities().items():
|
|
family = capability.model_family
|
|
if harness in _CLAUDE_FAMILY_HARNESSES:
|
|
assert family is ModelFamily.CLAUDE, harness
|
|
elif harness in _CODEX_FAMILY_HARNESSES:
|
|
assert family is ModelFamily.GPT, harness
|
|
elif harness in _ANTIGRAVITY_FAMILY_HARNESSES:
|
|
assert family is ModelFamily.GEMINI, harness
|
|
else:
|
|
assert family is ModelFamily.MULTI, harness
|
|
|
|
|
|
def test_subagents_matches_native_wrapper_label() -> None:
|
|
# subagents is derivable: only native agents with a subagent_wrapper_label
|
|
# can spawn Omnigent native sub-agents.
|
|
subagent_capable = {agent.harness for agent in native_agents() if agent.subagent_wrapper_label}
|
|
for harness, capability in harness_capabilities().items():
|
|
expected = harness in subagent_capable
|
|
assert capability.subagents == expected, harness
|
|
|
|
|
|
def test_native_harnesses_resume_warm() -> None:
|
|
caps = harness_capabilities()
|
|
native_harness_ids = {agent.harness for agent in native_agents()}
|
|
for harness in native_harness_ids:
|
|
assert caps[harness].resume is Resume.WARM_REATTACH, harness
|
|
|
|
|
|
def test_p0_bench_harnesses_declare_interrupt_and_streaming() -> None:
|
|
# The harness bench live-probes these four and declares them SUPPORTED for
|
|
# interrupt + streaming; the capability declaration must match so the bench
|
|
# can derive its expected matrix from capabilities without contradiction.
|
|
caps = harness_capabilities()
|
|
for harness in ("claude-sdk", "codex", "pi", "openai-agents"):
|
|
assert caps[harness].interrupt is True, harness
|
|
assert caps[harness].streaming is True, harness
|
|
|
|
|
|
def test_community_capabilities_cannot_override_builtin() -> None:
|
|
# `capabilities` is part of the collision-key set, so a community plugin that
|
|
# declares capabilities for a built-in harness id is rejected rather than
|
|
# silently overriding the built-in declaration (last-wins in _merge_dict).
|
|
evil = HarnessContribution(
|
|
name="omnigent-evil",
|
|
capabilities={
|
|
"claude-sdk": HarnessCapabilities(
|
|
IntegrationMode.SDK_IN_PROCESS,
|
|
Elicitation.NONE,
|
|
Resume.COLD_ONLY,
|
|
EffortFamily.NONE,
|
|
ModelFamily.MULTI,
|
|
AuthModel.OWN_AUTH,
|
|
subagents=False,
|
|
interrupt=False,
|
|
streaming=False,
|
|
)
|
|
},
|
|
)
|
|
error = hp._validate_community_contribution(
|
|
evil, entry_point_name="evil", existing=(hp._BUILTIN_CONTRIBUTION,)
|
|
)
|
|
assert error is not None
|
|
assert "claude-sdk" in error
|
|
|
|
|
|
def test_catalog_rows_include_capabilities() -> None:
|
|
rows = harness_catalog()
|
|
caps = harness_capabilities()
|
|
for row in rows:
|
|
if row["id"] in caps:
|
|
assert "capabilities" in row, row["id"]
|
|
# JSON-serializable: values are str/bool, not enums.
|
|
for value in row["capabilities"].values():
|
|
assert isinstance(value, (str, bool))
|