mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 05:06:59 +00:00
e4c893f2fb
* feat(acceptance): verify required CSV header columns * fix(acceptance): address CSV header review feedback * fix(acceptance): guard CSV criteria and malformed headers * test(acceptance): isolate CSV deadline clock simulations
344 lines
14 KiB
Python
344 lines
14 KiB
Python
"""CSV header acceptance through the public, workspace-scoped checklist."""
|
|
|
|
import json
|
|
from types import SimpleNamespace
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
from deerflow.subagents import acceptance_checks as checks
|
|
|
|
|
|
@pytest.fixture
|
|
def parser_clock(monkeypatch):
|
|
import time
|
|
|
|
original_monotonic = time.monotonic
|
|
clock = Mock(return_value=0)
|
|
monkeypatch.setattr(checks, "time", SimpleNamespace(monotonic=clock))
|
|
yield clock
|
|
# Deadline simulations must leave event loops and other threads on the real clock.
|
|
assert time.monotonic is original_monotonic
|
|
|
|
|
|
@pytest.fixture
|
|
def check_csv(tmp_path):
|
|
def check(content, columns=None, path="report.csv", **kwargs):
|
|
if content is not None:
|
|
(tmp_path / "report.csv").write_bytes(content)
|
|
criterion = f"file:{path} csv-columns:{json.dumps(columns or ['id', 'name'])}"
|
|
return checks.check_acceptance_criteria(
|
|
[criterion],
|
|
runtime=SimpleNamespace(state={"sandbox": {"sandbox_id": "local"}}),
|
|
thread_data={"workspace_path": str(tmp_path)},
|
|
**kwargs,
|
|
)
|
|
|
|
return check
|
|
|
|
|
|
@pytest.mark.parametrize("content", [b"id,name\n1,A", b"\xef\xbb\xbfid,name\r\n", b'"id","name",extra\n', b"id,name", b"id,name\n" + b"\xff" * 60_000], ids=["plain", "bom", "quoted", "no-newline", "large-invalid-body"])
|
|
def test_header_holds_without_validating_body(check_csv, content):
|
|
assert check_csv(content)["all_hold"] is True
|
|
|
|
|
|
@pytest.mark.parametrize("content", [b"", b"\n", b"\xef\xbb\xbf", b"id,other\n", b"id,name,name\n", b"id,name,\n", b"ID,name\n", b"id, name\n", b'"id","name\n', b"id,\xff\n"])
|
|
def test_invalid_or_missing_header_does_not_hold(check_csv, content):
|
|
leaf = check_csv(content)["leaves"][0]
|
|
assert (leaf["checked"], leaf["holds"]) == (True, False)
|
|
|
|
|
|
def test_quoted_columns_and_multiline_header(check_csv):
|
|
assert check_csv(b'"last,name","line\nbreak","a""b"\r\n', ["last,name", "line\nbreak", 'a"b'])["all_hold"] is True
|
|
|
|
|
|
@pytest.mark.parametrize("size,checked", [(50_000, True), (50_001, False)])
|
|
def test_header_byte_limit(check_csv, size, checked):
|
|
leaf = check_csv(b"id,name," + b"x" * (size - 9) + b"\n" + b"1" * 60_000)["leaves"][0]
|
|
assert (leaf["checked"], leaf["holds"]) == (checked, checked)
|
|
|
|
|
|
def test_partial_header_cannot_hold(check_csv):
|
|
leaf = check_csv(b"id,name", size_prober=lambda *args: 100)["leaves"][0]
|
|
assert leaf["checked"] is False
|
|
|
|
|
|
def test_missing_file_is_negative(check_csv):
|
|
leaf = check_csv(None)["leaves"][0]
|
|
assert (leaf["checked"], leaf["holds"]) == (True, False)
|
|
|
|
|
|
@pytest.mark.parametrize("path", ["../outside.csv", "/mnt/user-data/uploads/report.csv"])
|
|
def test_outside_shared_workspace_never_reads(check_csv, path):
|
|
probe = Mock(side_effect=AssertionError("out of scope"))
|
|
assert check_csv(b"id,name\n", path=path, size_prober=probe)["leaves"][0]["checked"] is False
|
|
probe.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize("payload", ["[]", '["id","id"]', "[1]", "null", '[""]', '["id",]', "id,name"])
|
|
def test_invalid_criteria_are_unverified(payload):
|
|
criterion = f"file:report.csv csv-columns:{payload}"
|
|
verdict = checks.check_acceptance_criteria([criterion])
|
|
assert verdict["unchecked"] == [criterion]
|
|
assert verdict["all_hold"] is False
|
|
|
|
|
|
def test_admission_uses_csv_path():
|
|
assert checks.parse_file_criterion('file:report.csv csv-columns:["id"]') == ("file_csv_columns", "report.csv")
|
|
|
|
|
|
@pytest.mark.parametrize("column", ["\ud800", "\udfff", "id\ud800"])
|
|
def test_surrogate_columns_are_invalid_before_file_access(check_csv, column):
|
|
probe = Mock(side_effect=AssertionError("invalid criteria must not access files"))
|
|
verdict = check_csv(b"id,name\n", [column], size_prober=probe)
|
|
assert verdict["leaves"][0]["checked"] is False
|
|
json.dumps(verdict, ensure_ascii=False).encode("utf-8")
|
|
probe.assert_not_called()
|
|
|
|
|
|
def test_escaped_surrogate_pair_is_a_valid_unicode_column(check_csv):
|
|
assert check_csv("id,\U0001f600\n".encode(), ["\U0001f600"])["all_hold"] is True
|
|
|
|
|
|
@pytest.mark.parametrize("header", [b'"id"x,name\n', b'"line\nbreak"x,name\r\n', b'"id"x,name\r'])
|
|
@pytest.mark.parametrize("large_body", [False, True])
|
|
def test_definite_malformed_header_is_negative_regardless_of_body(check_csv, header, large_body):
|
|
leaf = check_csv(header + (b"x" * 60_000 if large_body else b""))["leaves"][0]
|
|
assert (leaf["checked"], leaf["holds"]) == (True, False)
|
|
assert leaf["detail"] == "invalid CSV header"
|
|
|
|
|
|
def test_truncated_quoted_header_remains_unverified(check_csv):
|
|
leaf = check_csv(b'"id\n' + b"x" * 60_000)["leaves"][0]
|
|
assert (leaf["checked"], leaf["holds"]) == (False, False)
|
|
assert leaf["detail"] == "CSV header incomplete within the read cap"
|
|
|
|
|
|
@pytest.mark.parametrize("size,checked", [(50_000, True), (50_001, False)])
|
|
def test_malformed_header_respects_byte_budget(check_csv, size, checked):
|
|
leaf = check_csv(b'"id"x,' + b"x" * (size - 7) + b"\n" + b"x" * 60_000)["leaves"][0]
|
|
assert (leaf["checked"], leaf["holds"]) == (checked, False)
|
|
|
|
|
|
def test_malformed_header_respects_parser_deadline(check_csv, monkeypatch, parser_clock):
|
|
reader = checks.csv.reader
|
|
|
|
def delayed_parse(*args, **kwargs):
|
|
parser_clock.return_value = 6
|
|
return reader(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(checks.csv, "reader", delayed_parse)
|
|
leaf = check_csv(b'"id"x,name\n')["leaves"][0]
|
|
assert (leaf["checked"], leaf["holds"]) == (False, False)
|
|
assert leaf["detail"] == "CSV header deadline exceeded"
|
|
|
|
|
|
def test_revoked_access_prevents_metadata(check_csv, monkeypatch):
|
|
from deerflow.sandbox.exceptions import SandboxError
|
|
|
|
monkeypatch.setattr("deerflow.authz.sandbox_authz.authorize_sandbox_execution", Mock(side_effect=SandboxError("denied")))
|
|
probe = Mock(side_effect=AssertionError("must authorize first"))
|
|
assert check_csv(b"id,name\n", size_prober=probe)["leaves"][0]["checked"] is False
|
|
probe.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize("content", [b'id,name\n"unterminated', b"id,name\r1,A\r", b"id,name\r\n1,A\r\n"])
|
|
def test_only_first_logical_record_is_checked(check_csv, content):
|
|
assert check_csv(content)["all_hold"] is True
|
|
|
|
|
|
def test_unknown_size_does_not_read(check_csv, monkeypatch):
|
|
reader = Mock(side_effect=AssertionError("no size evidence"))
|
|
monkeypatch.setattr(checks, "_read_bounded_file_content", reader)
|
|
assert check_csv(b"id,name\n", size_prober=lambda *args: None)["leaves"][0]["checked"] is False
|
|
reader.assert_not_called()
|
|
|
|
|
|
def test_expired_read_is_unverified(check_csv, monkeypatch, parser_clock):
|
|
def delayed(*args, **kwargs):
|
|
parser_clock.return_value = 6
|
|
return b"id,name\n"
|
|
|
|
monkeypatch.setattr(checks, "_read_bounded_file_content", delayed)
|
|
assert check_csv(b"id,name\n")["leaves"][0]["checked"] is False
|
|
|
|
|
|
@pytest.mark.parametrize("failure", [None, OSError("unreadable")])
|
|
def test_read_failure_remains_unverified(check_csv, monkeypatch, failure):
|
|
reader = Mock(side_effect=failure) if failure else Mock(return_value=None)
|
|
monkeypatch.setattr(checks, "_read_bounded_file_content", reader)
|
|
assert check_csv(b"id,name\n")["leaves"][0]["checked"] is False
|
|
|
|
|
|
def test_remote_transport_checks_deadline_and_completion(monkeypatch, tmp_path):
|
|
import base64
|
|
|
|
payload = "JSON\n" + base64.b64encode(b"id,name\n\n000").decode() + "\nEND\n"
|
|
sandbox = Mock()
|
|
sandbox.execute_command.return_value = payload
|
|
monkeypatch.setattr("deerflow.sandbox.tools.ensure_sandbox_initialized", lambda runtime: sandbox)
|
|
|
|
def check():
|
|
return checks.check_acceptance_criteria(
|
|
['file:report.csv csv-columns:["id","name"]'],
|
|
runtime=SimpleNamespace(state=None),
|
|
thread_data={"workspace_path": str(tmp_path)},
|
|
size_prober=lambda *args: 8,
|
|
)
|
|
|
|
assert check()["all_hold"] is True
|
|
command = sandbox.execute_command.call_args
|
|
assert command.kwargs["timeout"] == 10
|
|
assert "/usr/bin/timeout 5" in command.args[0]
|
|
assert command.args[0].endswith("50001")
|
|
sandbox.execute_command.return_value = payload[:-5]
|
|
assert check()["leaves"][0]["checked"] is False
|
|
sandbox.execute_command.return_value = "JSON\n" + base64.b64encode(b"id,name\n\n124").decode() + "\nEND\n"
|
|
assert check()["leaves"][0]["checked"] is False
|
|
sandbox.read_file.assert_not_called()
|
|
|
|
|
|
def test_parser_resource_limit_is_unverified(check_csv, monkeypatch):
|
|
monkeypatch.setattr(checks.csv, "reader", Mock(side_effect=checks.csv.Error("field larger than field limit (2)")))
|
|
assert check_csv(b"id,name\n")["leaves"][0]["checked"] is False
|
|
|
|
|
|
def test_missing_workspace_is_unverified():
|
|
verdict = checks.check_acceptance_criteria(['file:report.csv csv-columns:["id"]'])
|
|
assert verdict["leaves"][0]["checked"] is False
|
|
|
|
|
|
def test_directory_is_unverified(check_csv, tmp_path):
|
|
(tmp_path / "report.csv").mkdir()
|
|
assert check_csv(None)["leaves"][0]["checked"] is False
|
|
|
|
|
|
def test_local_read_rechecks_authorization(check_csv, monkeypatch):
|
|
from deerflow.sandbox.exceptions import SandboxError
|
|
|
|
authorize = Mock(side_effect=[None, SandboxError("revoked")])
|
|
monkeypatch.setattr("deerflow.authz.sandbox_authz.authorize_sandbox_execution", authorize)
|
|
assert check_csv(b"id,name\n")["leaves"][0]["checked"] is False
|
|
assert authorize.call_count == 2
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"content,columns,checked,holds",
|
|
[
|
|
(b"\xef\xbb\xbfid,name\r\n", ["id", "name"], True, True),
|
|
(b'"last,name","line\nbreak"\n', ["last,name", "line\nbreak"], True, True),
|
|
(b"id,name,name\n", ["id"], True, False),
|
|
(b"", ["id"], True, False),
|
|
(b"id,name\n" + b"x" * 60_000, ["id", "name"], True, True),
|
|
(b'"' + b"x" * 50_001, ["id"], False, False),
|
|
(b'"id"x,name\n' + b"x" * 60_000, ["id"], True, False),
|
|
],
|
|
ids=["bom", "multiline", "duplicate", "empty", "large-body", "oversized-header", "malformed-header-large-body"],
|
|
)
|
|
def test_remote_csv_uses_real_bounded_transport(tmp_path, monkeypatch, content, columns, checked, holds):
|
|
import shlex
|
|
import subprocess
|
|
import sys
|
|
|
|
if sys.platform != "linux":
|
|
pytest.skip("Requires Linux sandbox command tools")
|
|
(tmp_path / "report.csv").write_bytes(content)
|
|
|
|
class ShellSandbox:
|
|
def execute_command(self, command, **kwargs):
|
|
assert kwargs["env"]
|
|
args = shlex.split(command)
|
|
prefix = "/mnt/user-data/workspace"
|
|
args = [str(tmp_path) + arg[len(prefix) :] if arg == prefix or arg.startswith(prefix + "/") else arg for arg in args]
|
|
return subprocess.run(args, capture_output=True, text=True, check=True, timeout=15).stdout
|
|
|
|
def read_file(self, *args, **kwargs):
|
|
pytest.fail("CSV acceptance must not fall back to unbounded reads")
|
|
|
|
monkeypatch.setattr("deerflow.sandbox.tools.ensure_sandbox_initialized", lambda runtime: ShellSandbox())
|
|
verdict = checks.check_acceptance_criteria(
|
|
[f"file:report.csv csv-columns:{json.dumps(columns)}"],
|
|
runtime=SimpleNamespace(state=None),
|
|
thread_data={"workspace_path": str(tmp_path)},
|
|
)
|
|
leaf = verdict["leaves"][0]
|
|
assert (leaf["checked"], leaf["holds"]) == (checked, holds)
|
|
|
|
|
|
def test_header_budget_counts_utf8_bytes(check_csv):
|
|
verdict = check_csv(("id,name," + "\u4e2d" * 16_665 + "\n").encode())
|
|
assert verdict["leaves"][0]["checked"] is False
|
|
|
|
|
|
def test_header_without_newline_at_exact_cap_holds(check_csv):
|
|
assert check_csv(b"id,name," + b"x" * (50_000 - 8))["all_hold"] is True
|
|
|
|
|
|
def test_required_columns_allow_reordering(check_csv):
|
|
assert check_csv(b"extra,name,id\n")["all_hold"] is True
|
|
|
|
|
|
@pytest.mark.parametrize("parse_seconds,holds", [(0, True), (6, False)])
|
|
def test_remote_read_latency_does_not_consume_parser_deadline(monkeypatch, tmp_path, parse_seconds, holds, parser_clock):
|
|
import base64
|
|
|
|
sandbox = Mock()
|
|
|
|
def delayed_read(*args, **kwargs):
|
|
parser_clock.return_value = 6
|
|
return "JSON\n" + base64.b64encode(b"id,name\n\n000").decode() + "\nEND\n"
|
|
|
|
sandbox.execute_command.side_effect = delayed_read
|
|
monkeypatch.setattr("deerflow.sandbox.tools.ensure_sandbox_initialized", lambda runtime: sandbox)
|
|
reader = checks.csv.reader
|
|
|
|
def delayed_parse(*args, **kwargs):
|
|
header = next(reader(*args, **kwargs))
|
|
parser_clock.return_value += parse_seconds
|
|
return iter([header])
|
|
|
|
monkeypatch.setattr(checks.csv, "reader", delayed_parse)
|
|
verdict = checks.check_acceptance_criteria(
|
|
['file:report.csv csv-columns:["id","name"]'],
|
|
runtime=SimpleNamespace(state=None),
|
|
thread_data={"workspace_path": str(tmp_path)},
|
|
size_prober=lambda *args: 8,
|
|
)
|
|
leaf = verdict["leaves"][0]
|
|
assert (leaf["checked"], leaf["holds"]) == (holds, holds)
|
|
if not holds:
|
|
assert leaf["detail"] == "CSV header deadline exceeded"
|
|
command = sandbox.execute_command.call_args
|
|
assert command.kwargs["timeout"] == 10
|
|
assert "/usr/bin/timeout 5" in command.args[0]
|
|
sandbox.read_file.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize("columns", [["z", "id", "a"], ['a"b', "line\nbreak", " id "]], ids=["ordered", "escaped"])
|
|
def test_missing_columns_detail_names_exact_columns(check_csv, columns):
|
|
leaf = check_csv(b"id,name\n", columns)["leaves"][0]
|
|
assert (leaf["checked"], leaf["holds"]) == (True, False)
|
|
missing = sorted(set(columns) - {"id", "name"})
|
|
assert leaf["detail"] == f"CSV header is missing required columns: {json.dumps(missing)}"
|
|
|
|
|
|
def test_missing_columns_detail_remains_bounded(check_csv):
|
|
leaf = check_csv(b"id,name\n", ["a" * 100, "b" * 100])["leaves"][0]
|
|
assert (leaf["checked"], leaf["holds"]) == (True, False)
|
|
assert len(leaf["detail"]) == 160
|
|
assert leaf["detail"].endswith("...")
|
|
|
|
|
|
@pytest.mark.parametrize("suffix", [b"\n", b"", b"\n" + b"1" * 60_000], ids=["complete", "eof", "large-body"])
|
|
def test_over_budget_header_detail_is_distinct(check_csv, suffix):
|
|
content = b"id,name," + b"x" * (50_001 - 8 - (1 if suffix.startswith(b"\n") else 0)) + suffix
|
|
leaf = check_csv(content)["leaves"][0]
|
|
assert (leaf["checked"], leaf["holds"]) == (False, False)
|
|
assert leaf["detail"] == "CSV header exceeds the 50000-byte read budget"
|
|
|
|
|
|
def test_truncated_header_detail_remains_incomplete(check_csv):
|
|
leaf = check_csv(b"id,name," + b"x" * 60_000)["leaves"][0]
|
|
assert (leaf["checked"], leaf["holds"]) == (False, False)
|
|
assert leaf["detail"] == "CSV header incomplete within the read cap"
|