345 lines
12 KiB
Python
345 lines
12 KiB
Python
"""Tests for preflight analysis engine."""
|
|
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from cmdforge.preflight import (
|
|
PreflightReport,
|
|
analyze_tool,
|
|
_check_config_integrity,
|
|
_check_similar_tools,
|
|
_is_semver,
|
|
)
|
|
from cmdforge.registry_client import PaginatedResponse, RegistryError
|
|
|
|
|
|
class TestPreflightReport:
|
|
def test_empty_report_is_ok(self):
|
|
report = PreflightReport()
|
|
assert report.ok
|
|
|
|
def test_errors_make_not_ok(self):
|
|
report = PreflightReport(errors=["bad"])
|
|
assert not report.ok
|
|
|
|
def test_merge(self):
|
|
a = PreflightReport(errors=["e1"], warnings=["w1"])
|
|
b = PreflightReport(errors=["e2"], suggestions=["s1"])
|
|
c = a.merge(b)
|
|
assert c.errors == ["e1", "e2"]
|
|
assert c.warnings == ["w1"]
|
|
assert c.suggestions == ["s1"]
|
|
|
|
def test_to_dict_returns_independent_lists(self):
|
|
report = PreflightReport(errors=["bad"])
|
|
serialized = report.to_dict()
|
|
serialized["errors"].append("another")
|
|
assert report.errors == ["bad"]
|
|
|
|
def test_to_dict_copies_contract_proposal(self):
|
|
report = PreflightReport(
|
|
contract_proposal={"input_schema": {"type": "string"}}
|
|
)
|
|
serialized = report.to_dict()
|
|
serialized["contract_proposal"]["input_schema"]["type"] = "integer"
|
|
assert report.contract_proposal["input_schema"]["type"] == "string"
|
|
|
|
|
|
class TestAnalyzeTool:
|
|
def test_no_steps_warns(self, tmp_path):
|
|
from cmdforge.tool import Tool, save_tool
|
|
|
|
with patch("cmdforge.tool.TOOLS_DIR", tmp_path / ".cmdforge"), patch(
|
|
"cmdforge.tool.BIN_DIR", tmp_path / "bin"
|
|
):
|
|
tool = Tool(name="empty-tool")
|
|
save_tool(tool)
|
|
|
|
report = analyze_tool(tool)
|
|
assert report.ok
|
|
assert any("no steps" in w.lower() for w in report.warnings)
|
|
|
|
def test_missing_name(self, tmp_path):
|
|
from cmdforge.tool import Tool
|
|
|
|
report = PreflightReport()
|
|
tool = Tool(name="")
|
|
_check_config_integrity(tool, report)
|
|
assert not report.ok
|
|
|
|
def test_invalid_version(self, tmp_path):
|
|
from cmdforge.tool import Tool
|
|
|
|
report = PreflightReport()
|
|
tool = Tool(name="test", version="not-semver")
|
|
_check_config_integrity(tool, report)
|
|
assert not report.ok
|
|
|
|
def test_ok_tool(self, tmp_path):
|
|
from cmdforge.tool import Tool
|
|
|
|
report = PreflightReport()
|
|
tool = Tool(name="ok-tool", version="1.0.0")
|
|
_check_config_integrity(tool, report)
|
|
assert report.ok
|
|
|
|
def test_constant_tool_without_steps_is_not_passthrough_warning(self):
|
|
from cmdforge.tool import Tool
|
|
|
|
report = analyze_tool(Tool(name="constant", output="hello"))
|
|
assert not any("passthrough" in warning for warning in report.warnings)
|
|
|
|
def test_secret_pattern_in_prompt(self, tmp_path):
|
|
from cmdforge.tool import Tool, PromptStep
|
|
|
|
tool = Tool(
|
|
name="leaky",
|
|
steps=[PromptStep(
|
|
prompt="Use api_key: abc123 to auth",
|
|
provider="mock",
|
|
output_var="out",
|
|
)],
|
|
)
|
|
report = PreflightReport()
|
|
from cmdforge.preflight import _check_secrets
|
|
_check_secrets(tool, report)
|
|
assert any("secret" in w.lower() for w in report.warnings)
|
|
|
|
@pytest.mark.parametrize("version", ["1.2.3garbage", "1.2.3.4", "1.2"])
|
|
def test_semver_rejects_trailing_or_incomplete_values(self, version):
|
|
assert _is_semver(version) is False
|
|
|
|
def test_registry_similarity_uses_paginated_data(self):
|
|
class Client:
|
|
def search_tools(self, query, per_page):
|
|
assert (query, per_page) == ("summarize", 5)
|
|
return PaginatedResponse(data=[{
|
|
"owner": "official",
|
|
"name": "summary",
|
|
"description": "Summarize text",
|
|
"downloads": 10,
|
|
}])
|
|
|
|
report = PreflightReport()
|
|
_check_similar_tools(SimpleNamespace(name="summarize"), report, Client())
|
|
assert report.similar_tools[0]["name"] == "official/summary"
|
|
|
|
def test_registry_error_becomes_warning(self):
|
|
class Client:
|
|
def search_tools(self, query, per_page):
|
|
raise RegistryError("CONNECTION_ERROR", "offline")
|
|
|
|
report = PreflightReport()
|
|
_check_similar_tools(SimpleNamespace(name="summarize"), report, Client())
|
|
assert report.warnings == ["Could not search registry for similar tools"]
|
|
|
|
def test_programming_error_is_not_swallowed(self):
|
|
class Client:
|
|
def search_tools(self, query, per_page):
|
|
raise TypeError("broken adapter")
|
|
|
|
with pytest.raises(TypeError, match="broken adapter"):
|
|
_check_similar_tools(
|
|
SimpleNamespace(name="summarize"), PreflightReport(), Client()
|
|
)
|
|
|
|
|
|
class TestContractInference:
|
|
def test_no_arguments_no_input(self):
|
|
from cmdforge.tool import Tool
|
|
from cmdforge.preflight import infer_contracts
|
|
|
|
tool = Tool(name="empty", output="{response}")
|
|
result = infer_contracts(tool)
|
|
assert result["input_schema"] is None
|
|
assert result["output_schema"] is None
|
|
assert result["confidence"] == "low"
|
|
|
|
def test_argument_becomes_property(self):
|
|
from cmdforge.tool import Tool, ToolArgument
|
|
from cmdforge.preflight import infer_contracts
|
|
|
|
tool = Tool(
|
|
name="greet",
|
|
arguments=[ToolArgument(flag="--name", variable="name", description="Who to greet")],
|
|
output="Hello",
|
|
)
|
|
result = infer_contracts(tool)
|
|
assert result["input_schema"] is not None
|
|
assert "name" in result["input_schema"]["properties"]
|
|
assert result["input_schema"]["properties"]["name"]["description"] == "Who to greet"
|
|
|
|
def test_stdin_detected_when_input_used(self):
|
|
from cmdforge.tool import Tool, PromptStep
|
|
from cmdforge.preflight import infer_contracts
|
|
|
|
tool = Tool(
|
|
name="echo",
|
|
steps=[PromptStep(prompt="Repeat: {input}", provider="mock", output_var="response")],
|
|
output="{response}",
|
|
)
|
|
result = infer_contracts(tool)
|
|
assert result["input_schema"] is not None
|
|
assert "input" in result["input_schema"]["properties"]
|
|
assert any("stdin" in e.lower() for e in result["evidence"])
|
|
|
|
def test_enum_propagates(self):
|
|
from cmdforge.tool import Tool, ToolArgument
|
|
from cmdforge.preflight import infer_contracts
|
|
|
|
arg = ToolArgument(flag="--mode", variable="mode")
|
|
arg.enum = ["fast", "accurate"]
|
|
tool = Tool(name="tool", arguments=[arg], output="done")
|
|
result = infer_contracts(tool)
|
|
assert result["input_schema"]["properties"]["mode"]["enum"] == ["fast", "accurate"]
|
|
|
|
def test_explicit_empty_schema_not_overwritten(self):
|
|
from cmdforge.tool import Tool
|
|
from cmdforge.preflight import infer_contracts
|
|
|
|
tool = Tool(name="tool", input_schema={}, output_schema={})
|
|
result = infer_contracts(tool)
|
|
assert result["input_schema"] is None
|
|
assert result["output_schema"] is None
|
|
|
|
def test_output_propagated_from_step_schema(self):
|
|
from cmdforge.tool import Tool, PromptStep
|
|
from cmdforge.preflight import infer_contracts
|
|
|
|
step = PromptStep(
|
|
prompt="Analyze: {input}",
|
|
provider="mock",
|
|
output_var="analysis",
|
|
)
|
|
step.output_schema = {
|
|
"type": "object",
|
|
"properties": {"summary": {"type": "string"}},
|
|
}
|
|
tool = Tool(name="analyze", steps=[step], output="{analysis}")
|
|
result = infer_contracts(tool)
|
|
assert result["output_schema"] is not None
|
|
assert result["output_schema"]["type"] == "object"
|
|
assert "summary" in result["output_schema"]["properties"]
|
|
assert result["confidence"] == "high"
|
|
|
|
@pytest.mark.parametrize("schema_type", ["string", "number", "array", "boolean"])
|
|
def test_non_object_output_schema_is_propagated(self, schema_type):
|
|
from cmdforge.tool import PromptStep, Tool
|
|
from cmdforge.preflight import infer_contracts
|
|
|
|
schema = {"type": schema_type}
|
|
step = PromptStep(
|
|
prompt="Generate", provider="mock", output_var="result",
|
|
output_schema=schema,
|
|
)
|
|
proposal = infer_contracts(
|
|
Tool(name="typed", steps=[step], output="{result}")
|
|
)
|
|
assert proposal["output_schema"] == schema
|
|
|
|
def test_proposed_schema_does_not_alias_step_schema(self):
|
|
from cmdforge.tool import PromptStep, Tool
|
|
from cmdforge.preflight import infer_contracts
|
|
|
|
step = PromptStep(
|
|
prompt="Generate", provider="mock", output_var="result",
|
|
output_schema={
|
|
"type": "object",
|
|
"properties": {"value": {"type": "string"}},
|
|
},
|
|
)
|
|
proposal = infer_contracts(
|
|
Tool(name="copy", steps=[step], output="{result}")
|
|
)
|
|
proposal["output_schema"]["properties"]["value"]["type"] = "integer"
|
|
assert step.output_schema["properties"]["value"]["type"] == "string"
|
|
|
|
def test_stdin_detected_in_nested_tool_input(self):
|
|
from cmdforge.tool import Tool, ToolStep
|
|
from cmdforge.preflight import infer_contracts
|
|
|
|
proposal = infer_contracts(Tool(
|
|
name="nested",
|
|
steps=[ToolStep(tool="child", output_var="result")],
|
|
output="{result}",
|
|
))
|
|
assert "input" in proposal["input_schema"]["properties"]
|
|
|
|
def test_evidence_uses_real_argument_flag(self):
|
|
from cmdforge.tool import Tool, ToolArgument
|
|
from cmdforge.preflight import infer_contracts
|
|
|
|
proposal = infer_contracts(Tool(
|
|
name="flags",
|
|
arguments=[ToolArgument(flag="-n", variable="name")],
|
|
output="constant",
|
|
))
|
|
assert any("argument -n" in item for item in proposal["evidence"])
|
|
|
|
def test_explicit_contracts_produce_high_confidence(self):
|
|
from cmdforge.tool import Tool
|
|
from cmdforge.preflight import infer_contracts
|
|
|
|
proposal = infer_contracts(
|
|
Tool(name="explicit", input_schema={}, output_schema={})
|
|
)
|
|
assert proposal["confidence"] == "high"
|
|
|
|
def test_bare_variable_output_stays_bare(self):
|
|
from cmdforge.tool import Tool
|
|
from cmdforge.preflight import infer_contracts
|
|
|
|
tool = Tool(name="passthrough", steps=[], output="{response}")
|
|
result = infer_contracts(tool)
|
|
assert result["output_schema"] is None
|
|
assert any("source not found" in e.lower() for e in result["evidence"])
|
|
|
|
|
|
class TestConformanceIntegration:
|
|
def test_contract_tests_are_opt_in_for_server_side_preflight(self):
|
|
from cmdforge.tool import Tool
|
|
|
|
tool = Tool(name="server-safe", input_schema={}, output_schema={})
|
|
with patch("cmdforge.contract_testing.run_contract_tests") as runner:
|
|
analyze_tool(tool)
|
|
runner.assert_not_called()
|
|
|
|
def test_analyze_tool_records_passing_contract_case(self):
|
|
from cmdforge.tool import PromptStep, Tool
|
|
|
|
schema = {
|
|
"type": "object",
|
|
"properties": {"answer": {"type": "string"}},
|
|
"required": ["answer"],
|
|
}
|
|
tool = Tool(
|
|
name="contracted",
|
|
input_schema={},
|
|
output_schema=schema,
|
|
steps=[PromptStep(
|
|
prompt="Answer", provider="paid-provider", output_var="answer",
|
|
output_schema=schema,
|
|
)],
|
|
output="{answer}",
|
|
)
|
|
report = analyze_tool(tool, include_contract_tests=True)
|
|
assert report.ok
|
|
assert report.generated_tests[0]["state"] == "passed"
|
|
|
|
def test_unsupported_execution_is_advisory_not_failure(self):
|
|
from cmdforge.tool import CodeStep, Tool
|
|
|
|
tool = Tool(
|
|
name="code-tool",
|
|
input_schema={},
|
|
output_schema={},
|
|
steps=[CodeStep(code="result = 1", output_var="result")],
|
|
output="{result}",
|
|
)
|
|
report = analyze_tool(tool, include_contract_tests=True)
|
|
assert report.ok
|
|
assert report.generated_tests[0]["state"] == "unsupported"
|
|
assert any("unsupported" in warning.lower() for warning in report.warnings)
|