projectAIpopular/tests/test_worker.py

"""T5 WorkerLoop + 接地验证单测（封闭，注入假 generate / 假沙箱 runner）。"""
import pytest

from router_system.knowledge import KnowledgeBase
from router_system.verifier import (
    Verifier,
    detect_artifact_language,
    extract_code_block,
    run_code_sandbox,
)
from router_system.worker import WorkerLoop, artifact_name_for
from router_system.workspace import Workspace


def _brief(domain="code"):
    return {
        "goal": "实现快排",
        "constraints": ["标准库"],
        "tags": [domain],
        "acceptance": [{"id": "a1", "check": "可运行", "machine_checkable": True}],
        "plan": [
            {"id": "s1", "task": "实现快排", "deps": [], "done_criteria": "函数可运行"},
            {"id": "s2", "task": "写测试", "deps": ["s1"], "done_criteria": "3/3 通过"},
        ],
    }


def _ws(domain="code"):
    ws = Workspace.new(request_id="abc123def456", query="写个快排",
                       api_token_cap=8000, rounds_cap=6)
    ws.apply_brief(_brief(domain))
    return ws


def _fake_gen(texts):
    """顺序返回预设生成文本的假 generate。"""
    it = iter(texts)

    async def _g(prompt):
        try:
            return next(it)
        except StopIteration:
            return ""

    return _g


# ---------- verifier: 代码沙箱 ----------
def test_verifier_code_pass(tmp_path):
    v = Verifier(sandbox_runner=lambda cmd, env, t: (0, "ok", ""))
    passed, det = v.verify("code", "s1.py", "def f():\n    return 1\n", "q")
    assert passed is True
    assert any("沙箱运行通过" in d for d in det)


def test_verifier_code_fail_then_structure():
    # 沙箱失败 + 无 facts + 文本过短 -> False
    v = Verifier(sandbox_runner=lambda cmd, env, t: (1, "", "boom"))
    passed, det = v.verify("code", "s1.py", "def f(): pass", "q")
    assert passed is False


def test_verifier_facts_hit(kb):
    v = Verifier()
    # medical 域：回答命中"高血压"事实关键词
    passed, det = v.verify("medical", "s1.md",
                           "高血压患者应低盐低脂饮食、控制体重、规律运动", "高血压饮食", kb=kb)
    assert passed is True
    assert any("facts" in d for d in det)


def test_verifier_structure_pass():
    v = Verifier()
    passed, _ = v.verify("general", "s1.md", "这是一段足够长的、非占位的正常回答内容。", "q")
    assert passed is True


def test_verifier_placeholder_fail():
    v = Verifier()
    passed, det = v.verify("general", "s1.md", "待实现", "q")
    assert passed is False


# ---------- run_code_sandbox ----------
def test_run_code_sandbox_good_code():
    rc, out, err = run_code_sandbox("print(1 + 1)")
    assert rc == 0
    assert out.strip() == "2"


def test_run_code_sandbox_bad_code():
    rc, out, err = run_code_sandbox("raise ValueError('x')")
    assert rc != 0
    assert "ValueError" in (err or "")


# ---------- extract_code_block ----------
def test_extract_code_block():
    bt = chr(96) * 3
    text = f"前言\n{bt}python\ndef f(): return 1\n{bt}\n后记"
    assert extract_code_block(text) == "def f(): return 1"


def test_extract_code_block_no_fence():
    assert extract_code_block("print(1)") == "print(1)"


# ---------- detect_artifact_language ----------
def test_detect_language():
    assert detect_artifact_language("s1.py") == "python"
    assert detect_artifact_language("s1.json") == "json"
    assert detect_artifact_language("s1.md") == "text"


# ---------- artifact_name ----------
def test_artifact_name():
    assert artifact_name_for("s1", "code") == "s1.py"
    assert artifact_name_for("s1", "legal") == "s1.md"


# ---------- WorkerLoop 三路径 ----------
def test_worker_one_pass():
    code = "def quicksort(arr):\n    return sorted(arr)"
    ws = _ws()
    w = WorkerLoop(_fake_gen([code]),
                   verifier=Verifier(sandbox_runner=lambda c, e, t: (0, "", "")),
                   max_fix_attempts=2)
    out = asyncio_run(w.run_step(ws, "s1"))
    assert out.status == "done"
    assert out.attempts == 1
    assert out.artifact_text == code
    assert len(ws["progress"]) == 1
    assert ws["progress"][0]["status"] == "done"


def test_worker_self_fix_success():
    bad = "def quicksort(arr):\n    return arr[0]  # 错"
    good = "def quicksort(arr):\n    return sorted(arr)"
    # 第一次沙箱失败，第二次通过
    results = iter([(1, "", "IndexError"), (0, "", "")])

    def fake_run(cmd, env, t):
        return next(results)

    ws = _ws()
    w = WorkerLoop(_fake_gen([bad, good]),
                   verifier=Verifier(sandbox_runner=fake_run),
                   max_fix_attempts=2)
    out = asyncio_run(w.run_step(ws, "s1"))
    assert out.status == "done"
    assert out.attempts == 2
    assert len(ws["progress"]) == 1


def test_worker_issue_after_exhaust():
    bad = "def quicksort(arr):\n    return arr[0]"
    ws = _ws()
    w = WorkerLoop(_fake_gen([bad, bad]),
                   verifier=Verifier(sandbox_runner=lambda c, e, t: (1, "", "Err")),
                   max_fix_attempts=2)
    out = asyncio_run(w.run_step(ws, "s1"))
    assert out.status == "issue"
    assert out.issue_id is not None
    assert len(ws["issues"]) == 1
    assert ws["issues"][0]["step"] == "s1"


def test_worker_medical_facts_pass(kb):
    # 医疗域：模型给出含事实关键词的回答 -> facts 验证通过
    answer = "高血压患者应低盐低脂饮食、控制体重、规律运动、戒烟限酒"
    ws = _ws(domain="medical")
    w = WorkerLoop(_fake_gen([answer]), kb=kb, max_fix_attempts=2)
    out = asyncio_run(w.run_step(ws, "s1"))
    assert out.status == "done"


def asyncio_run(coro):
    import asyncio
    return asyncio.run(coro)