mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 05:06:59 +00:00
8d994cc1e7
* fix(acceptance): require passing unittest counts * test(acceptance): retain unittest output attribution guard * test(acceptance): cover unittest errors and recorded failures * test(acceptance): honor unittest empty-selection exit status * docs(subagents): keep unittest guidance within inherited budget
161 lines
7.4 KiB
Python
161 lines
7.4 KiB
Python
"""Unittest acceptance requires actual passing tests, including annotated OK summaries."""
|
|
|
|
import asyncio
|
|
import importlib.util
|
|
import os
|
|
import shlex
|
|
import sys
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from langchain_core.messages import AIMessage
|
|
from langchain_core.tools import StructuredTool
|
|
from langgraph.graph import END, START, MessagesState, StateGraph
|
|
from langgraph.prebuilt import ToolNode
|
|
|
|
from deerflow.agents.middlewares.tool_error_handling_middleware import ToolErrorHandlingMiddleware
|
|
from deerflow.sandbox.local.local_sandbox import LocalSandbox, PathMapping
|
|
from deerflow.subagents.acceptance_checks import check_acceptance_criteria, render_acceptance_section, render_acceptance_segment
|
|
|
|
|
|
def _summary(total=2, options="", newline="\n"):
|
|
return newline.join([f"Ran {total} tests in 0.001s", "", f"OK{options}"])
|
|
|
|
|
|
def _verdict(output, **overrides):
|
|
execution = {"command": "python -m unittest", "status": "success", "shell_persistent": False, "output_tail": output, **overrides}
|
|
return check_acceptance_criteria(["tests_passed:python -m unittest"], bash_executions=[execution])
|
|
|
|
|
|
@pytest.mark.parametrize("options", ["", " (skipped=1)", " (expected failures=1)", " (skipped=1, expected failures=1)"])
|
|
@pytest.mark.parametrize("newline", ["\n", "\r\n"])
|
|
def test_passing_tests_survive_unittest_annotations(options, newline):
|
|
verdict = _verdict(_summary(total=3, options=options, newline=newline))
|
|
assert verdict["all_hold"] is True
|
|
assert "[holds]" in render_acceptance_section(verdict)
|
|
assert "acceptance:" in render_acceptance_segment(verdict)
|
|
assert "execution evidence only" in render_acceptance_section(verdict)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"output",
|
|
[
|
|
"OK",
|
|
"OK (skipped=1)",
|
|
"ok\nexample/package",
|
|
_summary(0),
|
|
_summary(2, " (skipped=2)"),
|
|
_summary(2, " (expected failures=2)"),
|
|
_summary(2, " (skipped=1, expected failures=1)"),
|
|
_summary(2, " (skipped=3)"),
|
|
_summary(2, " (skipped=1, skipped=0)"),
|
|
_summary(2, " (unknown=1)"),
|
|
_summary("9" * 100),
|
|
_summary(2, " (skipped=" + "9" * 100 + ")"),
|
|
"Ran 2 tests in\n\nOK",
|
|
"Ran 2 tests in 0.001s\n\nOK (skipped=",
|
|
"warning: " + _summary(),
|
|
],
|
|
)
|
|
def test_incomplete_empty_or_nonpassing_unittest_evidence_remains_unverified(output):
|
|
verdict = _verdict(output)
|
|
assert (verdict["leaves"][0]["checked"], verdict["all_hold"]) == (False, False)
|
|
assert "[UNVERIFIED]" in render_acceptance_section(verdict)
|
|
|
|
|
|
@pytest.mark.parametrize("failure", ["FAILED (failures=1)", "FAILED (errors=1)", "FAILED (unexpected successes=1)"])
|
|
@pytest.mark.parametrize("reverse", [False, True])
|
|
def test_failure_records_still_win_over_a_passing_summary(failure, reverse):
|
|
records = [_summary(2, " (skipped=1)"), failure]
|
|
verdict = _verdict("\n".join(reversed(records) if reverse else records))
|
|
assert (verdict["leaves"][0]["checked"], verdict["all_hold"]) == (True, False)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"overrides",
|
|
[
|
|
{"shell_persistent": True},
|
|
{"shell_persistent": None},
|
|
{"command_truncated": True},
|
|
{"command": "python -m unittest sample.OneTest"},
|
|
{"command": "echo 'other'; python -m unittest"},
|
|
{"command": 'cd "OK (skipped=1)" && python -m unittest'},
|
|
],
|
|
)
|
|
def test_annotations_do_not_bypass_evidence_admission(overrides):
|
|
assert _verdict(_summary(2, " (skipped=1)"), **overrides)["leaves"][0]["checked"] is False
|
|
|
|
|
|
def test_recorded_error_status_still_overrides_positive_counts():
|
|
verdict = _verdict(_summary(2, " (skipped=1)"), status="error")
|
|
assert (verdict["leaves"][0]["checked"], verdict["all_hold"]) == (True, False)
|
|
|
|
|
|
@pytest.fixture
|
|
def executor_module(monkeypatch):
|
|
# conftest substitutes the executor; use its actual harvest implementation.
|
|
path = Path(__file__).parents[1] / "packages/harness/deerflow/subagents/executor.py"
|
|
spec = importlib.util.spec_from_file_location("_unittest_evidence_executor", path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
monkeypatch.setitem(sys.modules, spec.name, module)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
class SandboxMessages(MessagesState):
|
|
sandbox: dict
|
|
|
|
|
|
@pytest.mark.skipif(os.name == "nt", reason="Real LocalSandbox command integration uses a POSIX Python executable")
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"methods,expected",
|
|
[
|
|
(" def test_pass(self): pass\n", (True, True)),
|
|
(' def test_pass(self): pass\n @unittest.skip("optional")\n def test_skip(self): pass\n', (True, True)),
|
|
(' @unittest.skip("optional")\n def test_skip(self): pass\n', (False, False)),
|
|
(" pass\n", (True, False)),
|
|
(" @unittest.expectedFailure\n def test_xfail(self): self.fail()\n", (False, False)),
|
|
(" def test_pass(self): pass\n @unittest.expectedFailure\n def test_xfail(self): self.fail()\n", (True, True)),
|
|
(" def test_fail(self): self.fail()\n", (True, False)),
|
|
(' def test_error(self): raise RuntimeError("fixture")\n', (True, False)),
|
|
(" @unittest.expectedFailure\n def test_xpass(self): pass\n", (True, False)),
|
|
],
|
|
ids=["pass", "pass-and-skip", "all-skipped", "empty", "all-expected-failure", "pass-and-expected-failure", "failure", "error", "unexpected-success"],
|
|
)
|
|
async def test_native_unittest_results_reach_parent_acceptance(tmp_path, monkeypatch, executor_module, methods, expected):
|
|
(tmp_path / "test_sample.py").write_text("import unittest\nclass Sample(unittest.TestCase):\n" + methods, encoding="utf-8")
|
|
sandbox = LocalSandbox("unittest-evidence", path_mappings=[PathMapping("/mnt/user-data/workspace", str(tmp_path))])
|
|
monkeypatch.setattr("deerflow.sandbox.sandbox_provider.get_sandbox_provider", lambda: SimpleNamespace(get=lambda sandbox_id: sandbox if sandbox_id == sandbox.id else None))
|
|
outputs = []
|
|
|
|
async def bash(command: str) -> str:
|
|
output = await asyncio.to_thread(sandbox.execute_command, command, timeout=30)
|
|
outputs.append(output)
|
|
return output
|
|
|
|
tool = StructuredTool.from_function(coroutine=bash, name="bash", description="Execute the offline Python unittest fixture.")
|
|
graph = StateGraph(SandboxMessages)
|
|
graph.add_node("tools", ToolNode([tool], awrap_tool_call=ToolErrorHandlingMiddleware().awrap_tool_call))
|
|
graph.add_edge(START, "tools")
|
|
graph.add_edge("tools", END)
|
|
command = shlex.quote(sys.executable) + " -m unittest discover -s /mnt/user-data/workspace"
|
|
state = await graph.compile().ainvoke(
|
|
{
|
|
"sandbox": {"sandbox_id": sandbox.id},
|
|
"messages": [AIMessage(content="", tool_calls=[{"name": "bash", "id": "unittest-call", "type": "tool_call", "args": {"command": command}}])],
|
|
}
|
|
)
|
|
assert len(outputs) == 1 and "Ran " in outputs[0], outputs
|
|
executions = executor_module._harvest_bash_executions(state)
|
|
assert len(executions) == 1 and executions[0]["tool_call_id"] == "unittest-call"
|
|
assert executions[0]["shell_persistent"] is False
|
|
verdict = check_acceptance_criteria([f"tests_passed:{command}"], bash_executions=executions)
|
|
assert (verdict["leaves"][0]["checked"], verdict["all_hold"]) == expected, (outputs, verdict)
|
|
if expected == (True, False):
|
|
# Python 3.12+ exits 5 for an empty selection, 1 for test errors/failures.
|
|
exit_code = 5 if methods == " pass\n" else 1
|
|
assert f"Exit Code: {exit_code}" in outputs[0]
|
|
assert executions[0]["status"] == "error"
|