mortal/strix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: nuke tests/ and the entire test toolchain

Browse Source 
The test suite was carrying migration scars and a long tail of
low-density assertions over SDK-derived behavior. Drop it wholesale.

- Delete ``tests/`` (42 files, ~4900 LoC).
- Drop ``pytest`` / ``pytest-asyncio`` / ``pytest-cov`` /
  ``pytest-mock`` from the dev dependency group; ``uv sync``
  uninstalls the matching wheels.
- Strip the pytest + coverage config blocks, the
  ``flake8-pytest-style`` ruff selector, the ``tests/**`` per-file
  ignores, the ``[tool.mypy.overrides] tests.*`` block, and the
  ``"tests"`` entry from bandit's ``exclude_dirs``.
- Drop the ``test`` / ``test-cov`` Makefile targets; ``dev`` no
  longer depends on tests.
- Strip the ``# Testing`` block from ``.gitignore`` (``.coverage``,
  ``.pytest_cache/``, ``htmlcov/``, ``coverage.xml``, ``nosetests.xml``,
  ``.tox/``, ``.hypothesis/``).

ruff (27) and mypy (82) baselines unchanged.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
0xallam
2026-04-25 13:01:20 -07:00
parent 49c38de3b2
commit a6d578c4a8
46 changed files with 3 additions and 5186 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/__init__.py
-1
View File
@@ -1 +0,0 @@
# Strix Test Suite
tests/agents/__init__.py
-1
View File
@@ -1 +0,0 @@
"""Tests for strix.agents module."""
tests/agents/test_factory.py
-200
View File
@@ -1,200 +0,0 @@
"""Phase 5 tests for the SDK agent factory + prompt renderer.
These two modules are the keystone wiring between Phases 2-4 and an
actual ``Runner.run`` invocation. The tests verify:
- The prompt renderer reuses the existing Jinja template (parity with
legacy LLM._load_system_prompt) and degrades gracefully when the
template isn't available.
- ``build_strix_agent(is_root=True)`` carries ``finish_scan`` and
stops on it; child agents carry ``agent_finish`` and stop on it.
- ``make_child_factory`` snapshots scan-level config into a closure
so each spawned child inherits the right scan_mode / is_whitebox /
prompt context without create_agent having to re-derive it.
"""
from __future__ import annotations
from typing import Any
from unittest.mock import patch
from agents import Agent
from agents.tool import FunctionTool
from strix.agents.factory import build_strix_agent, make_child_factory
from strix.agents.prompt import _resolve_skills, render_system_prompt
# --- prompt renderer ----------------------------------------------------
def test_resolve_skills_deduplicates_and_orders() -> None:
out = _resolve_skills(
requested=["recon", "xss", "recon"],
scan_mode="deep",
is_whitebox=False,
)
assert out == ["recon", "xss", "scan_modes/deep"]
def test_resolve_skills_adds_whitebox_pair() -> None:
out = _resolve_skills(requested=None, scan_mode="fast", is_whitebox=True)
# The whitebox pair sits at the tail; scan_modes goes in the middle
# because callers can append more skills after it via the requested arg.
assert out == [
"scan_modes/fast",
"coordination/source_aware_whitebox",
"custom/source_aware_sast",
]
def test_render_system_prompt_returns_string() -> None:
"""Smoke: the StrixAgent template is on disk and renders to non-empty."""
out = render_system_prompt(skills=[], scan_mode="deep")
assert isinstance(out, str)
# The first line of the template starts with 'You are Strix'.
assert out.startswith("You are Strix")
def test_render_system_prompt_swallows_template_errors() -> None:
"""If the template path can't be resolved, return an empty string
(not raise) — agent construction must never blow up on prompt load."""
with patch(
"strix.agents.prompt.get_strix_resource_path",
side_effect=RuntimeError("missing"),
):
out = render_system_prompt(skills=[])
assert out == ""
# --- factory: shape + tools --------------------------------------------
def test_root_agent_carries_finish_scan_and_stops_there() -> None:
agent = build_strix_agent(name="strix", is_root=True)
assert isinstance(agent, Agent)
tool_names = {t.name for t in agent.tools if isinstance(t, FunctionTool)}
assert "finish_scan" in tool_names
assert "agent_finish" not in tool_names
behavior = agent.tool_use_behavior
# StopAtTools is a TypedDict at runtime → behavior is a dict.
assert isinstance(behavior, dict)
assert behavior["stop_at_tool_names"] == ["finish_scan"]
def test_child_agent_carries_agent_finish_and_stops_there() -> None:
agent = build_strix_agent(name="recon-bot", is_root=False)
tool_names = {t.name for t in agent.tools if isinstance(t, FunctionTool)}
assert "agent_finish" in tool_names
assert "finish_scan" not in tool_names
behavior = agent.tool_use_behavior
assert isinstance(behavior, dict)
assert behavior["stop_at_tool_names"] == ["agent_finish"]
def test_root_and_child_share_base_tool_set() -> None:
"""The base tool set (think/todo/notes/file_edit/web_search/etc) is
identical between root and child — only the terminator differs."""
root = build_strix_agent(is_root=True)
child = build_strix_agent(is_root=False)
root_names = {t.name for t in root.tools if isinstance(t, FunctionTool)}
child_names = {t.name for t in child.tools if isinstance(t, FunctionTool)}
# Drop the terminators and compare.
assert root_names - {"finish_scan"} == child_names - {"agent_finish"}
def test_agent_includes_graph_and_sandbox_tools() -> None:
"""The graph + sandbox tool families are required for parity with
legacy. Spot-check the ones most likely to be forgotten in a refactor."""
agent = build_strix_agent(is_root=True)
names = {t.name for t in agent.tools if isinstance(t, FunctionTool)}
expected = {
"think",
"create_todo",
"create_note",
"web_search",
"str_replace_editor",
"create_vulnerability_report",
"browser_action",
"terminal_execute",
"python_action",
"view_agent_graph",
"agent_status",
"send_message_to_agent",
"wait_for_message",
"create_agent",
}
missing = expected - names
assert not missing, f"missing tools: {missing}"
def test_agent_does_not_include_caido_tools() -> None:
"""Caido tools come from CaidoCapability.tools(); the agent doesn't
declare them directly to avoid double-registration when the SDK
runtime merges capability tools."""
agent = build_strix_agent(is_root=True)
names = {t.name for t in agent.tools if isinstance(t, FunctionTool)}
caido = {
"list_requests",
"view_request",
"send_request",
"repeat_request",
"scope_rules",
"list_sitemap",
"view_sitemap_entry",
}
overlap = names & caido
assert overlap == set(), f"unexpected Caido tools in agent.tools: {overlap}"
def test_agent_uses_run_config_model() -> None:
"""``model=None`` so the RunConfig drives the model alias through
MultiProvider rather than an SDK default like gpt-4.1."""
agent = build_strix_agent(is_root=True)
assert agent.model is None
def test_agent_instructions_contain_rendered_prompt() -> None:
"""The factory must wire the rendered prompt into ``instructions``."""
agent = build_strix_agent(is_root=True, scan_mode="deep")
assert isinstance(agent.instructions, str)
assert agent.instructions.startswith("You are Strix")
# --- child factory ------------------------------------------------------
def test_make_child_factory_returns_callable_that_builds_child() -> None:
factory = make_child_factory(scan_mode="deep", is_whitebox=False)
assert callable(factory)
child = factory(name="sub-1", skills=["recon"])
assert isinstance(child, Agent)
assert child.name == "sub-1"
behavior = child.tool_use_behavior
assert isinstance(behavior, dict)
assert behavior["stop_at_tool_names"] == ["agent_finish"]
def test_make_child_factory_passes_scan_level_config() -> None:
"""Verify scan_mode + is_whitebox flow into the rendered prompt
via the closure rather than the create_agent call site."""
captured: dict[str, Any] = {}
def fake_render(**kwargs: Any) -> str:
captured.update(kwargs)
return "stub-prompt"
factory = make_child_factory(
scan_mode="fast",
is_whitebox=True,
interactive=True,
system_prompt_context={"scope_source": "test"},
)
with patch("strix.agents.factory.render_system_prompt", side_effect=fake_render):
factory(name="child", skills=["xss"])
assert captured["scan_mode"] == "fast"
assert captured["is_whitebox"] is True
assert captured["interactive"] is True
assert captured["system_prompt_context"] == {"scope_source": "test"}
assert captured["skills"] == ["xss"]
tests/config/__init__.py
-1
View File
@@ -1 +0,0 @@
"""Tests for strix.config module."""
tests/config/test_config_override.py
-31
View File
@@ -1,31 +0,0 @@
import json
import os
from strix.config.config import Config, apply_saved_config
def test_apply_config_override_clears_default_only_vars(monkeypatch, tmp_path) -> None:
from strix.interface.main import apply_config_override
default_cfg = tmp_path / "cli-config.json"
default_cfg.write_text(
json.dumps({"env": {"LLM_API_BASE": "https://default.api", "STRIX_LLM": "default-model"}}),
encoding="utf-8",
)
custom_cfg = tmp_path / "custom.json"
custom_cfg.write_text(json.dumps({"env": {"STRIX_LLM": "custom-model"}}), encoding="utf-8")
monkeypatch.setattr(Config, "_config_file_override", None)
monkeypatch.setattr(Config, "_applied_from_default", {})
monkeypatch.setattr(Config, "config_dir", classmethod(lambda cls: tmp_path))
for var_name in Config._llm_env_vars():
monkeypatch.delenv(var_name, raising=False)
apply_saved_config()
assert os.environ.get("LLM_API_BASE") == "https://default.api"
apply_config_override(str(custom_cfg))
assert os.environ.get("STRIX_LLM") == "custom-model"
assert "LLM_API_BASE" not in os.environ
tests/config/test_config_telemetry.py
-55
View File
@@ -1,55 +0,0 @@
import json
from strix.config.config import Config
def test_traceloop_vars_are_tracked() -> None:
tracked = Config.tracked_vars()
assert "STRIX_OTEL_TELEMETRY" in tracked
assert "STRIX_POSTHOG_TELEMETRY" in tracked
assert "TRACELOOP_BASE_URL" in tracked
assert "TRACELOOP_API_KEY" in tracked
assert "TRACELOOP_HEADERS" in tracked
def test_apply_saved_uses_saved_traceloop_vars(monkeypatch, tmp_path) -> None:
config_path = tmp_path / "cli-config.json"
config_path.write_text(
json.dumps(
{
"env": {
"TRACELOOP_BASE_URL": "https://otel.example.com",
"TRACELOOP_API_KEY": "api-key",
"TRACELOOP_HEADERS": "x-test=value",
}
}
),
encoding="utf-8",
)
monkeypatch.setattr(Config, "_config_file_override", config_path)
monkeypatch.delenv("TRACELOOP_BASE_URL", raising=False)
monkeypatch.delenv("TRACELOOP_API_KEY", raising=False)
monkeypatch.delenv("TRACELOOP_HEADERS", raising=False)
applied = Config.apply_saved()
assert applied["TRACELOOP_BASE_URL"] == "https://otel.example.com"
assert applied["TRACELOOP_API_KEY"] == "api-key"
assert applied["TRACELOOP_HEADERS"] == "x-test=value"
def test_apply_saved_respects_existing_env_traceloop_vars(monkeypatch, tmp_path) -> None:
config_path = tmp_path / "cli-config.json"
config_path.write_text(
json.dumps({"env": {"TRACELOOP_BASE_URL": "https://otel.example.com"}}),
encoding="utf-8",
)
monkeypatch.setattr(Config, "_config_file_override", config_path)
monkeypatch.setenv("TRACELOOP_BASE_URL", "https://env.example.com")
applied = Config.apply_saved(force=False)
assert "TRACELOOP_BASE_URL" not in applied
tests/conftest.py
-1
View File
@@ -1 +0,0 @@
"""Pytest configuration and shared fixtures for Strix tests."""
tests/interface/__init__.py
-1
View File
@@ -1 +0,0 @@
"""Tests for strix.interface module."""
tests/interface/test_diff_scope.py
-153
View File
@@ -1,153 +0,0 @@
import importlib.util
from pathlib import Path
import pytest
def _load_utils_module():
module_path = Path(__file__).resolve().parents[2] / "strix" / "interface" / "utils.py"
spec = importlib.util.spec_from_file_location("strix_interface_utils_test", module_path)
if spec is None or spec.loader is None:
raise RuntimeError("Failed to load strix.interface.utils for tests")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
utils = _load_utils_module()
def test_parse_name_status_uses_rename_destination_path() -> None:
raw = (
b"R100\x00old/path.py\x00new/path.py\x00"
b"R75\x00legacy/module.py\x00modern/module.py\x00"
b"M\x00src/app.py\x00"
b"A\x00src/new_file.py\x00"
b"D\x00src/deleted.py\x00"
)
entries = utils._parse_name_status_z(raw)
classified = utils._classify_diff_entries(entries)
assert "new/path.py" in classified["analyzable_files"]
assert "old/path.py" not in classified["analyzable_files"]
assert "modern/module.py" in classified["analyzable_files"]
assert classified["renamed_files"][0]["old_path"] == "old/path.py"
assert classified["renamed_files"][0]["new_path"] == "new/path.py"
assert "src/deleted.py" in classified["deleted_files"]
assert "src/deleted.py" not in classified["analyzable_files"]
def test_build_diff_scope_instruction_includes_added_modified_and_deleted_guidance() -> None:
scope = utils.RepoDiffScope(
source_path="/tmp/repo",
workspace_subdir="repo",
base_ref="refs/remotes/origin/main",
merge_base="abc123",
added_files=["src/added.py"],
modified_files=["src/changed.py"],
renamed_files=[{"old_path": "src/old.py", "new_path": "src/new.py", "similarity": 90}],
deleted_files=["src/deleted.py"],
analyzable_files=["src/added.py", "src/changed.py", "src/new.py"],
)
instruction = utils.build_diff_scope_instruction([scope])
assert "For Added files, review the entire file content." in instruction
assert "For Modified files, focus primarily on the changed areas." in instruction
assert "Note: These files were deleted" in instruction
assert "src/deleted.py" in instruction
assert "src/old.py -> src/new.py" in instruction
def test_resolve_base_ref_prefers_github_base_ref(monkeypatch) -> None:
calls: list[str] = []
def fake_ref_exists(_repo_path: Path, ref: str) -> bool:
calls.append(ref)
return ref == "refs/remotes/origin/release-2026"
monkeypatch.setattr(utils, "_git_ref_exists", fake_ref_exists)
monkeypatch.setattr(utils, "_extract_github_base_sha", lambda _env: None)
monkeypatch.setattr(utils, "_resolve_origin_head_ref", lambda _repo_path: None)
base_ref = utils._resolve_base_ref(
Path("/tmp/repo"),
diff_base=None,
env={"GITHUB_BASE_REF": "release-2026"},
)
assert base_ref == "refs/remotes/origin/release-2026"
assert calls[0] == "refs/remotes/origin/release-2026"
def test_resolve_base_ref_falls_back_to_remote_main(monkeypatch) -> None:
calls: list[str] = []
def fake_ref_exists(_repo_path: Path, ref: str) -> bool:
calls.append(ref)
return ref == "refs/remotes/origin/main"
monkeypatch.setattr(utils, "_git_ref_exists", fake_ref_exists)
monkeypatch.setattr(utils, "_extract_github_base_sha", lambda _env: None)
monkeypatch.setattr(utils, "_resolve_origin_head_ref", lambda _repo_path: None)
base_ref = utils._resolve_base_ref(Path("/tmp/repo"), diff_base=None, env={})
assert base_ref == "refs/remotes/origin/main"
assert "refs/remotes/origin/main" in calls
assert "origin/main" not in calls
def test_resolve_diff_scope_context_auto_degrades_when_repo_scope_resolution_fails(
monkeypatch,
) -> None:
source = {"source_path": "/tmp/repo", "workspace_subdir": "repo"}
monkeypatch.setattr(utils, "_should_activate_auto_scope", lambda *_args, **_kwargs: True)
monkeypatch.setattr(utils, "_is_git_repo", lambda _repo_path: True)
monkeypatch.setattr(
utils,
"_resolve_repo_diff_scope",
lambda *_args, **_kwargs: (_ for _ in ()).throw(ValueError("shallow history")),
)
result = utils.resolve_diff_scope_context(
local_sources=[source],
scope_mode="auto",
diff_base=None,
non_interactive=True,
env={},
)
assert result.active is False
assert result.mode == "auto"
assert result.metadata["active"] is False
assert result.metadata["mode"] == "auto"
assert "skipped_diff_scope_sources" in result.metadata
assert result.metadata["skipped_diff_scope_sources"] == [
"/tmp/repo (diff-scope skipped: shallow history)"
]
def test_resolve_diff_scope_context_diff_mode_still_raises_on_repo_scope_resolution_failure(
monkeypatch,
) -> None:
source = {"source_path": "/tmp/repo", "workspace_subdir": "repo"}
monkeypatch.setattr(utils, "_is_git_repo", lambda _repo_path: True)
monkeypatch.setattr(
utils,
"_resolve_repo_diff_scope",
lambda *_args, **_kwargs: (_ for _ in ()).throw(ValueError("shallow history")),
)
with pytest.raises(ValueError, match="shallow history"):
utils.resolve_diff_scope_context(
local_sources=[source],
scope_mode="diff",
diff_base=None,
non_interactive=True,
env={},
)
tests/llm/__init__.py
-1
View File
@@ -1 +0,0 @@
"""Tests for strix.llm module."""
tests/llm/test_anthropic_cache_wrapper.py
-74
View File
@@ -1,74 +0,0 @@
"""Smoke tests for AnthropicCachingLitellmModel."""
from __future__ import annotations
from strix.llm.anthropic_cache_wrapper import AnthropicCachingLitellmModel
def _make(model: str) -> AnthropicCachingLitellmModel:
# ``LitellmModel.__init__`` only validates that model is a string; we
# don't need a real API key for in-memory ``_patch`` testing.
return AnthropicCachingLitellmModel(model=model, api_key="test-key")
def test_is_anthropic_detects_anthropic_prefix() -> None:
m = _make("anthropic/claude-3-5-sonnet")
assert m._is_anthropic() is True
def test_is_anthropic_detects_claude_substring() -> None:
m = _make("openrouter/anthropic-claude-haiku")
assert m._is_anthropic() is True
def test_is_anthropic_false_for_openai() -> None:
m = _make("openai/gpt-4o")
assert m._is_anthropic() is False
def test_is_anthropic_false_for_gemini() -> None:
m = _make("gemini/gemini-1.5-pro")
assert m._is_anthropic() is False
def test_patch_anthropic_adds_cache_control_to_system() -> None:
m = _make("anthropic/claude-3-5-sonnet")
items: list = [
{"role": "system", "content": "You are a helpful agent."},
{"role": "user", "content": "hi"},
]
out = m._patch(items)
assert out[0]["role"] == "system"
content = out[0]["content"]
assert isinstance(content, list)
assert content[0]["type"] == "text"
assert content[0]["text"] == "You are a helpful agent."
assert content[0]["cache_control"] == {"type": "ephemeral"}
# Second item passes through unchanged.
assert out[1] == {"role": "user", "content": "hi"}
def test_patch_non_anthropic_passes_through() -> None:
m = _make("openai/gpt-4o")
items: list = [
{"role": "system", "content": "You are a helpful agent."},
{"role": "user", "content": "hi"},
]
assert m._patch(items) is items # exact same list reference, no copy
def test_patch_skips_non_string_system_content() -> None:
"""If system content is already structured (e.g., previously patched),
don't re-wrap — pass through unchanged."""
m = _make("anthropic/claude-3-5-sonnet")
items: list = [
{"role": "system", "content": [{"type": "text", "text": "x"}]},
{"role": "user", "content": "hi"},
]
out = m._patch(items)
assert out[0]["content"] == [{"type": "text", "text": "x"}]
def test_patch_handles_empty_list() -> None:
m = _make("anthropic/claude-3-5-sonnet")
assert m._patch([]) == []
tests/llm/test_multi_provider_setup.py
-42
View File
@@ -1,42 +0,0 @@
"""Smoke tests for the multi-provider setup."""
from __future__ import annotations
import pytest
from agents.exceptions import UserError
from strix.llm.anthropic_cache_wrapper import AnthropicCachingLitellmModel
from strix.llm.multi_provider_setup import (
_AnthropicCachingProvider,
build_multi_provider,
)
def test_anthropic_provider_wraps_in_caching_model() -> None:
provider = _AnthropicCachingProvider()
model = provider.get_model("claude-sonnet-4-6")
assert isinstance(model, AnthropicCachingLitellmModel)
# The provider re-prefixes with ``anthropic/`` so litellm routes correctly.
assert model.model == "anthropic/claude-sonnet-4-6"
assert model._is_anthropic() is True
def test_anthropic_provider_preserves_existing_anthropic_prefix() -> None:
"""If the alias already carries ``anthropic/``, don't double-prefix."""
provider = _AnthropicCachingProvider()
model = provider.get_model("anthropic/claude-3-5-sonnet-20241022")
assert model.model == "anthropic/claude-3-5-sonnet-20241022"
def test_anthropic_provider_empty_name_raises() -> None:
provider = _AnthropicCachingProvider()
with pytest.raises(UserError, match="non-empty"):
provider.get_model(None)
def test_build_multi_provider_routes_anthropic_through_caching_wrapper() -> None:
"""The configured MultiProvider should hit our caching wrapper for the
``anthropic/`` prefix."""
mp = build_multi_provider()
model = mp.get_model("anthropic/claude-sonnet-4-6")
assert isinstance(model, AnthropicCachingLitellmModel)
tests/orchestration/__init__.py
View File
tests/orchestration/test_bus.py
-195
View File
@@ -1,195 +0,0 @@
"""Phase 0 smoke tests for AgentMessageBus."""
from __future__ import annotations
import asyncio
import pytest
from strix.orchestration.bus import AgentMessageBus
@pytest.fixture
def bus() -> AgentMessageBus:
return AgentMessageBus()
@pytest.mark.asyncio
async def test_register_records_agent(bus: AgentMessageBus) -> None:
await bus.register("a1", "alpha", parent_id=None)
assert bus.statuses["a1"] == "running"
assert bus.parent_of["a1"] is None
assert bus.names["a1"] == "alpha"
assert bus.inboxes["a1"] == []
assert bus.stats_live["a1"]["calls"] == 0
@pytest.mark.asyncio
async def test_send_and_drain_fifo(bus: AgentMessageBus) -> None:
await bus.register("a1", "alpha", parent_id=None)
await bus.send("a1", {"from": "b", "content": "first"})
await bus.send("a1", {"from": "c", "content": "second"})
drained = await bus.drain("a1")
assert [m["content"] for m in drained] == ["first", "second"]
assert await bus.drain("a1") == []
@pytest.mark.asyncio
async def test_send_to_unknown_agent_is_dropped(bus: AgentMessageBus) -> None:
await bus.send("ghost", {"from": "user", "content": "x"})
assert "ghost" not in bus.inboxes
@pytest.mark.asyncio
async def test_finalize_clears_inbox_parent_name(bus: AgentMessageBus) -> None:
"""C13 (AUDIT_R3): finalize cleans up routing state to avoid orphan messages."""
await bus.register("a1", "alpha", parent_id=None)
await bus.register("a2", "beta", parent_id="a1")
await bus.send("a1", {"from": "a2", "content": "hi"})
await bus.finalize("a1", "completed")
# Inbox / parent / name removed so siblings can't accidentally re-fill.
assert "a1" not in bus.inboxes
assert "a1" not in bus.parent_of
assert "a1" not in bus.names
# Status remains for diagnostics.
assert bus.statuses["a1"] == "completed"
# Messages sent to a finalized agent are dropped silently.
await bus.send("a1", {"from": "a2", "content": "ignored"})
assert "a1" not in bus.inboxes
@pytest.mark.asyncio
async def test_record_usage_aggregates(bus: AgentMessageBus) -> None:
await bus.register("a1", "alpha", parent_id=None)
class _Details:
cached_tokens = 10
class _Usage:
input_tokens = 100
output_tokens = 50
input_tokens_details = _Details()
await bus.record_usage("a1", _Usage())
await bus.record_usage("a1", _Usage())
stats = bus.stats_live["a1"]
assert stats["in"] == 200
assert stats["out"] == 100
assert stats["cached"] == 20
assert stats["calls"] == 2
@pytest.mark.asyncio
async def test_record_usage_handles_none(bus: AgentMessageBus) -> None:
await bus.register("a1", "alpha", parent_id=None)
await bus.record_usage("a1", None)
assert bus.stats_live["a1"]["calls"] == 0
@pytest.mark.asyncio
async def test_total_stats_snapshot(bus: AgentMessageBus) -> None:
"""C12 (AUDIT_R2): total_stats acquires the lock for a consistent snapshot."""
await bus.register("a1", "alpha", parent_id=None)
await bus.register("a2", "beta", parent_id="a1")
class _Details:
cached_tokens = 5
class _Usage:
input_tokens = 10
output_tokens = 20
input_tokens_details = _Details()
await bus.record_usage("a1", _Usage())
await bus.record_usage("a2", _Usage())
await bus.finalize("a2", "completed")
totals = await bus.total_stats()
assert totals["in"] == 20
assert totals["out"] == 40
assert totals["cached"] == 10
assert totals["calls"] == 2
@pytest.mark.asyncio
async def test_concurrent_send_drain_no_lost_messages() -> None:
bus = AgentMessageBus()
await bus.register("a1", "alpha", parent_id=None)
async def producer(start: int, count: int) -> None:
for i in range(count):
await bus.send("a1", {"from": "p", "content": str(start + i)})
# 50 producers x 20 messages = 1000 messages; drain in 1 reader.
producers = [asyncio.create_task(producer(i * 20, 20)) for i in range(50)]
await asyncio.gather(*producers)
drained = await bus.drain("a1")
assert len(drained) == 1000
@pytest.mark.asyncio
async def test_cancel_descendants_cancels_whole_tree() -> None:
"""C9 (AUDIT_R2): cancel_descendants cancels every transitive child."""
bus = AgentMessageBus()
await bus.register("root", "root", parent_id=None)
await bus.register("child1", "c1", parent_id="root")
await bus.register("grandchild1", "g1", parent_id="child1")
await bus.register("child2", "c2", parent_id="root")
pending = asyncio.get_event_loop().create_future()
async def fake_run() -> None:
await pending # block until cancelled
for aid in ("root", "child1", "grandchild1", "child2"):
bus.tasks[aid] = asyncio.create_task(fake_run())
await bus.cancel_descendants("root")
for aid in ("root", "child1", "grandchild1", "child2"):
assert bus.tasks[aid].cancelled() or bus.tasks[aid].done()
@pytest.mark.asyncio
async def test_cancel_descendants_triggers_leaves_before_root() -> None:
"""C9: explicit ordering check — leaves' .cancel() called before root's."""
bus = AgentMessageBus()
await bus.register("root", "root", parent_id=None)
await bus.register("child1", "c1", parent_id="root")
await bus.register("grandchild1", "g1", parent_id="child1")
cancel_call_order: list[str] = []
pending = asyncio.get_event_loop().create_future()
class _RecordingTask:
"""Wrap a real Task; record the moment .cancel() is invoked."""
def __init__(self, name: str, task: asyncio.Task) -> None:
self._name = name
self._task = task
def done(self) -> bool:
return self._task.done()
def cancelled(self) -> bool:
return self._task.cancelled()
def cancel(self, *args: object, **kwargs: object) -> bool:
cancel_call_order.append(self._name)
return self._task.cancel()
def __await__(self):
return self._task.__await__()
async def fake_run() -> None:
await pending
for aid in ("root", "child1", "grandchild1"):
real = asyncio.create_task(fake_run())
bus.tasks[aid] = _RecordingTask(aid, real) # type: ignore[assignment]
await bus.cancel_descendants("root")
# grandchild and child must have .cancel() called before root.
assert cancel_call_order.index("grandchild1") < cancel_call_order.index("root")
assert cancel_call_order.index("child1") < cancel_call_order.index("root")
tests/orchestration/test_filter.py
-95
View File
@@ -1,95 +0,0 @@
"""Phase 0 smoke tests for inject_messages_filter."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
import pytest
from agents.run_config import CallModelData, ModelInputData
from strix.orchestration.bus import AgentMessageBus
from strix.orchestration.filter import inject_messages_filter
@dataclass
class _FakeAgent:
name: str = "agent"
def _call_data(
context: Any,
items: list[Any],
instructions: str | None = "system",
) -> CallModelData[Any]:
return CallModelData(
model_data=ModelInputData(input=items, instructions=instructions),
agent=_FakeAgent(),
context=context,
)
@pytest.mark.asyncio
async def test_empty_inbox_passes_through() -> None:
bus = AgentMessageBus()
await bus.register("a1", "alpha", parent_id=None)
data = _call_data({"bus": bus, "agent_id": "a1"}, [{"role": "user", "content": "x"}])
out = await inject_messages_filter(data)
assert out.input == [{"role": "user", "content": "x"}]
assert out.instructions == "system"
@pytest.mark.asyncio
async def test_pending_messages_appended_in_order() -> None:
bus = AgentMessageBus()
await bus.register("a1", "alpha", parent_id=None)
await bus.send("a1", {"from": "b", "content": "hello", "type": "info", "priority": "normal"})
await bus.send("a1", {"from": "c", "content": "second", "type": "info", "priority": "high"})
data = _call_data({"bus": bus, "agent_id": "a1"}, [{"role": "user", "content": "task"}])
out = await inject_messages_filter(data)
assert len(out.input) == 3
assert out.input[0] == {"role": "user", "content": "task"}
assert "Message from agent b" in out.input[1]["content"]
assert "hello" in out.input[1]["content"]
assert "second" in out.input[2]["content"]
assert "priority=high" in out.input[2]["content"]
@pytest.mark.asyncio
async def test_user_sender_skips_envelope() -> None:
bus = AgentMessageBus()
await bus.register("a1", "alpha", parent_id=None)
await bus.send("a1", {"from": "user", "content": "follow-up question"})
data = _call_data({"bus": bus, "agent_id": "a1"}, [])
out = await inject_messages_filter(data)
assert out.input == [{"role": "user", "content": "follow-up question"}]
@pytest.mark.asyncio
async def test_no_bus_in_context_passes_through() -> None:
data = _call_data({"agent_id": "a1"}, [{"role": "user", "content": "x"}])
out = await inject_messages_filter(data)
assert out.input == [{"role": "user", "content": "x"}]
@pytest.mark.asyncio
async def test_filter_exception_returns_unmodified() -> None:
"""C14 (AUDIT_R3): filter exception is caught; original data returned."""
class _BrokenBus:
async def drain(self, _: str) -> list[dict[str, Any]]:
raise RuntimeError("simulated bug")
data = _call_data(
{"bus": _BrokenBus(), "agent_id": "a1"},
[{"role": "user", "content": "still works"}],
)
out = await inject_messages_filter(data)
assert out.input == [{"role": "user", "content": "still works"}]
tests/orchestration/test_hooks.py
-174
View File
@@ -1,174 +0,0 @@
"""Phase 0 smoke tests for StrixOrchestrationHooks."""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
import pytest
from strix.orchestration.bus import AgentMessageBus
from strix.orchestration.hooks import StrixOrchestrationHooks
@dataclass
class _Ctx:
"""Minimal stand-in for RunContextWrapper / AgentHookContext.
Only ``.context`` is exercised by the hooks under test; SDK's real wrappers
expose much more, but the hooks treat ``.context`` as the dict we put in.
"""
context: dict[str, Any] = field(default_factory=dict)
@dataclass
class _Tool:
name: str
class _FakeTracer:
def __init__(self) -> None:
self.starts: list[tuple[str, str]] = []
self.ends: list[tuple[str, str, Any]] = []
def log_tool_start(self, agent_id: str, tool_name: str) -> None:
self.starts.append((agent_id, tool_name))
def log_tool_end(self, agent_id: str, tool_name: str, result: Any) -> None:
self.ends.append((agent_id, tool_name, result))
@pytest.mark.asyncio
async def test_on_llm_start_injects_85_percent_warning() -> None:
hooks = StrixOrchestrationHooks()
items: list[Any] = []
ctx = _Ctx(context={"max_turns": 100, "turn_count": 85})
await hooks.on_llm_start(ctx, agent=None, system_prompt="x", input_items=items)
assert len(items) == 1
assert "85%" in items[0]["content"]
@pytest.mark.asyncio
async def test_on_llm_start_injects_n_minus_3_warning() -> None:
hooks = StrixOrchestrationHooks()
items: list[Any] = []
ctx = _Ctx(context={"max_turns": 100, "turn_count": 97})
await hooks.on_llm_start(ctx, agent=None, system_prompt="x", input_items=items)
assert len(items) == 1
assert "3 iterations left" in items[0]["content"]
@pytest.mark.asyncio
async def test_on_llm_start_no_warning_at_other_turns() -> None:
hooks = StrixOrchestrationHooks()
items: list[Any] = []
ctx = _Ctx(context={"max_turns": 100, "turn_count": 50})
await hooks.on_llm_start(ctx, agent=None, system_prompt="x", input_items=items)
assert items == []
@pytest.mark.asyncio
async def test_on_llm_end_records_usage_and_increments_turn() -> None:
hooks = StrixOrchestrationHooks()
bus = AgentMessageBus()
await bus.register("a1", "alpha", parent_id=None)
ctx = _Ctx(context={"bus": bus, "agent_id": "a1", "turn_count": 0})
class _Details:
cached_tokens = 5
class _Usage:
input_tokens = 10
output_tokens = 20
input_tokens_details = _Details()
class _Resp:
usage = _Usage()
await hooks.on_llm_end(ctx, agent=None, response=_Resp())
assert ctx.context["turn_count"] == 1
assert bus.stats_live["a1"]["in"] == 10
@pytest.mark.asyncio
async def test_on_agent_end_detects_crash() -> None:
"""on_agent_end without agent_finish_called posts crash message to parent."""
hooks = StrixOrchestrationHooks()
bus = AgentMessageBus()
await bus.register("root", "root", parent_id=None)
await bus.register("child", "specialist", parent_id="root")
ctx = _Ctx(context={"bus": bus, "agent_id": "child"})
await hooks.on_agent_end(ctx, agent=None, output=None) # crashed (output=None)
drained = await bus.drain("root")
assert len(drained) == 1
assert "[Agent crash]" in drained[0]["content"]
assert "child" in drained[0]["content"]
assert drained[0]["type"] == "crash"
assert bus.statuses["child"] == "crashed"
@pytest.mark.asyncio
async def test_on_agent_end_no_crash_when_finish_called() -> None:
hooks = StrixOrchestrationHooks()
bus = AgentMessageBus()
await bus.register("root", "root", parent_id=None)
await bus.register("child", "specialist", parent_id="root")
ctx = _Ctx(
context={
"bus": bus,
"agent_id": "child",
"agent_finish_called": True,
}
)
await hooks.on_agent_end(ctx, agent=None, output="done")
assert await bus.drain("root") == []
assert bus.statuses["child"] == "completed"
@pytest.mark.asyncio
async def test_on_tool_end_marks_finish_called() -> None:
"""When agent_finish or finish_scan returns, mark context flag for crash detection."""
hooks = StrixOrchestrationHooks()
ctx = _Ctx(context={"agent_id": "a1"})
await hooks.on_tool_end(ctx, agent=None, tool=_Tool("agent_finish"), result="ok")
assert ctx.context["agent_finish_called"] is True
@pytest.mark.asyncio
async def test_on_tool_end_other_tool_does_not_set_flag() -> None:
hooks = StrixOrchestrationHooks()
ctx = _Ctx(context={"agent_id": "a1"})
await hooks.on_tool_end(ctx, agent=None, tool=_Tool("terminal_execute"), result="x")
assert ctx.context.get("agent_finish_called") is None
@pytest.mark.asyncio
async def test_on_tool_start_logs_to_tracer() -> None:
hooks = StrixOrchestrationHooks()
tracer = _FakeTracer()
ctx = _Ctx(context={"tracer": tracer, "agent_id": "a1"})
await hooks.on_tool_start(ctx, agent=None, tool=_Tool("browser_action"))
assert tracer.starts == [("a1", "browser_action")]
@pytest.mark.asyncio
async def test_hook_exception_does_not_propagate() -> None:
"""C15 (AUDIT_R3): a bug in the hook body must never tear down the run."""
hooks = StrixOrchestrationHooks()
class _BrokenBus:
async def record_usage(self, *_: Any, **__: Any) -> None:
raise RuntimeError("simulated")
ctx = _Ctx(context={"bus": _BrokenBus(), "agent_id": "a1"})
class _Resp:
usage = None
# Should not raise.
await hooks.on_llm_end(ctx, agent=None, response=_Resp())
tests/runtime/__init__.py
-1
View File
@@ -1 +0,0 @@
"""Tests for strix.runtime module."""
tests/runtime/test_strix_docker_client.py
-103
View File
@@ -1,103 +0,0 @@
"""Phase 0 smoke tests for StrixDockerSandboxClient.
These tests do NOT require Docker. They mock the Docker SDK client (passed
to the constructor) and verify our subclass injects the right kwargs into
``containers.create``. Live container tests are part of Phase 0 manual
smoke (TESTING_STRATEGY.md §9).
"""
from __future__ import annotations
import asyncio
from typing import Any
from unittest.mock import MagicMock
import docker.errors # type: ignore[import-untyped,unused-ignore]
import pytest
from strix.runtime.strix_docker_client import StrixDockerSandboxClient
@pytest.fixture
def fake_docker() -> MagicMock:
"""Stand-in for the Docker SDK client passed to the subclass."""
fake = MagicMock()
fake.images.get.return_value = MagicMock() # image_exists -> True
fake.images.pull = MagicMock()
fake.containers.create.return_value = MagicMock(name="container")
return fake
def _create_kwargs(fake: MagicMock) -> dict[str, Any]:
result: dict[str, Any] = fake.containers.create.call_args.kwargs
return result
def test_subclass_injects_net_admin_and_net_raw(fake_docker: MagicMock) -> None:
client = StrixDockerSandboxClient(docker_client=fake_docker)
asyncio.run(
client._create_container("strix-image:latest", manifest=None, exposed_ports=()),
)
kwargs = _create_kwargs(fake_docker)
assert "NET_ADMIN" in kwargs["cap_add"]
assert "NET_RAW" in kwargs["cap_add"]
def test_subclass_injects_host_gateway(fake_docker: MagicMock) -> None:
client = StrixDockerSandboxClient(docker_client=fake_docker)
asyncio.run(
client._create_container("strix-image:latest", manifest=None, exposed_ports=()),
)
kwargs = _create_kwargs(fake_docker)
assert kwargs["extra_hosts"]["host.docker.internal"] == "host-gateway"
def test_subclass_preserves_image_and_command(fake_docker: MagicMock) -> None:
client = StrixDockerSandboxClient(docker_client=fake_docker)
asyncio.run(
client._create_container("custom:tag", manifest=None, exposed_ports=()),
)
kwargs = _create_kwargs(fake_docker)
assert kwargs["image"] == "custom:tag"
assert kwargs["entrypoint"] == ["tail"]
assert kwargs["command"] == ["-f", "/dev/null"]
assert kwargs["detach"] is True
def test_subclass_emits_ports_dict_for_exposed_ports(fake_docker: MagicMock) -> None:
client = StrixDockerSandboxClient(docker_client=fake_docker)
asyncio.run(
client._create_container(
"strix-image:latest",
manifest=None,
exposed_ports=(48081, 48080),
),
)
kwargs = _create_kwargs(fake_docker)
assert "48081/tcp" in kwargs["ports"]
assert "48080/tcp" in kwargs["ports"]
def test_caps_appended_not_duplicated(fake_docker: MagicMock) -> None:
"""Idempotent injection: calling twice doesn't add duplicate caps."""
client = StrixDockerSandboxClient(docker_client=fake_docker)
asyncio.run(
client._create_container("img:latest", manifest=None, exposed_ports=()),
)
kwargs = _create_kwargs(fake_docker)
assert kwargs["cap_add"].count("NET_ADMIN") == 1
assert kwargs["cap_add"].count("NET_RAW") == 1
def test_pulls_image_when_missing(fake_docker: MagicMock) -> None:
"""If image_exists returns False on first check, pull is invoked."""
# First call raises ImageNotFound, second succeeds.
fake_docker.images.get.side_effect = [
docker.errors.ImageNotFound("not found"),
MagicMock(),
]
client = StrixDockerSandboxClient(docker_client=fake_docker)
asyncio.run(
client._create_container("registry.io/strix:tag", manifest=None, exposed_ports=()),
)
fake_docker.images.pull.assert_called_once()
tests/sandbox/__init__.py
View File
tests/sandbox/test_caido_capability.py
-156
View File
@@ -1,156 +0,0 @@
"""Phase 4 tests for CaidoCapability.
The capability has three observable behaviors that need parity with the
PLAYBOOK contract:
1. ``process_manifest`` injects http_proxy / https_proxy / ALL_PROXY env
vars into the manifest's ``Environment.value`` dict.
2. ``tools()`` returns the seven Caido SDK function tools we wrapped in
Phase 2.5 — same instances, in the documented order.
3. ``bind`` schedules an aggregated healthcheck task; the orchestration
hook later awaits it on first agent start.
The healthcheck task itself is exercised by the healthcheck unit tests;
here we only verify the wiring (task created, name set, points at the
right ports).
"""
from __future__ import annotations
import asyncio
from unittest.mock import MagicMock, patch
import pytest
from agents.sandbox.entries import LocalDir
from agents.sandbox.manifest import Environment, Manifest
from strix.sandbox.caido_capability import CaidoCapability
def test_capability_type_and_default_state() -> None:
cap = CaidoCapability()
assert cap.type == "caido"
assert cap._healthcheck_task is None
assert cap._tool_server_host_port is None
assert cap._caido_host_port is None
def test_process_manifest_injects_proxy_env_vars(tmp_path: object) -> None:
"""Existing env vars must be preserved; proxy keys are added."""
cap = CaidoCapability()
manifest = Manifest(
environment=Environment(
value={"PYTHONUNBUFFERED": "1", "TOOL_SERVER_TOKEN": "abc"},
),
)
out = cap.process_manifest(manifest)
env = out.environment.value
# Pre-existing entries preserved.
assert env["PYTHONUNBUFFERED"] == "1"
assert env["TOOL_SERVER_TOKEN"] == "abc"
# Proxy entries injected, all pointing at the in-container Caido port.
assert env["http_proxy"] == "http://127.0.0.1:48080"
assert env["https_proxy"] == "http://127.0.0.1:48080"
assert env["ALL_PROXY"] == "http://127.0.0.1:48080"
def test_process_manifest_handles_missing_environment() -> None:
"""A manifest without env entries should still get the proxy block."""
cap = CaidoCapability()
# ``LocalDir`` requires a real path on disk; use a temp one to satisfy
# the validator without actually mounting anything.
manifest = Manifest(entries={"src": LocalDir(src="/tmp")})
out = cap.process_manifest(manifest)
env = out.environment.value
assert env["http_proxy"] == "http://127.0.0.1:48080"
def test_tools_returns_seven_caido_tools_in_order() -> None:
cap = CaidoCapability()
names = [t.name for t in cap.tools()]
assert names == [
"list_requests",
"view_request",
"send_request",
"repeat_request",
"scope_rules",
"list_sitemap",
"view_sitemap_entry",
]
def test_tools_returns_a_fresh_list_per_call() -> None:
"""SDK convention — caller may mutate the returned list."""
cap = CaidoCapability()
a = cap.tools()
b = cap.tools()
assert a == b
assert a is not b
@pytest.mark.asyncio
async def test_instructions_mentions_caido_and_tools() -> None:
cap = CaidoCapability()
out = await cap.instructions(Manifest())
assert out is not None
assert "<caido_proxy>" in out
# Every tool name appears verbatim so the model knows what's available.
for name in (
"list_requests",
"view_request",
"send_request",
"repeat_request",
"scope_rules",
"list_sitemap",
"view_sitemap_entry",
):
assert name in out
def test_configure_host_ports_stores_both() -> None:
cap = CaidoCapability()
cap.configure_host_ports(tool_server_host_port=12345, caido_host_port=12346)
assert cap._tool_server_host_port == 12345
assert cap._caido_host_port == 12346
@pytest.mark.asyncio
async def test_bind_without_configured_ports_skips_healthcheck() -> None:
"""If the session manager forgets to configure ports, bind shouldn't
schedule a probe against ``None`` — it should warn and no-op.
"""
cap = CaidoCapability()
fake_session = MagicMock()
cap.bind(fake_session)
assert cap._healthcheck_task is None
assert cap.session is fake_session
@pytest.mark.asyncio
async def test_bind_schedules_healthcheck_task_when_ports_configured() -> None:
"""The hook chain (StrixOrchestrationHooks.on_agent_start) awaits this
task — it must exist as an asyncio.Task with a useful name.
"""
cap = CaidoCapability()
cap.configure_host_ports(tool_server_host_port=54321, caido_host_port=54322)
fake_session = MagicMock()
# Patch the actual probes so we don't try to connect for real.
async def _fake_probe(*args: object, **kwargs: object) -> None:
return None
with (
patch(
"strix.sandbox.caido_capability.wait_for_http_ready",
side_effect=_fake_probe,
),
patch(
"strix.sandbox.caido_capability.wait_for_tcp_ready",
side_effect=_fake_probe,
),
):
cap.bind(fake_session)
assert cap._healthcheck_task is not None
assert isinstance(cap._healthcheck_task, asyncio.Task)
assert "caido-healthcheck-54321" in cap._healthcheck_task.get_name()
await cap._healthcheck_task # must complete without error
tests/sandbox/test_healthcheck.py
-171
View File
@@ -1,171 +0,0 @@
"""Phase 4 tests for the sandbox port readiness probes.
The two helpers (``wait_for_http_ready`` and ``wait_for_tcp_ready``)
gate session bring-up, so a regression here would mean every fresh
scan hits a connection-refused on its first tool call. Tests cover:
- Happy path returns when the probe succeeds.
- Polling continues across transient failures.
- Timeout raises ``SandboxNotReadyError`` with a useful last-error.
- Real ``asyncio.open_connection`` against a local listener verifies
the TCP probe end-to-end (no mocking — the helper is small enough
that a real socket is the cheaper test).
"""
from __future__ import annotations
import asyncio
import contextlib
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
import httpx
import pytest
from strix.sandbox.healthcheck import (
SandboxNotReadyError,
wait_for_http_ready,
wait_for_tcp_ready,
)
# --- HTTP probe ----------------------------------------------------------
@pytest.mark.asyncio
async def test_wait_for_http_ready_returns_immediately_on_2xx() -> None:
response = MagicMock(spec=httpx.Response)
response.status_code = 200
client = AsyncMock()
client.get = AsyncMock(return_value=response)
client.__aenter__ = AsyncMock(return_value=client)
client.__aexit__ = AsyncMock(return_value=None)
with patch("strix.sandbox.healthcheck.httpx.AsyncClient", return_value=client):
await wait_for_http_ready("http://localhost:9999/health", timeout=1)
assert client.get.await_count == 1
@pytest.mark.asyncio
async def test_wait_for_http_ready_polls_through_connect_errors() -> None:
"""Two connect errors followed by a 200 — the helper should keep going."""
response_ok = MagicMock(spec=httpx.Response)
response_ok.status_code = 200
side_effects: list[Any] = [
httpx.ConnectError("conn refused"),
httpx.ConnectError("conn refused"),
response_ok,
]
client = AsyncMock()
client.get = AsyncMock(side_effect=side_effects)
client.__aenter__ = AsyncMock(return_value=client)
client.__aexit__ = AsyncMock(return_value=None)
with patch("strix.sandbox.healthcheck.httpx.AsyncClient", return_value=client):
await wait_for_http_ready(
"http://localhost:9999/health",
timeout=5,
poll_interval=0.01,
)
assert client.get.await_count == 3
@pytest.mark.asyncio
async def test_wait_for_http_ready_raises_after_timeout() -> None:
client = AsyncMock()
client.get = AsyncMock(side_effect=httpx.ConnectError("nope"))
client.__aenter__ = AsyncMock(return_value=client)
client.__aexit__ = AsyncMock(return_value=None)
with (
patch(
"strix.sandbox.healthcheck.httpx.AsyncClient",
return_value=client,
),
pytest.raises(SandboxNotReadyError) as exc_info,
):
await wait_for_http_ready(
"http://localhost:9999/health",
timeout=0.3,
poll_interval=0.05,
)
err = str(exc_info.value)
assert "http://localhost:9999/health" in err
assert "ConnectError" in err
@pytest.mark.asyncio
async def test_wait_for_http_ready_treats_5xx_as_not_ready() -> None:
response_500 = MagicMock(spec=httpx.Response)
response_500.status_code = 500
response_ok = MagicMock(spec=httpx.Response)
response_ok.status_code = 200
client = AsyncMock()
client.get = AsyncMock(side_effect=[response_500, response_ok])
client.__aenter__ = AsyncMock(return_value=client)
client.__aexit__ = AsyncMock(return_value=None)
with patch("strix.sandbox.healthcheck.httpx.AsyncClient", return_value=client):
await wait_for_http_ready(
"http://localhost:9999/health",
timeout=2,
poll_interval=0.01,
)
assert client.get.await_count == 2
# --- TCP probe -----------------------------------------------------------
@pytest.mark.asyncio
async def test_wait_for_tcp_ready_against_real_listener() -> None:
"""Spin up a local TCP echo server and verify the probe connects."""
async def _server_handler(
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
) -> None:
# Drain any bytes the test sends, then close.
await reader.read(0)
writer.close()
with contextlib.suppress(OSError):
await writer.wait_closed()
server = await asyncio.start_server(_server_handler, "127.0.0.1", 0)
port = server.sockets[0].getsockname()[1]
try:
await wait_for_tcp_ready("127.0.0.1", port, timeout=2, poll_interval=0.05)
finally:
server.close()
await server.wait_closed()
@pytest.mark.asyncio
async def test_wait_for_tcp_ready_raises_when_port_closed() -> None:
async def _no_handler(
_reader: asyncio.StreamReader,
_writer: asyncio.StreamWriter,
) -> None:
return
# Bind and immediately close to claim a definitely-unused port number.
server = await asyncio.start_server(_no_handler, "127.0.0.1", 0)
closed_port = server.sockets[0].getsockname()[1]
server.close()
await server.wait_closed()
with pytest.raises(SandboxNotReadyError) as exc_info:
await wait_for_tcp_ready(
"127.0.0.1",
closed_port,
timeout=0.3,
poll_interval=0.05,
)
err = str(exc_info.value)
assert f"127.0.0.1:{closed_port}" in err
tests/sandbox/test_session_manager.py
-206
View File
@@ -1,206 +0,0 @@
"""Phase 4 tests for the per-scan sandbox session manager.
We don't spin up real Docker here — the ``StrixDockerSandboxClient`` is
patched and we assert on the manifest / options / bundle shape. Goals:
- Cache hit: a second ``create_or_reuse(scan_id, ...)`` returns the same
bundle without calling client.create twice.
- Manifest carries the right env vars (TOOL_SERVER_TOKEN, container ports,
STRIX_SANDBOX_EXECUTION_TIMEOUT, PYTHONUNBUFFERED).
- The Docker client options request both container ports be exposed.
- Capability is configured with the resolved host ports *before* bind,
so its healthcheck task probes the right ones.
- Bundle is cached and surfaces in ``cached_scan_ids``.
- ``cleanup`` cancels the healthcheck task and calls ``client.delete``;
errors during delete are swallowed.
"""
from __future__ import annotations
from collections.abc import Iterator
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from strix.sandbox import session_manager
from strix.sandbox.caido_capability import CaidoCapability
@pytest.fixture(autouse=True)
def _isolate_cache() -> Iterator[None]:
session_manager._reset_cache_for_tests()
yield
session_manager._reset_cache_for_tests()
def _noop_bind(_self: Any, _session: Any) -> None:
"""Stand-in for CaidoCapability.bind that skips the healthcheck task."""
def _make_endpoint(port: int) -> Any:
ep = MagicMock()
ep.port = port
ep.host = "127.0.0.1"
ep.tls = False
return ep
def _make_client_and_session(
*,
tool_port: int = 12001,
caido_port: int = 12002,
) -> tuple[Any, Any]:
"""Build a fake DockerSandboxClient and session pair."""
session = MagicMock()
session._resolve_exposed_port = AsyncMock(
side_effect=lambda port: _make_endpoint(
tool_port if port == 48081 else caido_port,
),
)
client = MagicMock()
client.create = AsyncMock(return_value=session)
client.delete = AsyncMock()
return client, session
@pytest.mark.asyncio
async def test_create_or_reuse_creates_new_session(tmp_path: Any) -> None:
client, session = _make_client_and_session()
# Patch the capability's bind to a no-op so we don't spin up the
# healthcheck task in unit tests.
with (
patch(
"strix.sandbox.session_manager.StrixDockerSandboxClient",
return_value=client,
),
patch.object(CaidoCapability, "bind", _noop_bind),
):
bundle = await session_manager.create_or_reuse(
"scan-1",
image="strix-sandbox:test",
sources_path=tmp_path,
)
# Bundle shape.
assert bundle["client"] is client
assert bundle["session"] is session
assert bundle["tool_server_host_port"] == 12001
assert bundle["caido_host_port"] == 12002
assert isinstance(bundle["bearer"], str) and len(bundle["bearer"]) >= 32
assert isinstance(bundle["capability"], CaidoCapability)
# Capability got the resolved host ports BEFORE bind would have run.
assert bundle["capability"]._tool_server_host_port == 12001
assert bundle["capability"]._caido_host_port == 12002
# client.create called exactly once with manifest + exposed ports.
assert client.create.await_count == 1
options = client.create.await_args.kwargs["options"]
assert options.image == "strix-sandbox:test"
assert set(options.exposed_ports) == {48080, 48081}
manifest = client.create.await_args.kwargs["manifest"]
env = manifest.environment.value
assert env["TOOL_SERVER_TOKEN"] == bundle["bearer"]
assert env["TOOL_SERVER_PORT"] == "48081"
assert env["CAIDO_PORT"] == "48080"
assert env["STRIX_SANDBOX_EXECUTION_TIMEOUT"] == "120"
assert env["PYTHONUNBUFFERED"] == "1"
assert env["HOST_GATEWAY"] == "host.docker.internal"
@pytest.mark.asyncio
async def test_create_or_reuse_returns_cached_bundle(tmp_path: Any) -> None:
client, _ = _make_client_and_session()
with (
patch(
"strix.sandbox.session_manager.StrixDockerSandboxClient",
return_value=client,
),
patch.object(CaidoCapability, "bind", _noop_bind),
):
first = await session_manager.create_or_reuse(
"scan-X",
image="i",
sources_path=tmp_path,
)
second = await session_manager.create_or_reuse(
"scan-X",
image="i",
sources_path=tmp_path,
)
assert first is second
assert client.create.await_count == 1
assert "scan-X" in session_manager.cached_scan_ids()
@pytest.mark.asyncio
async def test_create_or_reuse_passes_custom_execution_timeout(tmp_path: Any) -> None:
client, _ = _make_client_and_session()
with (
patch(
"strix.sandbox.session_manager.StrixDockerSandboxClient",
return_value=client,
),
patch.object(CaidoCapability, "bind", _noop_bind),
):
await session_manager.create_or_reuse(
"scan-2",
image="i",
sources_path=tmp_path,
execution_timeout=300,
)
manifest = client.create.await_args.kwargs["manifest"]
assert manifest.environment.value["STRIX_SANDBOX_EXECUTION_TIMEOUT"] == "300"
@pytest.mark.asyncio
async def test_cleanup_calls_delete_and_drops_cache(tmp_path: Any) -> None:
client, session = _make_client_and_session()
with (
patch(
"strix.sandbox.session_manager.StrixDockerSandboxClient",
return_value=client,
),
patch.object(CaidoCapability, "bind", _noop_bind),
):
await session_manager.create_or_reuse(
"scan-3",
image="i",
sources_path=tmp_path,
)
assert "scan-3" in session_manager.cached_scan_ids()
await session_manager.cleanup("scan-3")
client.delete.assert_awaited_once_with(session)
assert "scan-3" not in session_manager.cached_scan_ids()
@pytest.mark.asyncio
async def test_cleanup_swallows_delete_errors(tmp_path: Any) -> None:
"""A flaky Docker daemon shouldn't prevent cache eviction."""
client, _ = _make_client_and_session()
client.delete = AsyncMock(side_effect=RuntimeError("docker daemon went away"))
with (
patch(
"strix.sandbox.session_manager.StrixDockerSandboxClient",
return_value=client,
),
patch.object(CaidoCapability, "bind", _noop_bind),
):
await session_manager.create_or_reuse(
"scan-4",
image="i",
sources_path=tmp_path,
)
await session_manager.cleanup("scan-4") # must not raise
assert "scan-4" not in session_manager.cached_scan_ids()
@pytest.mark.asyncio
async def test_cleanup_unknown_scan_is_noop() -> None:
"""No cached entry → cleanup is a quiet no-op."""
await session_manager.cleanup("never-existed") # must not raise
tests/skills/__init__.py
-1
View File
@@ -1 +0,0 @@
# Tests for skill-related runtime behavior.
tests/static/__init__.py
View File
tests/telemetry/__init__.py
-1
View File
@@ -1 +0,0 @@
"""Tests for strix.telemetry module."""
tests/telemetry/test_flags.py
-28
View File
@@ -1,28 +0,0 @@
from strix.telemetry.flags import is_otel_enabled, is_posthog_enabled
def test_flags_fallback_to_strix_telemetry(monkeypatch) -> None:
monkeypatch.delenv("STRIX_OTEL_TELEMETRY", raising=False)
monkeypatch.delenv("STRIX_POSTHOG_TELEMETRY", raising=False)
monkeypatch.setenv("STRIX_TELEMETRY", "0")
assert is_otel_enabled() is False
assert is_posthog_enabled() is False
def test_otel_flag_overrides_global_telemetry(monkeypatch) -> None:
monkeypatch.setenv("STRIX_TELEMETRY", "0")
monkeypatch.setenv("STRIX_OTEL_TELEMETRY", "1")
monkeypatch.delenv("STRIX_POSTHOG_TELEMETRY", raising=False)
assert is_otel_enabled() is True
assert is_posthog_enabled() is False
def test_posthog_flag_overrides_global_telemetry(monkeypatch) -> None:
monkeypatch.setenv("STRIX_TELEMETRY", "0")
monkeypatch.setenv("STRIX_POSTHOG_TELEMETRY", "1")
monkeypatch.delenv("STRIX_OTEL_TELEMETRY", raising=False)
assert is_otel_enabled() is False
assert is_posthog_enabled() is True
tests/telemetry/test_strix_processor.py
-201
View File
@@ -1,201 +0,0 @@
"""Phase 1 smoke tests for StrixTracingProcessor."""
from __future__ import annotations
import json
import threading
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
import pytest
from strix.telemetry.strix_processor import StrixTracingProcessor
@dataclass
class _FakeSpanData:
"""Minimal stand-in for an SDK SpanData class."""
name: str = "FunctionSpanData" # used by class .__name__
payload: dict[str, Any] = field(default_factory=dict)
def export(self) -> dict[str, Any]:
return dict(self.payload)
# Concrete SpanData subclasses so the processor's ``_span_kind`` heuristic
# (drop ``SpanData`` suffix, lowercase) produces stable event_types.
class FunctionSpanData(_FakeSpanData):
pass
class GenerationSpanData(_FakeSpanData):
pass
class AgentSpanData(_FakeSpanData):
pass
@dataclass
class _FakeSpan:
span_id: str
trace_id: str
span_data: _FakeSpanData
@dataclass
class _FakeTrace:
trace_id: str
name: str = "test-workflow"
metadata: dict[str, Any] = field(default_factory=dict)
def export(self) -> dict[str, Any]:
return {"name": self.name, "metadata": self.metadata}
def _read_events(path: Path) -> list[dict[str, Any]]:
return [json.loads(line) for line in path.read_text().splitlines() if line.strip()]
@pytest.fixture
def run_dir(tmp_path: Path) -> Path:
return tmp_path / "strix_runs" / "test-run"
def test_constructor_creates_run_dir(run_dir: Path) -> None:
StrixTracingProcessor(run_dir=run_dir)
assert run_dir.exists()
def test_on_trace_start_writes_run_started(run_dir: Path) -> None:
p = StrixTracingProcessor(run_dir=run_dir)
p.on_trace_start(_FakeTrace(trace_id="t-1", metadata={"scan_id": "abc"}))
events = _read_events(p.events_path)
assert events == [
{
"event_type": "run.started",
"trace_id": "t-1",
"metadata": {"name": "test-workflow", "metadata": {"scan_id": "abc"}},
}
]
def test_on_trace_end_writes_run_completed(run_dir: Path) -> None:
p = StrixTracingProcessor(run_dir=run_dir)
p.on_trace_end(_FakeTrace(trace_id="t-1"))
events = _read_events(p.events_path)
assert events == [{"event_type": "run.completed", "trace_id": "t-1"}]
def test_span_start_and_end_emit_typed_events(run_dir: Path) -> None:
"""``GenerationSpanData`` → ``generation.started`` / ``generation.completed``."""
p = StrixTracingProcessor(run_dir=run_dir)
span = _FakeSpan(
span_id="s-1",
trace_id="t-1",
span_data=GenerationSpanData(payload={"model": "gpt-foo"}),
)
p.on_span_start(span)
p.on_span_end(span)
events = _read_events(p.events_path)
assert [e["event_type"] for e in events] == [
"generation.started",
"generation.completed",
]
assert events[0]["span_id"] == "s-1"
assert events[0]["data"] == {"model": "gpt-foo"}
def test_concurrent_writes_yield_valid_jsonl(run_dir: Path) -> None:
"""C7 (AUDIT_R2): per-path lock prevents JSONL corruption under contention."""
p = StrixTracingProcessor(run_dir=run_dir)
def writer(idx: int) -> None:
for i in range(50):
p._emit({"event_type": "synthetic", "writer": idx, "i": i})
threads = [threading.Thread(target=writer, args=(i,)) for i in range(10)]
for t in threads:
t.start()
for t in threads:
t.join()
# 10 threads x 50 events = 500 lines; all valid JSON.
lines = p.events_path.read_text().splitlines()
assert len(lines) == 500
for line in lines:
json.loads(line) # raises on corrupt line
def test_emit_swallows_oserror_and_logs(
run_dir: Path,
caplog: pytest.LogCaptureFixture,
) -> None:
"""C16 (AUDIT_R3): a write failure must NOT propagate."""
p = StrixTracingProcessor(run_dir=run_dir)
# Make events_path point to a directory so open(..., "a") raises.
p.events_path = run_dir
with caplog.at_level("ERROR", logger="strix.telemetry.strix_processor"):
p._emit({"event_type": "boom"})
assert any("Failed to append" in rec.message for rec in caplog.records)
def test_span_export_failure_does_not_propagate(run_dir: Path) -> None:
"""If span_data.export raises, we still emit an event with data=None."""
p = StrixTracingProcessor(run_dir=run_dir)
class _BoomSpanData:
def export(self) -> dict[str, Any]:
raise RuntimeError("nope")
# Reuse the lowercase rule: class name has no "SpanData" suffix → "boomspandata"
# would not be ideal; use a properly-named subclass.
class FunctionSpanDataBroken(_BoomSpanData):
pass
span = _FakeSpan(span_id="s-1", trace_id="t-1", span_data=FunctionSpanDataBroken())
p.on_span_end(span)
events = _read_events(p.events_path)
assert len(events) == 1
assert events[0]["data"] is None
def test_pii_scrubbed_via_sanitizer(run_dir: Path) -> None:
"""Sanitizer is invoked on every emit before write."""
seen: list[Any] = []
class _StubSanitizer:
def sanitize(self, data: Any, key_hint: str | None = None) -> Any:
seen.append(data)
# Replace any "secret" string with [REDACTED].
if isinstance(data, dict):
clean = {k: "[REDACTED]" if k == "api_key" else v for k, v in data.items()}
if "metadata" in clean and isinstance(clean["metadata"], dict):
md = dict(clean["metadata"])
md.pop("api_key", None)
clean["metadata"] = md
return clean
return data
p = StrixTracingProcessor(run_dir=run_dir, sanitizer=_StubSanitizer())
p._emit({"event_type": "test", "api_key": "sk-very-secret"})
events = _read_events(p.events_path)
assert events[0]["api_key"] == "[REDACTED]"
assert seen and seen[0]["api_key"] == "sk-very-secret"
def test_force_flush_and_shutdown_are_noops(run_dir: Path) -> None:
p = StrixTracingProcessor(run_dir=run_dir)
# Should not raise.
p.force_flush()
p.shutdown()
tests/telemetry/test_tracer.py
-428
View File
@@ -1,428 +0,0 @@
import json
import sys
import types
from pathlib import Path
from typing import Any, ClassVar
import pytest
from opentelemetry.sdk.trace.export import SimpleSpanProcessor, SpanExportResult
from strix.telemetry import tracer as tracer_module
from strix.telemetry import utils as telemetry_utils
from strix.telemetry.tracer import Tracer, set_global_tracer
def _load_events(events_path: Path) -> list[dict[str, Any]]:
lines = events_path.read_text(encoding="utf-8").splitlines()
return [json.loads(line) for line in lines if line]
@pytest.fixture(autouse=True)
def _reset_tracer_globals(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(tracer_module, "_global_tracer", None)
monkeypatch.setattr(tracer_module, "_OTEL_BOOTSTRAPPED", False)
monkeypatch.setattr(tracer_module, "_OTEL_REMOTE_ENABLED", False)
telemetry_utils.reset_events_write_locks()
monkeypatch.delenv("STRIX_TELEMETRY", raising=False)
monkeypatch.delenv("STRIX_OTEL_TELEMETRY", raising=False)
monkeypatch.delenv("STRIX_POSTHOG_TELEMETRY", raising=False)
monkeypatch.delenv("TRACELOOP_BASE_URL", raising=False)
monkeypatch.delenv("TRACELOOP_API_KEY", raising=False)
monkeypatch.delenv("TRACELOOP_HEADERS", raising=False)
def test_tracer_local_mode_writes_jsonl_with_correlation(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
monkeypatch.chdir(tmp_path)
tracer = Tracer("local-observability")
set_global_tracer(tracer)
tracer.set_scan_config({"targets": ["https://example.com"], "user_instructions": "focus auth"})
tracer.log_chat_message("starting scan", "user", "agent-1")
tracer.log_chat_message("scanning login form", "assistant", "agent-1")
events_path = tmp_path / "strix_runs" / "local-observability" / "events.jsonl"
assert events_path.exists()
events = _load_events(events_path)
assert any(event["event_type"] == "chat.message" for event in events)
assert any(event["event_type"] == "run.configured" for event in events)
for event in events:
assert event["run_id"] == "local-observability"
assert event["trace_id"]
assert event["span_id"]
def test_tracer_redacts_sensitive_payloads(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
monkeypatch.chdir(tmp_path)
tracer = Tracer("redaction-run")
set_global_tracer(tracer)
tracer.log_chat_message(
"request failed with token sk-secret-token-value",
"assistant",
"agent-1",
metadata={
"api_key": "sk-secret-token-value",
"authorization": "Bearer super-secret-token",
},
)
events_path = tmp_path / "strix_runs" / "redaction-run" / "events.jsonl"
events = _load_events(events_path)
serialized = json.dumps(events)
assert "sk-secret-token-value" not in serialized
assert "super-secret-token" not in serialized
assert "[REDACTED]" in serialized
def test_tracer_remote_mode_configures_traceloop_export(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
monkeypatch.chdir(tmp_path)
class FakeTraceloop:
init_calls: ClassVar[list[dict[str, Any]]] = []
@staticmethod
def init(**kwargs: Any) -> None:
FakeTraceloop.init_calls.append(kwargs)
@staticmethod
def set_association_properties(properties: dict[str, Any]) -> None: # noqa: ARG004
return None
monkeypatch.setattr(tracer_module, "Traceloop", FakeTraceloop)
monkeypatch.setenv("TRACELOOP_BASE_URL", "https://otel.example.com")
monkeypatch.setenv("TRACELOOP_API_KEY", "test-api-key")
monkeypatch.setenv("TRACELOOP_HEADERS", '{"x-custom":"header"}')
tracer = Tracer("remote-observability")
set_global_tracer(tracer)
tracer.log_chat_message("hello", "user", "agent-1")
assert tracer._remote_export_enabled is True
assert FakeTraceloop.init_calls
init_kwargs = FakeTraceloop.init_calls[-1]
assert init_kwargs["api_endpoint"] == "https://otel.example.com"
assert init_kwargs["api_key"] == "test-api-key"
assert init_kwargs["headers"] == {"x-custom": "header"}
assert isinstance(init_kwargs["processor"], SimpleSpanProcessor)
assert "strix.run_id" not in init_kwargs["resource_attributes"]
assert "strix.run_name" not in init_kwargs["resource_attributes"]
events_path = tmp_path / "strix_runs" / "remote-observability" / "events.jsonl"
events = _load_events(events_path)
run_started = next(event for event in events if event["event_type"] == "run.started")
assert run_started["payload"]["remote_export_enabled"] is True
def test_tracer_local_mode_avoids_traceloop_remote_endpoint(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
monkeypatch.chdir(tmp_path)
class FakeTraceloop:
init_calls: ClassVar[list[dict[str, Any]]] = []
@staticmethod
def init(**kwargs: Any) -> None:
FakeTraceloop.init_calls.append(kwargs)
@staticmethod
def set_association_properties(properties: dict[str, Any]) -> None: # noqa: ARG004
return None
monkeypatch.setattr(tracer_module, "Traceloop", FakeTraceloop)
tracer = Tracer("local-traceloop")
set_global_tracer(tracer)
tracer.log_chat_message("hello", "user", "agent-1")
assert FakeTraceloop.init_calls
init_kwargs = FakeTraceloop.init_calls[-1]
assert "api_endpoint" not in init_kwargs
assert "api_key" not in init_kwargs
assert "headers" not in init_kwargs
assert isinstance(init_kwargs["processor"], SimpleSpanProcessor)
assert tracer._remote_export_enabled is False
def test_otlp_fallback_includes_auth_and_custom_headers(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
monkeypatch.chdir(tmp_path)
monkeypatch.setattr(tracer_module, "Traceloop", None)
monkeypatch.setenv("TRACELOOP_BASE_URL", "https://otel.example.com")
monkeypatch.setenv("TRACELOOP_API_KEY", "test-api-key")
monkeypatch.setenv("TRACELOOP_HEADERS", '{"x-custom":"header"}')
captured: dict[str, Any] = {}
class FakeOTLPSpanExporter:
def __init__(self, endpoint: str, headers: dict[str, str] | None = None, **kwargs: Any):
captured["endpoint"] = endpoint
captured["headers"] = headers or {}
captured["kwargs"] = kwargs
def export(self, spans: Any) -> SpanExportResult:
return SpanExportResult.SUCCESS
def shutdown(self) -> None:
return None
def force_flush(self, timeout_millis: int = 30_000) -> bool:
return True
fake_module = types.ModuleType("opentelemetry.exporter.otlp.proto.http.trace_exporter")
fake_module.OTLPSpanExporter = FakeOTLPSpanExporter
monkeypatch.setitem(
sys.modules,
"opentelemetry.exporter.otlp.proto.http.trace_exporter",
fake_module,
)
tracer = Tracer("otlp-fallback")
set_global_tracer(tracer)
assert tracer._remote_export_enabled is True
assert captured["endpoint"] == "https://otel.example.com/v1/traces"
assert captured["headers"]["Authorization"] == "Bearer test-api-key"
assert captured["headers"]["x-custom"] == "header"
def test_traceloop_init_failure_does_not_mark_bootstrapped_on_provider_failure(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
monkeypatch.chdir(tmp_path)
class FakeTraceloop:
@staticmethod
def init(**kwargs: Any) -> None: # noqa: ARG004
raise RuntimeError("traceloop init failed")
@staticmethod
def set_association_properties(properties: dict[str, Any]) -> None: # noqa: ARG004
return None
monkeypatch.setattr(tracer_module, "Traceloop", FakeTraceloop)
def _raise_provider_error(provider: Any) -> None:
raise RuntimeError("provider setup failed")
monkeypatch.setattr(tracer_module.trace, "set_tracer_provider", _raise_provider_error)
tracer = Tracer("bootstrap-failure")
set_global_tracer(tracer)
assert tracer_module._OTEL_BOOTSTRAPPED is False
assert tracer._remote_export_enabled is False
def test_run_completed_event_emitted_once(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
monkeypatch.chdir(tmp_path)
tracer = Tracer("single-complete")
set_global_tracer(tracer)
tracer.save_run_data(mark_complete=True)
tracer.save_run_data(mark_complete=True)
events_path = tmp_path / "strix_runs" / "single-complete" / "events.jsonl"
events = _load_events(events_path)
run_completed = [event for event in events if event["event_type"] == "run.completed"]
assert len(run_completed) == 1
def test_events_with_agent_id_include_agent_name(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
monkeypatch.chdir(tmp_path)
tracer = Tracer("agent-name-enrichment")
set_global_tracer(tracer)
# _enrich_actor pulls names from the tracer's agents dict; populate it
# the same way the orchestration path will once wired (placeholder
# until live wiring lands).
tracer.agents["agent-1"] = {"name": "Root Agent"}
tracer.log_chat_message("hello", "assistant", "agent-1")
events_path = tmp_path / "strix_runs" / "agent-name-enrichment" / "events.jsonl"
events = _load_events(events_path)
chat_event = next(event for event in events if event["event_type"] == "chat.message")
assert chat_event["actor"]["agent_id"] == "agent-1"
assert chat_event["actor"]["agent_name"] == "Root Agent"
def test_get_total_llm_stats_aggregates_recordings(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
monkeypatch.chdir(tmp_path)
tracer = Tracer("cost-rollup")
set_global_tracer(tracer)
tracer.record_llm_usage(
input_tokens=1_000,
output_tokens=250,
cached_tokens=100,
cost=0.12345,
requests=2,
)
tracer.record_llm_usage(
input_tokens=2_000,
output_tokens=500,
cached_tokens=400,
cost=0.54321,
requests=3,
)
stats = tracer.get_total_llm_stats()
assert stats["total"] == {
"input_tokens": 3_000,
"output_tokens": 750,
"cached_tokens": 500,
"cost": 0.6667,
"requests": 5,
}
assert stats["total_tokens"] == 3_750
def test_run_metadata_is_only_on_run_lifecycle_events(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
monkeypatch.chdir(tmp_path)
tracer = Tracer("metadata-scope")
set_global_tracer(tracer)
tracer.log_chat_message("hello", "assistant", "agent-1")
tracer.save_run_data(mark_complete=True)
events_path = tmp_path / "strix_runs" / "metadata-scope" / "events.jsonl"
events = _load_events(events_path)
run_started = next(event for event in events if event["event_type"] == "run.started")
run_completed = next(event for event in events if event["event_type"] == "run.completed")
chat_event = next(event for event in events if event["event_type"] == "chat.message")
assert "run_metadata" in run_started
assert "run_metadata" in run_completed
assert "run_metadata" not in chat_event
def test_set_run_name_resets_cached_paths(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
monkeypatch.chdir(tmp_path)
tracer = Tracer()
set_global_tracer(tracer)
old_events_path = tracer.events_file_path
tracer.set_run_name("renamed-run")
tracer.log_chat_message("hello", "assistant", "agent-1")
new_events_path = tracer.events_file_path
assert new_events_path != old_events_path
assert new_events_path == tmp_path / "strix_runs" / "renamed-run" / "events.jsonl"
events = _load_events(new_events_path)
assert any(event["event_type"] == "run.started" for event in events)
assert any(event["event_type"] == "chat.message" for event in events)
def test_set_run_name_resets_run_completed_flag(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
monkeypatch.chdir(tmp_path)
tracer = Tracer()
set_global_tracer(tracer)
tracer.save_run_data(mark_complete=True)
tracer.set_run_name("renamed-complete")
tracer.save_run_data(mark_complete=True)
events_path = tmp_path / "strix_runs" / "renamed-complete" / "events.jsonl"
events = _load_events(events_path)
run_completed = [event for event in events if event["event_type"] == "run.completed"]
assert any(event["event_type"] == "run.started" for event in events)
assert len(run_completed) == 1
def test_set_run_name_updates_traceloop_association_properties(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
monkeypatch.chdir(tmp_path)
class FakeTraceloop:
associations: ClassVar[list[dict[str, Any]]] = []
@staticmethod
def init(**kwargs: Any) -> None: # noqa: ARG004
return None
@staticmethod
def set_association_properties(properties: dict[str, Any]) -> None:
FakeTraceloop.associations.append(properties)
monkeypatch.setattr(tracer_module, "Traceloop", FakeTraceloop)
tracer = Tracer()
set_global_tracer(tracer)
tracer.set_run_name("renamed-run")
assert FakeTraceloop.associations
assert FakeTraceloop.associations[-1]["run_id"] == "renamed-run"
assert FakeTraceloop.associations[-1]["run_name"] == "renamed-run"
def test_events_write_locks_are_scoped_by_events_file(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
monkeypatch.chdir(tmp_path)
monkeypatch.setenv("STRIX_TELEMETRY", "0")
tracer_one = Tracer("lock-run-a")
tracer_two = Tracer("lock-run-b")
lock_a_from_one = tracer_one._get_events_write_lock(tracer_one.events_file_path)
lock_a_from_two = tracer_two._get_events_write_lock(tracer_one.events_file_path)
lock_b = tracer_two._get_events_write_lock(tracer_two.events_file_path)
assert lock_a_from_one is lock_a_from_two
assert lock_a_from_one is not lock_b
def test_tracer_skips_jsonl_when_telemetry_disabled(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
monkeypatch.chdir(tmp_path)
monkeypatch.setenv("STRIX_TELEMETRY", "0")
tracer = Tracer("telemetry-disabled")
set_global_tracer(tracer)
tracer.log_chat_message("hello", "assistant", "agent-1")
tracer.save_run_data(mark_complete=True)
events_path = tmp_path / "strix_runs" / "telemetry-disabled" / "events.jsonl"
assert not events_path.exists()
def test_tracer_otel_flag_overrides_global_telemetry(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
monkeypatch.chdir(tmp_path)
monkeypatch.setenv("STRIX_TELEMETRY", "0")
monkeypatch.setenv("STRIX_OTEL_TELEMETRY", "1")
tracer = Tracer("otel-enabled")
set_global_tracer(tracer)
tracer.log_chat_message("hello", "assistant", "agent-1")
tracer.save_run_data(mark_complete=True)
events_path = tmp_path / "strix_runs" / "otel-enabled" / "events.jsonl"
assert events_path.exists()
tests/telemetry/test_utils.py
-39
View File
@@ -1,39 +0,0 @@
from strix.telemetry.utils import prune_otel_span_attributes
def test_prune_otel_span_attributes_drops_high_volume_prompt_content() -> None:
attributes = {
"gen_ai.operation.name": "openai.chat",
"gen_ai.request.model": "gpt-5.2",
"gen_ai.prompt.0.role": "system",
"gen_ai.prompt.0.content": "a" * 20_000,
"gen_ai.completion.0.content": "b" * 10_000,
"llm.input_messages.0.content": "c" * 5_000,
"llm.output_messages.0.content": "d" * 5_000,
"llm.input": "x" * 3_000,
"llm.output": "y" * 3_000,
}
pruned = prune_otel_span_attributes(attributes)
assert "gen_ai.prompt.0.content" not in pruned
assert "gen_ai.completion.0.content" not in pruned
assert "llm.input_messages.0.content" not in pruned
assert "llm.output_messages.0.content" not in pruned
assert "llm.input" not in pruned
assert "llm.output" not in pruned
assert pruned["gen_ai.operation.name"] == "openai.chat"
assert pruned["gen_ai.prompt.0.role"] == "system"
assert pruned["strix.filtered_attributes_count"] == 6
def test_prune_otel_span_attributes_keeps_metadata_when_nothing_is_dropped() -> None:
attributes = {
"gen_ai.operation.name": "openai.chat",
"gen_ai.request.model": "gpt-5.2",
"gen_ai.prompt.0.role": "user",
}
pruned = prune_otel_span_attributes(attributes)
assert pruned == attributes
tests/test_entry.py
-319
View File
@@ -1,319 +0,0 @@
"""Phase 5 tests for the top-level SDK scan entry point.
We never spin up a real Docker container or hit a real LLM here. The
tests patch ``session_manager.create_or_reuse``, ``Runner.run``, and
the agent factory so we can verify the wiring shape:
- The bus is registered with a root agent before Runner.run.
- The context dict carries every field downstream code (tools, hooks,
filter) reads.
- The session manager's bundle flows through to the context (host
ports, bearer, sandbox session/client).
- ``cleanup_on_exit=True`` always cleans up, even when Runner.run
raises.
- ``cleanup_on_exit=False`` preserves the cached session.
- Cancellation propagates: if Runner.run raises, descendants are
cancelled before re-raising.
"""
from __future__ import annotations
from pathlib import Path
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from strix.entry import _build_root_task, _build_scope_context, run_strix_scan
from strix.orchestration.bus import AgentMessageBus
# --- helpers ------------------------------------------------------------
def _bundle_for_test() -> dict[str, Any]:
return {
"client": MagicMock(name="docker_client"),
"session": MagicMock(name="sandbox_session"),
"capability": MagicMock(),
"tool_server_host_port": 12001,
"caido_host_port": 12002,
"bearer": "test-bearer-token-1234567890",
}
def _scan_config(**overrides: Any) -> dict[str, Any]:
base: dict[str, Any] = {
"targets": [
{
"type": "web_application",
"details": {"target_url": "https://example.com"},
},
],
"user_instructions": "find xss",
"scan_mode": "deep",
"is_whitebox": False,
}
base.update(overrides)
return base
# --- task / scope builders ---------------------------------------------
def test_build_root_task_groups_targets_and_appends_instructions() -> None:
config = _scan_config(
targets=[
{
"type": "repository",
"details": {
"target_repo": "https://github.com/x/y",
"cloned_repo_path": "/tmp/y",
"workspace_subdir": "y",
},
},
{
"type": "ip_address",
"details": {"target_ip": "10.0.0.1"},
},
],
user_instructions="report only critical issues",
)
task = _build_root_task(config)
assert "Repositories:" in task
assert "https://github.com/x/y (available at: /workspace/y)" in task
assert "IP Addresses:" in task
assert "10.0.0.1" in task
assert "Special instructions: report only critical issues" in task
def test_build_root_task_renders_diff_scope_block() -> None:
config = _scan_config(
diff_scope={
"active": True,
"repos": [
{
"workspace_subdir": "service-x",
"analyzable_files_count": 7,
"deleted_files_count": 2,
},
],
},
)
task = _build_root_task(config)
assert "Scope Constraints:" in task
assert "service-x: 7 changed file(s)" in task
assert "service-x: 2 deleted file(s)" in task
def test_build_scope_context_marks_authorization_source() -> None:
config = _scan_config(
targets=[
{
"type": "web_application",
"details": {"target_url": "https://target.test"},
},
],
)
ctx = _build_scope_context(config)
assert ctx["scope_source"] == "system_scan_config"
assert ctx["authorization_source"] == "strix_platform_verified_targets"
assert ctx["user_instructions_do_not_expand_scope"] is True
assert ctx["authorized_targets"] == [
{"type": "web_application", "value": "https://target.test", "workspace_path": ""},
]
# --- run_strix_scan wiring ---------------------------------------------
@pytest.mark.asyncio
async def test_run_strix_scan_wires_context_and_calls_runner(tmp_path: Path) -> None:
"""End-to-end (mocked) — assert every downstream consumer of context
sees the bundle's bearer + host ports."""
bundle = _bundle_for_test()
captured_context: dict[str, Any] = {}
async def fake_runner_run(*args: Any, **kwargs: Any) -> Any:
captured_context.update(kwargs.get("context", {}))
return MagicMock(name="run_result")
with (
patch(
"strix.entry.session_manager.create_or_reuse",
new=AsyncMock(return_value=bundle),
) as create_mock,
patch(
"strix.entry.session_manager.cleanup",
new=AsyncMock(),
) as cleanup_mock,
patch("strix.entry.Runner.run", side_effect=fake_runner_run) as runner_mock,
# Stub the factory to avoid rendering the 158k-char prompt for
# every test (it's covered by sdk_prompt tests).
patch(
"strix.entry.build_strix_agent",
return_value=MagicMock(name="root_agent"),
) as factory_mock,
):
await run_strix_scan(
scan_config=_scan_config(),
scan_id="scan-test",
image="strix-sandbox:test",
sources_path=tmp_path,
)
# Session manager calls.
create_mock.assert_awaited_once()
create_args = create_mock.await_args
assert create_args is not None
assert create_args.args == ("scan-test",)
assert create_args.kwargs["image"] == "strix-sandbox:test"
assert create_args.kwargs["sources_path"] == tmp_path
cleanup_mock.assert_awaited_once_with("scan-test")
# Factory called with is_root=True.
factory_mock.assert_called_once()
assert factory_mock.call_args.kwargs["is_root"] is True
# Runner.run called once with the root agent.
assert runner_mock.call_count == 1
# Context shape passed into Runner.run.
assert captured_context["sandbox_session"] is bundle["session"]
assert captured_context["sandbox_client"] is bundle["client"]
assert captured_context["sandbox_token"] == bundle["bearer"]
assert captured_context["tool_server_host_port"] == bundle["tool_server_host_port"]
assert captured_context["caido_host_port"] == bundle["caido_host_port"]
# Bus is registered and root agent_id is populated.
bus = captured_context["bus"]
assert isinstance(bus, AgentMessageBus)
assert captured_context["agent_id"] in bus.statuses
assert bus.parent_of[captured_context["agent_id"]] is None
# Child factory wired through so create_agent works.
assert callable(captured_context["agent_factory"])
@pytest.mark.asyncio
async def test_run_strix_scan_cleans_up_on_runner_failure(tmp_path: Path) -> None:
"""If Runner.run raises, cleanup must still fire (the finally branch)."""
bundle = _bundle_for_test()
with (
patch(
"strix.entry.session_manager.create_or_reuse",
new=AsyncMock(return_value=bundle),
),
patch(
"strix.entry.session_manager.cleanup",
new=AsyncMock(),
) as cleanup_mock,
patch(
"strix.entry.Runner.run",
side_effect=RuntimeError("simulated LLM blow-up"),
),
patch("strix.entry.build_strix_agent", return_value=MagicMock()),
pytest.raises(RuntimeError, match="simulated LLM"),
):
await run_strix_scan(
scan_config=_scan_config(),
scan_id="scan-fail",
image="i",
sources_path=tmp_path,
)
cleanup_mock.assert_awaited_once_with("scan-fail")
@pytest.mark.asyncio
async def test_run_strix_scan_skips_cleanup_when_disabled(tmp_path: Path) -> None:
bundle = _bundle_for_test()
async def fake_runner_run(*args: Any, **kwargs: Any) -> Any:
return MagicMock()
with (
patch(
"strix.entry.session_manager.create_or_reuse",
new=AsyncMock(return_value=bundle),
),
patch(
"strix.entry.session_manager.cleanup",
new=AsyncMock(),
) as cleanup_mock,
patch("strix.entry.Runner.run", side_effect=fake_runner_run),
patch("strix.entry.build_strix_agent", return_value=MagicMock()),
):
await run_strix_scan(
scan_config=_scan_config(),
scan_id="scan-keep",
image="i",
sources_path=tmp_path,
cleanup_on_exit=False,
)
cleanup_mock.assert_not_called()
@pytest.mark.asyncio
async def test_run_strix_scan_auto_generates_scan_id(tmp_path: Path) -> None:
"""A caller without a stable id should still get a valid scan_id
flowing into create_or_reuse."""
bundle = _bundle_for_test()
captured_scan_id: list[str] = []
async def fake_create(scan_id: str, **_kwargs: Any) -> Any:
captured_scan_id.append(scan_id)
return bundle
with (
patch(
"strix.entry.session_manager.create_or_reuse",
new=AsyncMock(side_effect=fake_create),
),
patch("strix.entry.session_manager.cleanup", new=AsyncMock()),
patch("strix.entry.Runner.run", new=AsyncMock(return_value=MagicMock())),
patch("strix.entry.build_strix_agent", return_value=MagicMock()),
):
await run_strix_scan(
scan_config=_scan_config(),
image="i",
sources_path=tmp_path,
)
assert len(captured_scan_id) == 1
assert captured_scan_id[0].startswith("scan-")
assert len(captured_scan_id[0]) > len("scan-")
@pytest.mark.asyncio
async def test_run_strix_scan_passes_scan_level_config_into_factory(
tmp_path: Path,
) -> None:
"""scan_mode / is_whitebox flow from scan_config into both the
root factory call and the child factory closure."""
bundle = _bundle_for_test()
factory_calls: list[dict[str, Any]] = []
def fake_factory(**kwargs: Any) -> Any:
factory_calls.append(kwargs)
return MagicMock()
with (
patch(
"strix.entry.session_manager.create_or_reuse",
new=AsyncMock(return_value=bundle),
),
patch("strix.entry.session_manager.cleanup", new=AsyncMock()),
patch("strix.entry.Runner.run", new=AsyncMock(return_value=MagicMock())),
patch("strix.entry.build_strix_agent", side_effect=fake_factory),
):
await run_strix_scan(
scan_config=_scan_config(scan_mode="fast", is_whitebox=True),
scan_id="s",
image="i",
sources_path=tmp_path,
)
assert factory_calls[0]["scan_mode"] == "fast"
assert factory_calls[0]["is_whitebox"] is True
assert factory_calls[0]["is_root"] is True
tests/test_run_config_factory.py
-173
View File
@@ -1,173 +0,0 @@
"""Smoke tests for make_run_config / make_agent_context."""
from __future__ import annotations
from agents import RunConfig
from agents.model_settings import ModelSettings
from agents.retry import ModelRetryBackoffSettings
from strix.orchestration.bus import AgentMessageBus
from strix.run_config_factory import make_agent_context, make_run_config
def test_make_run_config_returns_run_config() -> None:
cfg = make_run_config(sandbox_session=None)
assert isinstance(cfg, RunConfig)
def test_default_parallel_tool_calls_is_false() -> None:
"""Default is sequential — the tool server serializes one task per agent."""
cfg = make_run_config(sandbox_session=None)
assert cfg.model_settings is not None
assert cfg.model_settings.parallel_tool_calls is False
def test_default_tool_choice_is_required() -> None:
cfg = make_run_config(sandbox_session=None)
assert cfg.model_settings is not None
assert cfg.model_settings.tool_choice == "required"
def test_call_model_input_filter_is_wired() -> None:
cfg = make_run_config(sandbox_session=None)
assert cfg.call_model_input_filter is not None
# Wired to inject_messages_filter (validated by name to keep import light).
assert cfg.call_model_input_filter.__name__ == "inject_messages_filter"
def test_retry_settings_have_max_retries_5() -> None:
cfg = make_run_config(sandbox_session=None)
assert cfg.model_settings is not None
retry = cfg.model_settings.retry
assert retry is not None
assert retry.max_retries == 5
def test_retry_backoff_uses_strix_defaults() -> None:
"""min(90, 2*2^n) with initial 2s, max 90s, x2."""
cfg = make_run_config(sandbox_session=None)
assert cfg.model_settings is not None
retry = cfg.model_settings.retry
assert retry is not None
backoff = retry.backoff
assert isinstance(backoff, ModelRetryBackoffSettings)
assert backoff.initial_delay == 2.0
assert backoff.max_delay == 90.0
assert backoff.multiplier == 2.0
def test_retry_policy_is_set() -> None:
"""Retry policy is wired (auth/validation 4xx excluded by construction)."""
cfg = make_run_config(sandbox_session=None)
assert cfg.model_settings is not None
retry = cfg.model_settings.retry
assert retry is not None
assert retry.policy is not None
def test_trace_include_sensitive_data_is_false() -> None:
cfg = make_run_config(sandbox_session=None)
assert cfg.trace_include_sensitive_data is False
def test_model_settings_override_merges() -> None:
"""Per-call override path."""
override = ModelSettings(tool_choice="auto", parallel_tool_calls=True)
cfg = make_run_config(sandbox_session=None, model_settings_override=override)
assert cfg.model_settings is not None
assert cfg.model_settings.tool_choice == "auto"
assert cfg.model_settings.parallel_tool_calls is True
# Retry settings (not in override) preserved from base.
assert cfg.model_settings.retry is not None
assert cfg.model_settings.retry.max_retries == 5
def test_reasoning_effort_propagates() -> None:
cfg = make_run_config(sandbox_session=None, reasoning_effort="high")
assert cfg.model_settings is not None
assert cfg.model_settings.reasoning is not None
assert cfg.model_settings.reasoning.effort == "high"
def test_max_turns_default_is_300() -> None:
"""Default max_turns=300 in make_agent_context.
``max_turns`` itself is passed to ``Runner.run``; the context copy is
consumed by the budget-warning hook.
"""
bus = AgentMessageBus()
ctx = make_agent_context(
bus=bus,
sandbox_session=None,
sandbox_token=None,
tool_server_host_port=None,
caido_host_port=None,
agent_id="root",
parent_id=None,
tracer=None,
)
assert ctx["max_turns"] == 300
def test_make_agent_context_full_shape() -> None:
"""The context dict carries every field tools/hooks reach for."""
bus = AgentMessageBus()
ctx = make_agent_context(
bus=bus,
sandbox_session=None,
sandbox_token="bearer-xyz",
tool_server_host_port=48081,
caido_host_port=48080,
agent_id="agent-1",
parent_id=None,
tracer="not-a-real-tracer",
is_whitebox=True,
diff_scope={"changed_files": ["src/app.py"]},
run_id="strix_runs/abc_def",
)
assert ctx["bus"] is bus
assert ctx["agent_id"] == "agent-1"
assert ctx["parent_id"] is None
assert ctx["agent_finish_called"] is False
assert ctx["turn_count"] == 0
assert ctx["is_whitebox"] is True
assert ctx["diff_scope"] == {"changed_files": ["src/app.py"]}
assert ctx["run_id"] == "strix_runs/abc_def"
assert ctx["sandbox_token"] == "bearer-xyz"
assert ctx["tool_server_host_port"] == 48081
assert ctx["caido_host_port"] == 48080
def test_make_agent_context_is_whitebox_defaults_false() -> None:
bus = AgentMessageBus()
ctx = make_agent_context(
bus=bus,
sandbox_session=None,
sandbox_token=None,
tool_server_host_port=None,
caido_host_port=None,
agent_id="r",
parent_id=None,
tracer=None,
)
assert ctx["is_whitebox"] is False
assert ctx["diff_scope"] is None
def test_sandbox_config_omitted_when_no_session() -> None:
cfg = make_run_config(sandbox_session=None)
assert cfg.sandbox is None
def test_model_default_is_strix_claude() -> None:
cfg = make_run_config(sandbox_session=None)
assert cfg.model == "anthropic/claude-sonnet-4-6"
def test_multi_provider_is_built() -> None:
"""Verify the factory wires our custom MultiProvider, not the SDK default."""
cfg = make_run_config(sandbox_session=None)
# MultiProvider is opaque, but our build_multi_provider returns
# an instance with our prefix routes installed.
assert cfg.model_provider is not None
tests/tools/__init__.py
-1
View File
@@ -1 +0,0 @@
"""Tests for strix.tools module."""
tests/tools/conftest.py
-34
View File
@@ -1,34 +0,0 @@
"""Fixtures for strix.tools tests."""
from collections.abc import Callable
from typing import Any
import pytest
@pytest.fixture
def sample_function_with_types() -> Callable[..., None]:
"""Create a sample function with type annotations for testing argument conversion."""
def func(
name: str,
count: int,
enabled: bool,
ratio: float,
items: list[Any],
config: dict[str, Any],
optional: str | None = None,
) -> None:
pass
return func
@pytest.fixture
def sample_function_no_annotations() -> Callable[..., None]:
"""Create a sample function without type annotations."""
def func(arg1, arg2, arg3): # type: ignore[no-untyped-def]
pass
return func
tests/tools/test_decorator.py
-73
View File
@@ -1,73 +0,0 @@
"""Phase 0 smoke tests for the strix_tool decorator factory.
The SDK's ``FunctionTool`` only honors ``timeout_seconds`` for ``async``
handlers (verified at ``agents.tool._validate_function_tool_timeout_config``):
sync function bodies cannot be cleanly cancelled, so the SDK refuses to
attach a timeout to them. Every Strix tool is therefore an ``async def``;
sync libraries (libtmux, IPython) get wrapped in ``asyncio.to_thread``
inside the async tool body.
"""
from __future__ import annotations
import pytest
from agents.tool import FunctionTool
from strix.tools._decorator import strix_tool
def test_strix_tool_returns_function_tool() -> None:
@strix_tool()
async def my_tool(x: int) -> str:
"""Do a thing."""
return str(x)
assert isinstance(my_tool, FunctionTool)
def test_strix_tool_default_timeout_is_120s() -> None:
@strix_tool()
async def my_tool(x: int) -> str:
"""Do a thing."""
return str(x)
assert my_tool.timeout_seconds == 120.0
def test_strix_tool_default_timeout_behavior_is_error_as_result() -> None:
@strix_tool()
async def my_tool(x: int) -> str:
"""Do a thing."""
return str(x)
assert my_tool.timeout_behavior == "error_as_result"
def test_strix_tool_timeout_override() -> None:
@strix_tool(timeout=300)
async def my_tool(x: int) -> str:
"""Do a thing."""
return str(x)
assert my_tool.timeout_seconds == 300.0
def test_strix_tool_critical_tool_can_raise() -> None:
"""C20 (AUDIT_R3): critical tools opt into raise_exception."""
@strix_tool(timeout=30, timeout_behavior="raise_exception")
async def critical_tool(x: int) -> str:
"""Do a critical thing."""
return str(x)
assert critical_tool.timeout_behavior == "raise_exception"
def test_strix_tool_sync_handlers_rejected_by_sdk() -> None:
"""SDK explicitly rejects timeout on sync handlers; documenting the constraint."""
with pytest.raises(ValueError, match="async @function_tool"):
@strix_tool()
def my_sync_tool(x: int) -> str:
"""Sync tools can't have a timeout in the SDK."""
return str(x)
tests/tools/test_graph_tools.py
-514
View File
@@ -1,514 +0,0 @@
"""Phase 3 tests for the multi-agent graph SDK tools.
Six tools: view_agent_graph, agent_status, send_message_to_agent,
wait_for_message, create_agent, agent_finish.
Strategy:
- Build a real ``AgentMessageBus`` in each test and put it under
``ctx.context['bus']`` so the tools exercise the same code path
production runs do.
- ``create_agent`` is the only tool that touches ``Runner.run`` and
``asyncio.create_task``. Its test injects a stub agent factory and
patches the SDK's ``Runner.run`` to a sentinel coroutine — we verify
the spawn shape (bus.register called, task handle stored, identity
block in initial input) without spinning up a real LLM.
- ``wait_for_message`` is exercised on both branches: a message arrives
mid-poll, and the timeout path.
"""
from __future__ import annotations
import asyncio
import json
from dataclasses import dataclass, field
from typing import Any
from unittest.mock import AsyncMock, patch
import pytest
from agents.tool import FunctionTool
from strix.orchestration.bus import AgentMessageBus
from strix.tools.agents_graph.tools import (
agent_finish,
agent_status,
create_agent,
send_message_to_agent,
view_agent_graph,
wait_for_message,
)
@dataclass
class _Ctx:
context: dict[str, Any] = field(default_factory=dict)
async def _invoke(tool: FunctionTool, ctx: _Ctx, **kwargs: Any) -> dict[str, Any]:
from agents.tool_context import ToolContext
tool_ctx = ToolContext(
context=ctx.context,
usage=None,
tool_name=tool.name,
tool_call_id="test-call-id",
tool_arguments=json.dumps(kwargs),
)
result = await tool.on_invoke_tool(tool_ctx, json.dumps(kwargs))
assert isinstance(result, str)
decoded = json.loads(result)
assert isinstance(decoded, dict)
return decoded
async def _make_bus_with_agents() -> AgentMessageBus:
"""Bus prepopulated with a root + two registered children."""
bus = AgentMessageBus()
await bus.register("root-1", "root", parent_id=None)
await bus.register("child-A", "scanner", parent_id="root-1")
await bus.register("child-B", "exploiter", parent_id="root-1")
return bus
def _ctx_for(bus: AgentMessageBus, agent_id: str = "root-1") -> _Ctx:
return _Ctx(context={"bus": bus, "agent_id": agent_id, "parent_id": None})
# --- registration ---------------------------------------------------------
def test_all_graph_tools_are_function_tools() -> None:
for tool in (
view_agent_graph,
agent_status,
send_message_to_agent,
wait_for_message,
create_agent,
agent_finish,
):
assert isinstance(tool, FunctionTool)
# --- view_agent_graph -----------------------------------------------------
@pytest.mark.asyncio
async def test_view_agent_graph_renders_tree() -> None:
bus = await _make_bus_with_agents()
out = await _invoke(view_agent_graph, _ctx_for(bus))
assert out["success"] is True
assert "root (root-1)" in out["graph_structure"]
assert "scanner (child-A)" in out["graph_structure"]
assert "exploiter (child-B)" in out["graph_structure"]
# The "you" marker should be on the calling agent's line.
you_lines = [line for line in out["graph_structure"].splitlines() if "← you" in line]
assert len(you_lines) == 1
assert "root-1" in you_lines[0]
assert out["summary"]["total"] == 3
assert out["summary"]["running"] == 3
@pytest.mark.asyncio
async def test_view_agent_graph_handles_missing_bus() -> None:
out = await _invoke(view_agent_graph, _Ctx(context={}))
assert out["success"] is False
assert "Bus" in out["error"]
# --- agent_status ---------------------------------------------------------
@pytest.mark.asyncio
async def test_agent_status_returns_state() -> None:
bus = await _make_bus_with_agents()
out = await _invoke(agent_status, _ctx_for(bus), agent_id="child-A")
assert out["success"] is True
assert out["agent_id"] == "child-A"
assert out["name"] == "scanner"
assert out["status"] == "running"
assert out["parent_id"] == "root-1"
assert out["pending_messages"] == 0
@pytest.mark.asyncio
async def test_agent_status_unknown_id() -> None:
bus = await _make_bus_with_agents()
out = await _invoke(agent_status, _ctx_for(bus), agent_id="nope")
assert out["success"] is False
assert "Unknown" in out["error"]
# --- send_message_to_agent -----------------------------------------------
@pytest.mark.asyncio
async def test_send_message_queues_into_target_inbox() -> None:
bus = await _make_bus_with_agents()
out = await _invoke(
send_message_to_agent,
_ctx_for(bus, agent_id="child-A"),
target_agent_id="child-B",
message="hello sibling",
priority="high",
)
assert out["success"] is True
assert out["delivery_status"] == "queued"
# Drain via the same API the filter uses; confirm the message landed.
msgs = await bus.drain("child-B")
assert len(msgs) == 1
assert msgs[0]["from"] == "child-A"
assert msgs[0]["content"] == "hello sibling"
assert msgs[0]["priority"] == "high"
@pytest.mark.asyncio
async def test_send_message_unknown_target() -> None:
bus = await _make_bus_with_agents()
out = await _invoke(
send_message_to_agent,
_ctx_for(bus),
target_agent_id="ghost",
message="hi",
)
assert out["success"] is False
assert "not found" in out["error"]
@pytest.mark.asyncio
async def test_send_message_to_finalized_agent_is_rejected() -> None:
"""A finalized target should not silently swallow messages."""
bus = await _make_bus_with_agents()
await bus.finalize("child-A", "completed")
out = await _invoke(
send_message_to_agent,
_ctx_for(bus),
target_agent_id="child-A",
message="too late",
)
assert out["success"] is False
# finalize() also clears parent_of/names, so the user-visible state is
# "completed" — confirm the wrapper treats finalized agents as
# undeliverable rather than queuing into a dropped inbox.
assert "completed" in out["error"] or "not found" in out["error"]
# --- wait_for_message ----------------------------------------------------
@pytest.mark.asyncio
async def test_wait_for_message_returns_when_message_arrives() -> None:
bus = await _make_bus_with_agents()
async def deliver_after_short_pause() -> None:
await asyncio.sleep(0.1)
await bus.send("child-A", {"from": "child-B", "content": "ping", "type": "info"})
sender = asyncio.create_task(deliver_after_short_pause())
out = await _invoke(
wait_for_message,
_ctx_for(bus, agent_id="child-A"),
timeout_seconds=3,
)
await sender
assert out["success"] is True
assert out["status"] == "message_arrived"
assert out["pending_messages"] >= 1
# Status must be returned to "running" after the wait completes.
assert bus.statuses["child-A"] == "running"
@pytest.mark.asyncio
async def test_wait_for_message_times_out() -> None:
bus = await _make_bus_with_agents()
out = await _invoke(
wait_for_message,
_ctx_for(bus, agent_id="child-A"),
timeout_seconds=1,
)
assert out["success"] is True
assert out["status"] == "timeout"
assert out["timeout_seconds"] == 1
assert bus.statuses["child-A"] == "running"
# --- create_agent --------------------------------------------------------
@pytest.mark.asyncio
async def test_create_agent_requires_factory_in_context() -> None:
bus = await _make_bus_with_agents()
out = await _invoke(
create_agent,
_ctx_for(bus),
name="recon-bot",
task="enumerate hosts",
)
assert out["success"] is False
assert "agent_factory" in out["error"]
@pytest.mark.asyncio
async def test_create_agent_spawns_and_registers_child() -> None:
"""Verify the spawn shape without running a real LLM."""
bus = await _make_bus_with_agents()
factory_calls: list[dict[str, Any]] = []
def fake_factory(*, name: str, skills: list[str]) -> Any:
factory_calls.append({"name": name, "skills": list(skills)})
# The Runner.run patch below ignores this object; any sentinel
# works.
return object()
runner_calls: list[dict[str, Any]] = []
async def fake_runner_run(*args: Any, **kwargs: Any) -> Any:
# Capture the input + max_turns so the test can assert the
# identity block + delegation envelope are present.
runner_calls.append({"args": args, "kwargs": kwargs})
await asyncio.sleep(0) # cooperate so create_task can return
return None
ctx = _Ctx(
context={
"bus": bus,
"agent_id": "root-1",
"parent_id": None,
"agent_factory": fake_factory,
"sandbox_session": None,
"sandbox_client": None,
"sandbox_token": "token",
"tool_server_host_port": 12345,
"caido_host_port": None,
"tracer": None,
"model": "anthropic/claude-sonnet-4-6",
"model_settings": None,
"max_turns": 300,
"is_whitebox": False,
}
)
with patch(
"strix.tools.agents_graph.tools.Runner.run",
side_effect=fake_runner_run,
):
out = await _invoke(
create_agent,
ctx,
name="recon-bot",
task="enumerate hosts",
inherit_context=False,
skills=["recon"],
)
# The spawned task must be allowed to run so Runner.run side-effect
# records the call.
new_id = out["agent_id"]
await asyncio.gather(*(t for t in bus.tasks.values()), return_exceptions=True)
assert out["success"] is True
assert factory_calls == [{"name": "recon-bot", "skills": ["recon"]}]
assert len(runner_calls) == 1
# Bus state: child registered, task stored.
assert new_id in bus.statuses
assert bus.parent_of[new_id] == "root-1"
assert bus.names[new_id] == "recon-bot"
assert new_id in bus.tasks
# Initial input shape: identity preamble + task message at the end.
initial_input = runner_calls[0]["kwargs"]["input"]
assert any(
isinstance(item, dict) and "You are agent recon-bot" in item.get("content", "")
for item in initial_input
)
assert initial_input[-1]["content"] == "enumerate hosts"
@pytest.mark.asyncio
async def test_create_agent_inherits_parent_history() -> None:
bus = await _make_bus_with_agents()
def fake_factory(*, name: str, skills: list[str]) -> Any:
return object()
runner_calls: list[dict[str, Any]] = []
async def fake_runner_run(*args: Any, **kwargs: Any) -> Any:
runner_calls.append(kwargs)
return None
parent_history = [
{"role": "user", "content": "scope: example.com"},
{"role": "assistant", "content": "I'll start with subdomain enum."},
]
ctx = _Ctx(
context={
"bus": bus,
"agent_id": "root-1",
"parent_id": None,
"agent_factory": fake_factory,
"parent_input_items": parent_history,
"sandbox_session": None,
"sandbox_client": None,
"sandbox_token": "token",
"tool_server_host_port": 12345,
"caido_host_port": None,
"tracer": None,
"model": "anthropic/claude-sonnet-4-6",
"model_settings": None,
"max_turns": 300,
}
)
with patch(
"strix.tools.agents_graph.tools.Runner.run",
side_effect=fake_runner_run,
):
await _invoke(
create_agent,
ctx,
name="child",
task="do thing",
inherit_context=True,
)
await asyncio.gather(*(t for t in bus.tasks.values()), return_exceptions=True)
initial_input = runner_calls[0]["input"]
contents = [item.get("content", "") for item in initial_input]
assert any("Inherited context from parent" in c for c in contents)
assert any("End of inherited context" in c for c in contents)
# Parent's exact items should be in between.
assert any(c == "scope: example.com" for c in contents)
# --- agent_finish --------------------------------------------------------
@pytest.mark.asyncio
async def test_agent_finish_rejects_root() -> None:
bus = await _make_bus_with_agents()
out = await _invoke(
agent_finish,
_ctx_for(bus, agent_id="root-1"),
result_summary="done",
)
assert out["success"] is False
assert "agent_finish is for subagents" in out["error"]
@pytest.mark.asyncio
async def test_agent_finish_posts_report_to_parent_inbox() -> None:
bus = await _make_bus_with_agents()
ctx = _Ctx(
context={
"bus": bus,
"agent_id": "child-A",
"parent_id": "root-1",
"agent_finish_called": False,
}
)
out = await _invoke(
agent_finish,
ctx,
result_summary="found 3 issues",
findings=["xss in /search", "open redirect", "stored xss"],
final_recommendations=["sanitize search input"],
success=True,
)
assert out["success"] is True
assert out["agent_completed"] is True
assert out["parent_notified"] is True
# Side effects: agent_finish_called flipped (so on_agent_end records
# "completed", not "crashed"), and the parent's inbox got the report.
assert ctx.context["agent_finish_called"] is True
parent_msgs = await bus.drain("root-1")
assert len(parent_msgs) == 1
msg = parent_msgs[0]
assert msg["type"] == "completion"
assert msg["from"] == "child-A"
payload = json.loads(msg["content"])
assert payload["kind"] == "agent_completion_report"
assert payload["summary"] == "found 3 issues"
assert "xss in /search" in payload["findings"]
assert "sanitize search input" in payload["recommendations"]
@pytest.mark.asyncio
async def test_agent_finish_skips_parent_when_report_to_parent_false() -> None:
bus = await _make_bus_with_agents()
ctx = _Ctx(
context={
"bus": bus,
"agent_id": "child-A",
"parent_id": "root-1",
"agent_finish_called": False,
}
)
out = await _invoke(
agent_finish,
ctx,
result_summary="silent done",
report_to_parent=False,
)
assert out["success"] is True
assert out["parent_notified"] is False
assert ctx.context["agent_finish_called"] is True
parent_msgs = await bus.drain("root-1")
assert parent_msgs == []
# --- bus integration sanity ---------------------------------------------
@pytest.mark.asyncio
async def test_create_agent_spawn_is_cancelable_via_bus() -> None:
"""Verify bus.cancel_descendants reaches a child task we just spawned."""
bus = await _make_bus_with_agents()
def fake_factory(*, name: str, skills: list[str]) -> Any:
return object()
# Long-lived child that yields control so we can cancel it before it
# finishes naturally.
async def slow_runner_run(*args: Any, **kwargs: Any) -> Any:
await asyncio.sleep(60)
return None
ctx = _Ctx(
context={
"bus": bus,
"agent_id": "root-1",
"parent_id": None,
"agent_factory": fake_factory,
"sandbox_session": None,
"sandbox_client": None,
"sandbox_token": "t",
"tool_server_host_port": 12345,
"caido_host_port": None,
"tracer": None,
"model": "anthropic/claude-sonnet-4-6",
"model_settings": None,
"max_turns": 300,
}
)
runner_mock = AsyncMock(side_effect=slow_runner_run)
with patch(
"strix.tools.agents_graph.tools.Runner.run",
new=runner_mock,
):
out = await _invoke(
create_agent,
ctx,
name="long-running",
task="do thing",
inherit_context=False,
)
child_id = out["agent_id"]
# Let the task actually start.
await asyncio.sleep(0.05)
await bus.cancel_descendants("root-1")
# The cancel should have propagated; the task is done (cancelled).
assert bus.tasks[child_id].done()
tests/tools/test_local_tools.py
-250
View File
@@ -1,250 +0,0 @@
"""Phase 2.3 smoke tests for the simplest SDK-wrapped local tools.
Validates the wrapping pattern (legacy implementation in, JSON string out)
on three tool families: think (trivial), todo (in-memory + agent_state
adapter), notes (in-memory + JSONL persistence).
If this slice works end-to-end the same pattern carries the rest of the
local tools (reporting, web_search, file_edit, finish_scan, load_skill).
"""
from __future__ import annotations
import json
from collections.abc import Iterator
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
from unittest.mock import patch
import pytest
from agents.tool import FunctionTool
from strix.tools.notes import tools as _notes_impl
from strix.tools.notes.tools import (
create_note,
delete_note,
get_note,
list_notes,
update_note,
)
from strix.tools.thinking.tool import think
from strix.tools.todo.tools import (
create_todo,
delete_todo,
list_todos,
mark_todo_done,
mark_todo_pending,
update_todo,
)
@dataclass
class _Ctx:
"""Stand-in for ``RunContextWrapper``."""
context: dict[str, Any] = field(default_factory=dict)
def _ctx_for(agent_id: str = "test-agent") -> _Ctx:
return _Ctx(context={"agent_id": agent_id})
async def _invoke(tool: FunctionTool, ctx: _Ctx, **kwargs: Any) -> dict[str, Any]:
"""Invoke a function tool the way the SDK would and JSON-decode the result."""
from agents.tool_context import ToolContext
tool_ctx = ToolContext(
context=ctx.context,
usage=None,
tool_name=tool.name,
tool_call_id="test-call-id",
tool_arguments=json.dumps(kwargs),
)
result = await tool.on_invoke_tool(tool_ctx, json.dumps(kwargs))
assert isinstance(result, str)
decoded = json.loads(result)
assert isinstance(decoded, dict)
return decoded
# --- think ----------------------------------------------------------------
def test_think_is_a_function_tool() -> None:
assert isinstance(think, FunctionTool)
assert think.name == "think"
@pytest.mark.asyncio
async def test_think_records_thought() -> None:
ctx = _ctx_for()
thought = "planning my next move"
out = await _invoke(think, ctx, thought=thought)
assert out["success"] is True
assert f"{len(thought)} characters" in out["message"]
@pytest.mark.asyncio
async def test_think_rejects_empty() -> None:
ctx = _ctx_for()
out = await _invoke(think, ctx, thought=" ")
assert out["success"] is False
# --- todo -----------------------------------------------------------------
@pytest.fixture(autouse=True)
def _isolate_todo_storage() -> None:
"""Each test starts with an empty todo store so tests don't bleed."""
from strix.tools.todo import tools as todo_module
todo_module._todos_storage.clear()
def test_todo_tools_are_function_tools() -> None:
for tool in (
create_todo,
list_todos,
update_todo,
mark_todo_done,
mark_todo_pending,
delete_todo,
):
assert isinstance(tool, FunctionTool)
@pytest.mark.asyncio
async def test_todo_lifecycle() -> None:
ctx = _ctx_for("agent-A")
# Create
created = await _invoke(create_todo, ctx, title="audit endpoint", priority="high")
assert created["success"] is True
assert created["count"] == 1
todo_id = created["created"][0]["todo_id"]
# List
listed = await _invoke(list_todos, ctx)
assert listed["success"] is True
assert any(t["todo_id"] == todo_id for t in listed["todos"])
# Update
updated = await _invoke(update_todo, ctx, todo_id=todo_id, status="in_progress")
assert updated["success"] is True
# Mark done
done = await _invoke(mark_todo_done, ctx, todo_id=todo_id)
assert done["success"] is True
# Reset to pending
pending = await _invoke(mark_todo_pending, ctx, todo_id=todo_id)
assert pending["success"] is True
# Delete
deleted = await _invoke(delete_todo, ctx, todo_id=todo_id)
assert deleted["success"] is True
@pytest.mark.asyncio
async def test_todos_are_per_agent_isolated() -> None:
"""Two agents should have independent todo stores."""
ctx_a = _ctx_for("agent-A")
ctx_b = _ctx_for("agent-B")
await _invoke(create_todo, ctx_a, title="A's task")
await _invoke(create_todo, ctx_b, title="B's task")
list_a = await _invoke(list_todos, ctx_a)
list_b = await _invoke(list_todos, ctx_b)
titles_a = [t["title"] for t in list_a["todos"]]
titles_b = [t["title"] for t in list_b["todos"]]
assert titles_a == ["A's task"]
assert titles_b == ["B's task"]
@pytest.mark.asyncio
async def test_create_todo_bulk_via_json_string() -> None:
ctx = _ctx_for()
out = await _invoke(
create_todo,
ctx,
todos=json.dumps(
[
{"title": "t1", "priority": "high"},
{"title": "t2", "priority": "low"},
],
),
)
assert out["success"] is True
assert out["count"] == 2
# --- notes ----------------------------------------------------------------
@pytest.fixture
def notes_run_dir(tmp_path: Path) -> Iterator[Path]:
"""Point the legacy notes module at a fresh run dir per test."""
run_dir = tmp_path / "strix_runs" / "test"
run_dir.mkdir(parents=True)
_notes_impl._notes_storage.clear()
_notes_impl._loaded_notes_run_dir = None
with patch.object(_notes_impl, "_get_run_dir", return_value=run_dir):
yield run_dir
def test_notes_tools_are_function_tools() -> None:
for tool in (create_note, list_notes, get_note, update_note, delete_note):
assert isinstance(tool, FunctionTool)
@pytest.mark.asyncio
async def test_note_lifecycle(notes_run_dir: Path) -> None:
ctx = _ctx_for()
created = await _invoke(
create_note,
ctx,
title="SQLi at /login",
content="Form param `email` reflects into the WHERE clause.",
category="findings",
tags=["sqli", "auth"],
)
assert created["success"] is True
note_id = created["note_id"]
listed = await _invoke(list_notes, ctx, category="findings")
assert listed["success"] is True
assert listed["total_count"] == 1
fetched = await _invoke(get_note, ctx, note_id=note_id)
assert fetched["success"] is True
assert "WHERE clause" in fetched["note"]["content"]
updated = await _invoke(
update_note,
ctx,
note_id=note_id,
content="Confirmed boolean-blind SQLi.",
)
assert updated["success"] is True
deleted = await _invoke(delete_note, ctx, note_id=note_id)
assert deleted["success"] is True
@pytest.mark.asyncio
async def test_notes_jsonl_appended(notes_run_dir: Path) -> None:
"""Verify side effect: notes.jsonl receives one event per op."""
ctx = _ctx_for()
await _invoke(create_note, ctx, title="t", content="c", category="general")
jsonl = notes_run_dir / "notes" / "notes.jsonl"
assert jsonl.exists()
events = [json.loads(line) for line in jsonl.read_text().splitlines() if line]
assert events[0]["op"] == "create"
assert events[0]["note"]["title"] == "t"
tests/tools/test_notes_jsonl_concurrency.py
-79
View File
@@ -1,79 +0,0 @@
"""C6 regression test — concurrent notes JSONL writes must produce valid JSONL.
This test would fail before the C6 fix (AUDIT_R2 §1.1, applied in Phase 2.2):
the legacy ``_append_note_event`` opened the file and called ``f.write``
without holding ``_notes_lock``, so two threads writing simultaneously
could interleave bytes mid-line and corrupt the JSONL.
"""
from __future__ import annotations
import json
import threading
from collections.abc import Iterator
from pathlib import Path
from typing import Any
from unittest.mock import patch
import pytest
from strix.tools.notes.tools import _append_note_event
@pytest.fixture
def notes_path(tmp_path: Path) -> Iterator[Path]:
"""Point ``_get_notes_jsonl_path`` at a tmp file for the test."""
target = tmp_path / "notes" / "notes.jsonl"
target.parent.mkdir(parents=True, exist_ok=True)
with patch(
"strix.tools.notes.tools._get_notes_jsonl_path",
return_value=target,
):
yield target
def test_concurrent_note_writes_yield_valid_jsonl(notes_path: Path) -> None:
"""C6 fix: 50 threads x 20 events = 1000 lines, all valid JSON.
Without the lock, byte-level interleaving on the file produces
fragments like ``{"timesta{"timestamp"...`` that fail json.loads.
"""
def writer(thread_idx: int) -> None:
for i in range(20):
note: dict[str, Any] = {
"title": f"thread-{thread_idx}-note-{i}",
"content": "x" * 200, # non-trivial body to widen the race
"category": "general",
}
_append_note_event(
op="create",
note_id=f"t{thread_idx}-i{i}",
note=note,
)
threads = [threading.Thread(target=writer, args=(t,)) for t in range(50)]
for t in threads:
t.start()
for t in threads:
t.join()
lines = notes_path.read_text(encoding="utf-8").splitlines()
assert len(lines) == 1000, f"expected 1000 lines, got {len(lines)}"
for line in lines:
# raises if the line is malformed JSON
event = json.loads(line)
assert event["op"] == "create"
assert "note_id" in event
def test_single_writer_still_works(notes_path: Path) -> None:
"""Sanity: serial writes still produce a valid JSONL log."""
_append_note_event("create", "n1", {"title": "first"})
_append_note_event("update", "n1", {"title": "first updated"})
_append_note_event("delete", "n1")
events = [json.loads(line) for line in notes_path.read_text().splitlines()]
assert [e["op"] for e in events] == ["create", "update", "delete"]
assert all(e["note_id"] == "n1" for e in events)
tests/tools/test_notes_wiki.py
-187
View File
@@ -1,187 +0,0 @@
from pathlib import Path
import pytest
from strix.telemetry.tracer import Tracer, get_global_tracer, set_global_tracer
from strix.tools.notes import tools as notes_actions
def _reset_notes_state() -> None:
notes_actions._notes_storage.clear()
notes_actions._loaded_notes_run_dir = None
def test_wiki_notes_are_persisted_and_removed(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
_reset_notes_state()
previous_tracer = get_global_tracer()
tracer = Tracer("wiki-test-run")
set_global_tracer(tracer)
try:
created = notes_actions._create_note_impl(
title="Repo Map",
content="## Architecture\n- monolith",
category="wiki",
tags=["source-map"],
)
assert created["success"] is True
note_id = created["note_id"]
assert isinstance(note_id, str)
note = notes_actions._notes_storage[note_id]
wiki_filename = note.get("wiki_filename")
assert isinstance(wiki_filename, str)
wiki_path = tmp_path / "strix_runs" / "wiki-test-run" / "wiki" / wiki_filename
assert wiki_path.exists()
assert "## Architecture" in wiki_path.read_text(encoding="utf-8")
updated = notes_actions._update_note_impl(
note_id=note_id,
content="## Architecture\n- service-oriented",
)
assert updated["success"] is True
assert "service-oriented" in wiki_path.read_text(encoding="utf-8")
deleted = notes_actions._delete_note_impl(note_id=note_id)
assert deleted["success"] is True
assert wiki_path.exists() is False
finally:
_reset_notes_state()
set_global_tracer(previous_tracer)
def test_notes_jsonl_replay_survives_memory_reset(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
_reset_notes_state()
previous_tracer = get_global_tracer()
tracer = Tracer("notes-replay-run")
set_global_tracer(tracer)
try:
created = notes_actions._create_note_impl(
title="Auth findings",
content="initial finding",
category="findings",
tags=["auth"],
)
assert created["success"] is True
note_id = created["note_id"]
assert isinstance(note_id, str)
notes_path = tmp_path / "strix_runs" / "notes-replay-run" / "notes" / "notes.jsonl"
assert notes_path.exists() is True
_reset_notes_state()
listed = notes_actions._list_notes_impl(category="findings")
assert listed["success"] is True
assert listed["total_count"] == 1
assert listed["notes"][0]["note_id"] == note_id
assert "content" not in listed["notes"][0]
assert "content_preview" in listed["notes"][0]
updated = notes_actions._update_note_impl(note_id=note_id, content="updated finding")
assert updated["success"] is True
_reset_notes_state()
listed_after_update = notes_actions._list_notes_impl(search="updated finding")
assert listed_after_update["success"] is True
assert listed_after_update["total_count"] == 1
assert listed_after_update["notes"][0]["note_id"] == note_id
assert listed_after_update["notes"][0]["content_preview"] == "updated finding"
listed_with_content = notes_actions._list_notes_impl(
category="findings",
include_content=True,
)
assert listed_with_content["success"] is True
assert listed_with_content["total_count"] == 1
assert listed_with_content["notes"][0]["content"] == "updated finding"
deleted = notes_actions._delete_note_impl(note_id=note_id)
assert deleted["success"] is True
_reset_notes_state()
listed_after_delete = notes_actions._list_notes_impl(category="findings")
assert listed_after_delete["success"] is True
assert listed_after_delete["total_count"] == 0
finally:
_reset_notes_state()
set_global_tracer(previous_tracer)
def test_get_note_returns_full_note(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.chdir(tmp_path)
_reset_notes_state()
previous_tracer = get_global_tracer()
tracer = Tracer("get-note-run")
set_global_tracer(tracer)
try:
created = notes_actions._create_note_impl(
title="Repo wiki",
content="entrypoints and sinks",
category="wiki",
tags=["repo:appsmith"],
)
assert created["success"] is True
note_id = created["note_id"]
assert isinstance(note_id, str)
result = notes_actions._get_note_impl(note_id=note_id)
assert result["success"] is True
assert result["note"]["note_id"] == note_id
assert result["note"]["content"] == "entrypoints and sinks"
finally:
_reset_notes_state()
set_global_tracer(previous_tracer)
def test_list_and_get_note_handle_wiki_repersist_oserror_gracefully(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
_reset_notes_state()
previous_tracer = get_global_tracer()
tracer = Tracer("wiki-repersist-oserror-run")
set_global_tracer(tracer)
try:
created = notes_actions._create_note_impl(
title="Repo wiki",
content="initial wiki content",
category="wiki",
tags=["repo:demo"],
)
assert created["success"] is True
note_id = created["note_id"]
assert isinstance(note_id, str)
_reset_notes_state()
def _raise_oserror(*_args: object, **_kwargs: object) -> None:
raise OSError("disk full")
monkeypatch.setattr(notes_actions, "_persist_wiki_note", _raise_oserror)
listed = notes_actions._list_notes_impl(category="wiki")
assert listed["success"] is True
assert listed["total_count"] == 1
assert listed["notes"][0]["note_id"] == note_id
fetched = notes_actions._get_note_impl(note_id=note_id)
assert fetched["success"] is True
assert fetched["note"]["note_id"] == note_id
assert fetched["note"]["content"] == "initial wiki content"
finally:
_reset_notes_state()
set_global_tracer(previous_tracer)
tests/tools/test_remaining_local_tools.py
-305
View File
@@ -1,305 +0,0 @@
"""Smoke tests for the remaining local SDK tool wrappers.
Covers: web_search, file_edit (str_replace_editor + list_files +
search_files), reporting (create_vulnerability_report), finish_scan.
"""
from __future__ import annotations
import json
from dataclasses import dataclass, field
from typing import Any
from unittest.mock import patch
import pytest
from agents.tool import FunctionTool
from strix.tools.file_edit.tools import (
list_files,
search_files,
str_replace_editor,
)
from strix.tools.finish.tool import finish_scan
from strix.tools.reporting.tool import create_vulnerability_report
from strix.tools.web_search.tool import web_search
@dataclass
class _Ctx:
context: dict[str, Any] = field(default_factory=dict)
def _ctx_for(agent_id: str = "test-agent") -> _Ctx:
return _Ctx(
context={
"agent_id": agent_id,
"tool_server_host_port": 12345,
"sandbox_token": "test-token",
},
)
async def _invoke(tool: FunctionTool, ctx: _Ctx, **kwargs: Any) -> dict[str, Any]:
from agents.tool_context import ToolContext
tool_ctx = ToolContext(
context=ctx.context,
usage=None,
tool_name=tool.name,
tool_call_id="test-call-id",
tool_arguments=json.dumps(kwargs),
)
result = await tool.on_invoke_tool(tool_ctx, json.dumps(kwargs))
assert isinstance(result, str)
decoded = json.loads(result)
assert isinstance(decoded, dict)
return decoded
def test_all_remaining_tools_are_function_tools() -> None:
for tool in (
web_search,
str_replace_editor,
list_files,
search_files,
create_vulnerability_report,
finish_scan,
):
assert isinstance(tool, FunctionTool)
# --- web_search -----------------------------------------------------------
@pytest.mark.asyncio
async def test_web_search_no_api_key_returns_structured_error(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""The legacy function returns a structured error when the env var is
missing — verify the wrapper passes that through verbatim."""
monkeypatch.delenv("PERPLEXITY_API_KEY", raising=False)
out = await _invoke(web_search, _ctx_for(), query="anything")
assert out["success"] is False
assert "PERPLEXITY_API_KEY" in out["message"]
@pytest.mark.asyncio
async def test_web_search_delegates_to_perplexity(monkeypatch: pytest.MonkeyPatch) -> None:
"""The wrapper invokes the Perplexity HTTP path on a thread."""
monkeypatch.setenv("PERPLEXITY_API_KEY", "fake-key")
fake_result = {
"success": True,
"query": "xss techniques",
"content": "Reflected XSS payload examples...",
"message": "Web search completed successfully",
}
with patch(
"strix.tools.web_search.tool._do_search",
return_value=fake_result,
) as do_search:
out = await _invoke(web_search, _ctx_for(), query="xss techniques")
assert out == fake_result
do_search.assert_called_once_with("xss techniques")
# --- file_edit (sandbox-bound) -------------------------------------------
@pytest.mark.asyncio
async def test_str_replace_editor_routes_to_sandbox() -> None:
"""file_edit tools must POST to the in-sandbox tool server, not run locally."""
fake_response = {"result": {"content": "file viewed"}}
with patch(
"strix.tools.file_edit.tools.post_to_sandbox",
return_value=fake_response,
) as dispatch:
out = await _invoke(
str_replace_editor,
_ctx_for(),
command="view",
path="src/foo.py",
)
assert out == fake_response
assert dispatch.call_count == 1
# post_to_sandbox is called positionally as (ctx, tool_name, kwargs).
args, _ = dispatch.call_args
assert args[1] == "str_replace_editor"
assert args[2]["command"] == "view"
assert args[2]["path"] == "src/foo.py"
# All optional file-edit params are forwarded as None (parity with legacy schema).
assert args[2]["file_text"] is None
assert args[2]["old_str"] is None
@pytest.mark.asyncio
async def test_list_files_routes_to_sandbox() -> None:
fake_response = {"result": {"files": ["a.py"], "directories": []}}
with patch(
"strix.tools.file_edit.tools.post_to_sandbox",
return_value=fake_response,
) as dispatch:
out = await _invoke(list_files, _ctx_for(), path="src", recursive=True)
assert out == fake_response
args, _ = dispatch.call_args
assert args[1] == "list_files"
assert args[2] == {"path": "src", "recursive": True}
@pytest.mark.asyncio
async def test_search_files_routes_to_sandbox() -> None:
fake_response = {"result": {"output": "src/foo.py:1:match"}}
with patch(
"strix.tools.file_edit.tools.post_to_sandbox",
return_value=fake_response,
) as dispatch:
out = await _invoke(
search_files,
_ctx_for(),
path="src",
regex="TODO",
file_pattern="*.py",
)
assert out == fake_response
args, _ = dispatch.call_args
assert args[1] == "search_files"
assert args[2] == {"path": "src", "regex": "TODO", "file_pattern": "*.py"}
# --- reporting -----------------------------------------------------------
@pytest.mark.asyncio
async def test_create_vulnerability_report_validates_required_fields() -> None:
"""Empty required fields should be rejected by the validator."""
out = await _invoke(
create_vulnerability_report,
_ctx_for(),
title="", # empty -> validation error
description="d",
impact="i",
target="t",
technical_analysis="ta",
poc_description="pd",
poc_script_code="curl ...",
remediation_steps="rs",
cvss_breakdown={"attack_vector": "N"},
)
assert out["success"] is False
assert "errors" in out
assert any("Title" in e for e in out["errors"])
@pytest.mark.asyncio
async def test_create_vulnerability_report_delegates_to_impl() -> None:
"""Verify the wrapper threads its kwargs through to the implementation."""
fake_result = {
"success": True,
"message": "Vulnerability report 'X' created successfully",
"report_id": "abc123",
"severity": "high",
"cvss_score": 7.5,
}
with patch(
"strix.tools.reporting.tool._do_create",
return_value=fake_result,
) as do_create:
out = await _invoke(
create_vulnerability_report,
_ctx_for(),
title="t",
description="d",
impact="i",
target="tg",
technical_analysis="ta",
poc_description="pd",
poc_script_code="pc",
remediation_steps="rs",
cvss_breakdown={"attack_vector": "N"},
cve="CVE-2024-12345",
)
assert out == fake_result
kwargs = do_create.call_args.kwargs
assert kwargs["title"] == "t"
assert kwargs["cve"] == "CVE-2024-12345"
# Optional params we didn't pass should still be forwarded as None.
assert kwargs["endpoint"] is None
assert kwargs["method"] is None
# --- finish_scan ---------------------------------------------------------
@pytest.mark.asyncio
async def test_finish_scan_validates_empty_fields() -> None:
"""Legacy validation: every section must be non-empty."""
out = await _invoke(
finish_scan,
_ctx_for(),
executive_summary="",
methodology="m",
technical_analysis="ta",
recommendations="r",
)
assert out["success"] is False
assert any("Executive summary" in e for e in out["errors"])
@pytest.mark.asyncio
async def test_finish_scan_rejects_subagent() -> None:
"""A subagent (parent_id is set) must not be able to finish the scan."""
ctx = _Ctx(
context={
"agent_id": "child-1",
"parent_id": "root-1",
"tool_server_host_port": 12345,
"sandbox_token": "test-token",
},
)
out = await _invoke(
finish_scan,
ctx,
executive_summary="es",
methodology="m",
technical_analysis="ta",
recommendations="r",
)
assert out["success"] is False
assert out["error"] == "finish_scan_wrong_agent"
@pytest.mark.asyncio
async def test_finish_scan_persists_via_tracer() -> None:
"""When a global tracer exists, finish_scan should write the four sections."""
from unittest.mock import MagicMock
fake_tracer = MagicMock()
fake_tracer.vulnerability_reports = [{}, {}, {}]
with patch(
"strix.telemetry.tracer.get_global_tracer",
return_value=fake_tracer,
):
out = await _invoke(
finish_scan,
_ctx_for("root-agent"),
executive_summary="es",
methodology="m",
technical_analysis="ta",
recommendations="r",
)
assert out["success"] is True
assert out["vulnerabilities_found"] == 3
fake_tracer.update_scan_final_fields.assert_called_once_with(
executive_summary="es",
methodology="m",
technical_analysis="ta",
recommendations="r",
)
tests/tools/test_sandbox_dispatch.py
-206
View File
@@ -1,206 +0,0 @@
"""Phase 2.1 smoke tests for the sandbox dispatch helper."""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
import httpx
import pytest
from strix.tools._sandbox_dispatch import post_to_sandbox
@dataclass
class _Ctx:
"""Stand-in for ``RunContextWrapper``. Only ``.context`` is touched."""
context: dict[str, Any] = field(default_factory=dict)
def _ok_ctx(**overrides: Any) -> _Ctx:
base: dict[str, Any] = {
"tool_server_host_port": 48081,
"sandbox_token": "test-bearer",
"agent_id": "agent-1",
}
base.update(overrides)
return _Ctx(context=base)
@pytest.mark.asyncio
async def test_missing_context_returns_error() -> None:
"""If ``ctx.context`` isn't a dict, return error — never raise."""
ctx = _Ctx(context="not a dict")
result = await post_to_sandbox(ctx, "browser_action", {"action": "launch"})
assert "error" in result
assert "context" in result["error"].lower()
@pytest.mark.asyncio
async def test_missing_port_or_token_returns_error() -> None:
ctx = _Ctx(context={"sandbox_token": "tok"}) # no port
result = await post_to_sandbox(ctx, "x", {})
assert "error" in result
assert "tool server port" in result["error"].lower()
@pytest.mark.asyncio
async def test_successful_response_returned_as_dict(
monkeypatch: pytest.MonkeyPatch,
) -> None:
captured: dict[str, Any] = {}
async def fake_post(
self: httpx.AsyncClient,
url: str,
*,
json: dict[str, Any] | None = None,
headers: dict[str, str] | None = None,
) -> httpx.Response:
captured["url"] = url
captured["json"] = json
captured["headers"] = headers
return httpx.Response(
status_code=200,
json={"result": "ok"},
request=httpx.Request("POST", url),
)
monkeypatch.setattr(httpx.AsyncClient, "post", fake_post)
result = await post_to_sandbox(_ok_ctx(), "terminal_execute", {"command": "ls"})
assert result == {"result": "ok"}
assert captured["url"] == "http://127.0.0.1:48081/execute"
assert captured["json"] == {
"agent_id": "agent-1",
"tool_name": "terminal_execute",
"kwargs": {"command": "ls"},
}
assert captured["headers"]["Authorization"] == "Bearer test-bearer"
@pytest.mark.asyncio
async def test_401_returns_auth_error(monkeypatch: pytest.MonkeyPatch) -> None:
async def fake_post(
self: httpx.AsyncClient,
url: str,
**_: Any,
) -> httpx.Response:
return httpx.Response(
status_code=401,
text="forbidden",
request=httpx.Request("POST", url),
)
monkeypatch.setattr(httpx.AsyncClient, "post", fake_post)
result = await post_to_sandbox(_ok_ctx(), "x", {})
assert "error" in result
assert "authorization" in result["error"].lower()
@pytest.mark.asyncio
async def test_5xx_returns_http_error(monkeypatch: pytest.MonkeyPatch) -> None:
async def fake_post(
self: httpx.AsyncClient,
url: str,
**_: Any,
) -> httpx.Response:
return httpx.Response(
status_code=503,
text="server fell over",
request=httpx.Request("POST", url),
)
monkeypatch.setattr(httpx.AsyncClient, "post", fake_post)
result = await post_to_sandbox(_ok_ctx(), "browser_action", {})
assert "error" in result
assert "503" in result["error"]
@pytest.mark.asyncio
async def test_timeout_returns_timeout_error(monkeypatch: pytest.MonkeyPatch) -> None:
async def fake_post(*_args: Any, **_kwargs: Any) -> httpx.Response:
raise httpx.ReadTimeout("read timeout")
monkeypatch.setattr(httpx.AsyncClient, "post", fake_post)
result = await post_to_sandbox(_ok_ctx(), "python_action", {})
assert "error" in result
assert "timed out" in result["error"].lower()
@pytest.mark.asyncio
async def test_connection_error_returns_error(monkeypatch: pytest.MonkeyPatch) -> None:
async def fake_post(*_args: Any, **_kwargs: Any) -> httpx.Response:
raise httpx.ConnectError("refused")
monkeypatch.setattr(httpx.AsyncClient, "post", fake_post)
result = await post_to_sandbox(_ok_ctx(), "x", {})
assert "error" in result
assert "connection" in result["error"].lower()
@pytest.mark.asyncio
async def test_response_too_large_capped(monkeypatch: pytest.MonkeyPatch) -> None:
"""C18 (AUDIT_R3): response > 50MB returns error, doesn't OOM."""
async def fake_post(
self: httpx.AsyncClient,
url: str,
**_: Any,
) -> httpx.Response:
# Construct a response with a >50MB body. Use a string trick —
# httpx.Response stores .content directly; we make it look huge.
big = b"x" * (51 * 1024 * 1024)
return httpx.Response(
status_code=200,
content=big,
request=httpx.Request("POST", url),
)
monkeypatch.setattr(httpx.AsyncClient, "post", fake_post)
result = await post_to_sandbox(_ok_ctx(), "browser_action", {})
assert "error" in result
assert "too large" in result["error"].lower()
@pytest.mark.asyncio
async def test_non_json_response_returns_error(
monkeypatch: pytest.MonkeyPatch,
) -> None:
async def fake_post(
self: httpx.AsyncClient,
url: str,
**_: Any,
) -> httpx.Response:
return httpx.Response(
status_code=200,
text="hello not json",
request=httpx.Request("POST", url),
)
monkeypatch.setattr(httpx.AsyncClient, "post", fake_post)
result = await post_to_sandbox(_ok_ctx(), "x", {})
assert "error" in result
assert "non-json" in result["error"].lower()
@pytest.mark.asyncio
async def test_non_object_json_returns_error(
monkeypatch: pytest.MonkeyPatch,
) -> None:
async def fake_post(
self: httpx.AsyncClient,
url: str,
**_: Any,
) -> httpx.Response:
return httpx.Response(
status_code=200,
json=["a", "list", "not", "an", "object"],
request=httpx.Request("POST", url),
)
monkeypatch.setattr(httpx.AsyncClient, "post", fake_post)
result = await post_to_sandbox(_ok_ctx(), "x", {})
assert "error" in result
assert "non-object" in result["error"].lower()
tests/tools/test_sandbox_tools.py
-348
View File
@@ -1,348 +0,0 @@
"""Smoke tests for the sandbox-bound SDK tool wrappers.
Covers: browser_action, terminal_execute, python_action, file_edit
helpers, and the seven Caido proxy tools.
These wrappers are pure pass-throughs to ``post_to_sandbox`` — there's
no per-tool logic to assert, so the tests focus on:
- ``FunctionTool`` registration succeeds (which proves the SDK could
derive a JSON schema from the type hints — a non-trivial check given
Literal types, ``dict[str, str]``, and strict-mode opt-outs).
- The dispatch payload to ``post_to_sandbox`` carries the right
``kwargs`` shape so the in-container tool server can dispatch to the
matching action.
- The ``send_request`` / ``repeat_request`` tools opt out of strict
schema mode (their ``headers`` / ``modifications`` dicts are
free-form and would otherwise fail registration).
"""
from __future__ import annotations
import json
from dataclasses import dataclass, field
from typing import Any
from unittest.mock import patch
import pytest
from agents.tool import FunctionTool
from strix.tools.browser.tool import browser_action
from strix.tools.proxy.tools import (
list_requests,
list_sitemap,
repeat_request,
scope_rules,
send_request,
view_request,
view_sitemap_entry,
)
from strix.tools.python.tool import python_action
from strix.tools.terminal.tool import terminal_execute
_ALL_SANDBOX_TOOLS = (
browser_action,
terminal_execute,
python_action,
list_requests,
view_request,
send_request,
repeat_request,
scope_rules,
list_sitemap,
view_sitemap_entry,
)
@dataclass
class _Ctx:
context: dict[str, Any] = field(default_factory=dict)
def _ctx_for(agent_id: str = "test-agent") -> _Ctx:
return _Ctx(
context={
"agent_id": agent_id,
"tool_server_host_port": 12345,
"sandbox_token": "test-token",
},
)
async def _invoke(tool: FunctionTool, ctx: _Ctx, **kwargs: Any) -> dict[str, Any]:
from agents.tool_context import ToolContext
tool_ctx = ToolContext(
context=ctx.context,
usage=None,
tool_name=tool.name,
tool_call_id="test-call-id",
tool_arguments=json.dumps(kwargs),
)
result = await tool.on_invoke_tool(tool_ctx, json.dumps(kwargs))
assert isinstance(result, str)
decoded = json.loads(result)
assert isinstance(decoded, dict)
return decoded
def test_all_sandbox_tools_register() -> None:
for tool in _ALL_SANDBOX_TOOLS:
assert isinstance(tool, FunctionTool)
def test_send_and_repeat_request_opt_out_of_strict_mode() -> None:
"""The two free-form-dict tools must turn off strict JSON schema."""
assert send_request.strict_json_schema is False
assert repeat_request.strict_json_schema is False
# All other tools should keep strict mode on (the SDK default).
for tool in (
browser_action,
terminal_execute,
python_action,
list_requests,
view_request,
scope_rules,
list_sitemap,
view_sitemap_entry,
):
assert tool.strict_json_schema is True, tool.name
# --- browser ---------------------------------------------------------------
@pytest.mark.asyncio
async def test_browser_action_dispatches_full_payload() -> None:
"""All optional browser_action params must forward as None when unset
(the in-container handler distinguishes ``None`` from missing)."""
fake = {"result": {"screenshot": "data:image/png;base64,..."}}
with patch(
"strix.tools.browser.tool.post_to_sandbox",
return_value=fake,
) as dispatch:
out = await _invoke(
browser_action,
_ctx_for(),
action="goto",
url="https://example.com",
)
assert out == fake
args, _ = dispatch.call_args
assert args[1] == "browser_action"
payload = args[2]
assert payload["action"] == "goto"
assert payload["url"] == "https://example.com"
# All optional params are present in the payload (as None / defaults)
# so the legacy in-container handler sees the full kwarg surface.
for key in (
"coordinate",
"text",
"tab_id",
"js_code",
"duration",
"key",
"file_path",
):
assert key in payload, key
assert payload[key] is None
assert payload["clear"] is False
# --- terminal --------------------------------------------------------------
@pytest.mark.asyncio
async def test_terminal_execute_dispatches() -> None:
fake = {"result": {"content": "hello\n", "exit_code": 0}}
with patch(
"strix.tools.terminal.tool.post_to_sandbox",
return_value=fake,
) as dispatch:
out = await _invoke(
terminal_execute,
_ctx_for(),
command="echo hello",
terminal_id="term-1",
)
assert out == fake
args, _ = dispatch.call_args
assert args[1] == "terminal_execute"
assert args[2]["command"] == "echo hello"
assert args[2]["terminal_id"] == "term-1"
assert args[2]["is_input"] is False
assert args[2]["no_enter"] is False
assert args[2]["timeout"] is None
# --- python ---------------------------------------------------------------
@pytest.mark.asyncio
async def test_python_action_dispatches() -> None:
fake = {"result": {"stdout": "42\n", "is_running": False}}
with patch(
"strix.tools.python.tool.post_to_sandbox",
return_value=fake,
) as dispatch:
out = await _invoke(
python_action,
_ctx_for(),
action="execute",
code="print(6*7)",
session_id="sess-1",
)
assert out == fake
args, _ = dispatch.call_args
assert args[1] == "python_action"
assert args[2] == {
"action": "execute",
"code": "print(6*7)",
"timeout": 30,
"session_id": "sess-1",
}
# --- proxy / Caido --------------------------------------------------------
@pytest.mark.asyncio
async def test_list_requests_forwards_full_query() -> None:
fake: dict[str, Any] = {"result": {"requests": []}}
with patch(
"strix.tools.proxy.tools.post_to_sandbox",
return_value=fake,
) as dispatch:
await _invoke(
list_requests,
_ctx_for(),
httpql_filter="resp.code:eq:500",
page_size=20,
sort_by="response_time",
sort_order="asc",
)
args, _ = dispatch.call_args
assert args[1] == "list_requests"
assert args[2]["httpql_filter"] == "resp.code:eq:500"
assert args[2]["page_size"] == 20
assert args[2]["sort_by"] == "response_time"
assert args[2]["sort_order"] == "asc"
# Defaults preserved.
assert args[2]["start_page"] == 1
assert args[2]["end_page"] == 1
@pytest.mark.asyncio
async def test_view_request_dispatches() -> None:
fake = {"result": {"raw": "GET / HTTP/1.1..."}}
with patch(
"strix.tools.proxy.tools.post_to_sandbox",
return_value=fake,
) as dispatch:
await _invoke(
view_request,
_ctx_for(),
request_id="req-9",
part="response",
search_pattern="Set-Cookie",
)
args, _ = dispatch.call_args
assert args[1] == "view_request"
assert args[2]["request_id"] == "req-9"
assert args[2]["part"] == "response"
assert args[2]["search_pattern"] == "Set-Cookie"
@pytest.mark.asyncio
async def test_send_request_normalizes_missing_headers() -> None:
"""Legacy schema treats omitted ``headers`` as ``{}``; the wrapper must too."""
fake = {"result": {"status": 200}}
with patch(
"strix.tools.proxy.tools.post_to_sandbox",
return_value=fake,
) as dispatch:
await _invoke(
send_request,
_ctx_for(),
method="POST",
url="https://api.example.com/login",
body='{"u":"x"}',
)
args, _ = dispatch.call_args
assert args[1] == "send_request"
assert args[2]["headers"] == {} # not None
assert args[2]["method"] == "POST"
assert args[2]["body"] == '{"u":"x"}'
@pytest.mark.asyncio
async def test_repeat_request_normalizes_missing_modifications() -> None:
fake = {"result": {"status": 200}}
with patch(
"strix.tools.proxy.tools.post_to_sandbox",
return_value=fake,
) as dispatch:
await _invoke(repeat_request, _ctx_for(), request_id="req-1")
args, _ = dispatch.call_args
assert args[1] == "repeat_request"
assert args[2]["modifications"] == {}
@pytest.mark.asyncio
async def test_scope_rules_dispatches() -> None:
fake = {"result": {"scope_id": "s-1"}}
with patch(
"strix.tools.proxy.tools.post_to_sandbox",
return_value=fake,
) as dispatch:
await _invoke(
scope_rules,
_ctx_for(),
action="create",
scope_name="prod",
allowlist=["*.example.com"],
)
args, _ = dispatch.call_args
assert args[1] == "scope_rules"
assert args[2]["action"] == "create"
assert args[2]["scope_name"] == "prod"
assert args[2]["allowlist"] == ["*.example.com"]
assert args[2]["denylist"] is None
@pytest.mark.asyncio
async def test_list_sitemap_defaults() -> None:
fake: dict[str, Any] = {"result": {"entries": []}}
with patch(
"strix.tools.proxy.tools.post_to_sandbox",
return_value=fake,
) as dispatch:
await _invoke(list_sitemap, _ctx_for())
args, _ = dispatch.call_args
assert args[1] == "list_sitemap"
assert args[2]["depth"] == "DIRECT"
assert args[2]["page"] == 1
@pytest.mark.asyncio
async def test_view_sitemap_entry_dispatches() -> None:
fake = {"result": {"entry_id": "e-1"}}
with patch(
"strix.tools.proxy.tools.post_to_sandbox",
return_value=fake,
) as dispatch:
await _invoke(view_sitemap_entry, _ctx_for(), entry_id="e-1")
args, _ = dispatch.call_args
assert args[1] == "view_sitemap_entry"
assert args[2] == {"entry_id": "e-1"}
tests/tools/test_tool_registration_modes.py
-62
View File
@@ -1,62 +0,0 @@
import importlib
import sys
from types import ModuleType
from typing import Any
from strix.config import Config
from strix.tools.registry import clear_registry
def _empty_config_load(_cls: type[Config]) -> dict[str, dict[str, str]]:
return {"env": {}}
def _reload_tools_module() -> ModuleType:
clear_registry()
for name in list(sys.modules):
if name == "strix.tools" or name.startswith("strix.tools."):
sys.modules.pop(name, None)
return importlib.import_module("strix.tools")
def test_non_sandbox_skips_browser_and_web_search_when_disabled(
monkeypatch: Any,
) -> None:
"""Browser registration is gated on STRIX_DISABLE_BROWSER and
web_search on PERPLEXITY_API_KEY; both should stay out of the
in-container ``register_tool`` registry when their gates are off.
Agents_graph is no longer in this registry — those tools are SDK
function tools (host-side only), not in-container tools.
"""
monkeypatch.setenv("STRIX_SANDBOX_MODE", "false")
monkeypatch.setenv("STRIX_DISABLE_BROWSER", "true")
monkeypatch.delenv("PERPLEXITY_API_KEY", raising=False)
monkeypatch.setattr(Config, "load", classmethod(_empty_config_load))
tools = _reload_tools_module()
names = set(tools.get_tool_names())
assert "browser_action" not in names
assert "web_search" not in names
def test_sandbox_registers_sandbox_tools_but_not_non_sandbox_tools(
monkeypatch: Any,
) -> None:
monkeypatch.setenv("STRIX_SANDBOX_MODE", "true")
monkeypatch.setenv("STRIX_DISABLE_BROWSER", "true")
monkeypatch.delenv("PERPLEXITY_API_KEY", raising=False)
monkeypatch.setattr(Config, "load", classmethod(_empty_config_load))
tools = _reload_tools_module()
names = set(tools.get_tool_names())
assert "terminal_execute" in names
assert "python_action" in names
assert "list_requests" in names
assert "create_agent" not in names
assert "finish_scan" not in names
assert "browser_action" not in names
assert "web_search" not in names