522 lines
20 KiB
Python
522 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
"""run_selected 的标准库离线回归测试。
|
|
|
|
测试项目、manifest 和 .venv/bin/python 均在临时目录中生成，不连接数据库、网络或模型。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import stat
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from contextlib import redirect_stdout
|
|
from pathlib import Path
|
|
from typing import Any, Iterable
|
|
|
|
try:
|
|
from .run_selected import main
|
|
except ImportError: # 允许直接执行 `.venv/bin/python muse/lifecycle/quality/harness/test_run_selected.py`
|
|
from run_selected import main
|
|
|
|
|
|
class RunSelectedTests(unittest.TestCase):
|
|
def make_project(
|
|
self,
|
|
entries: Iterable[dict[str, Any]],
|
|
*,
|
|
generated_scope: str | None = None,
|
|
test_asset_scan_roots: list[str] | None = None,
|
|
) -> tuple[tempfile.TemporaryDirectory[str], Path]:
|
|
temporary = tempfile.TemporaryDirectory()
|
|
root = Path(temporary.name)
|
|
(root / ".venv" / "bin").mkdir(parents=True)
|
|
fake_python = root / ".venv" / "bin" / "python"
|
|
fake_python.write_text(
|
|
f"#!{sys.executable}\n"
|
|
"import pathlib\n"
|
|
"import sys\n"
|
|
"import time\n"
|
|
"script = pathlib.Path(sys.argv[1])\n"
|
|
"(pathlib.Path.cwd() / 'invocations.log').open('a', encoding='utf-8').write(script.name + '\\n')\n"
|
|
"if script.name == 'empty.py':\n"
|
|
" raise SystemExit(0)\n"
|
|
"if script.name == 'skip.py':\n"
|
|
" print('1 skipped')\n"
|
|
" raise SystemExit(0)\n"
|
|
"if script.name == 'fail.py':\n"
|
|
" print('failure stdout')\n"
|
|
" print('failure stderr', file=sys.stderr)\n"
|
|
" raise SystemExit(7)\n"
|
|
"if script.name == 'sleep.py':\n"
|
|
" time.sleep(5)\n"
|
|
"if script.name == 'noisy.py':\n"
|
|
" print('x' * 5000)\n"
|
|
"else:\n"
|
|
" print('ran ' + script.name)\n",
|
|
encoding="utf-8",
|
|
)
|
|
fake_python.chmod(fake_python.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
|
|
|
|
manifest_entries = []
|
|
for entry in entries:
|
|
entry_copy = dict(entry)
|
|
path = root / str(entry_copy["path"])
|
|
if entry_copy.pop("create", True):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text("# temporary test script\n", encoding="utf-8")
|
|
manifest_entries.append(entry_copy)
|
|
|
|
manifest_path = root / "manifest.json"
|
|
manifest_payload: dict[str, Any] = {
|
|
"schema_version": 1,
|
|
"entries": manifest_entries,
|
|
}
|
|
if generated_scope is not None:
|
|
manifest_payload["generated_scope"] = generated_scope
|
|
if test_asset_scan_roots is not None:
|
|
manifest_payload["test_asset_scan_roots"] = test_asset_scan_roots
|
|
manifest_path.write_text(
|
|
json.dumps(manifest_payload, indent=2) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
self.addCleanup(temporary.cleanup)
|
|
return temporary, manifest_path
|
|
|
|
def invoke(self, root: Path, manifest: Path, *arguments: str) -> tuple[int, dict[str, Any]]:
|
|
output = io.StringIO()
|
|
with redirect_stdout(output):
|
|
return_code = main(
|
|
[
|
|
"--root",
|
|
str(root),
|
|
"--manifest",
|
|
str(manifest),
|
|
*arguments,
|
|
"--json",
|
|
]
|
|
)
|
|
return return_code, json.loads(output.getvalue())
|
|
|
|
@staticmethod
|
|
def entry(
|
|
path: str,
|
|
*,
|
|
owner: str = "alpha",
|
|
kind: str = "tool_unit",
|
|
scope: str = "runtime_skill",
|
|
requires: list[str] | None = None,
|
|
create: bool = True,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"path": path,
|
|
"owner_skill_or_domain": owner,
|
|
"kind": kind,
|
|
"scope": scope,
|
|
"requires": ["offline"] if requires is None else requires,
|
|
"create": create,
|
|
}
|
|
|
|
def test_selector_alias_and_intersection_are_reported_as_json(self) -> None:
|
|
_, manifest = self.make_project(
|
|
[
|
|
self.entry("pass.py", owner="alpha", kind="tool_unit"),
|
|
self.entry("other.py", owner="beta", kind="tool_contract"),
|
|
]
|
|
)
|
|
root = manifest.parent
|
|
|
|
return_code, report = self.invoke(
|
|
root,
|
|
manifest,
|
|
"--skill",
|
|
"alpha",
|
|
"--kind",
|
|
"tool_unit",
|
|
"--path",
|
|
"pass.py",
|
|
)
|
|
|
|
self.assertEqual(return_code, 0)
|
|
self.assertEqual(report["status"], "passed")
|
|
self.assertEqual(report["selected_count"], 1)
|
|
self.assertEqual(report["entries"][0]["path"], "pass.py")
|
|
self.assertEqual(report["entries"][0]["returncode"], 0)
|
|
self.assertIn("ran pass.py", report["entries"][0]["stdout"])
|
|
|
|
def test_offline_dependency_is_blocked_without_running_child(self) -> None:
|
|
_, manifest = self.make_project(
|
|
[self.entry("db.py", requires=["offline", "postgresql"])]
|
|
)
|
|
root = manifest.parent
|
|
|
|
return_code, report = self.invoke(root, manifest, "--path", "db.py")
|
|
|
|
result = report["entries"][0]
|
|
self.assertEqual(return_code, 1)
|
|
self.assertEqual(report["status"], "failed")
|
|
self.assertEqual(result["status"], "blocked_dependency")
|
|
self.assertEqual(result["requires"], ["offline", "postgresql"])
|
|
self.assertEqual(result["blocked_requires"], ["postgresql"])
|
|
self.assertFalse((root / "invocations.log").exists())
|
|
|
|
def test_allow_requires_runs_and_still_reports_dependency(self) -> None:
|
|
_, manifest = self.make_project(
|
|
[self.entry("db.py", requires=["offline", "postgresql"])]
|
|
)
|
|
root = manifest.parent
|
|
|
|
return_code, report = self.invoke(
|
|
root,
|
|
manifest,
|
|
"--path",
|
|
"db.py",
|
|
"--allow-requires",
|
|
"postgresql",
|
|
)
|
|
|
|
result = report["entries"][0]
|
|
self.assertEqual(return_code, 0)
|
|
self.assertEqual(result["status"], "passed")
|
|
self.assertEqual(result["requires"], ["offline", "postgresql"])
|
|
self.assertEqual((root / "invocations.log").read_text(encoding="utf-8"), "db.py\n")
|
|
|
|
def test_nonzero_and_timeout_have_distinct_structured_statuses(self) -> None:
|
|
_, manifest = self.make_project(
|
|
[self.entry("fail.py"), self.entry("sleep.py")]
|
|
)
|
|
root = manifest.parent
|
|
|
|
return_code, report = self.invoke(
|
|
root,
|
|
manifest,
|
|
"--path",
|
|
"fail.py",
|
|
"--path",
|
|
"sleep.py",
|
|
"--timeout-seconds",
|
|
"1.5",
|
|
)
|
|
|
|
self.assertEqual(return_code, 1)
|
|
results = {entry["path"]: entry for entry in report["entries"]}
|
|
self.assertEqual(results["fail.py"]["status"], "failed")
|
|
self.assertEqual(results["fail.py"]["returncode"], 7)
|
|
self.assertIn("failure stdout", results["fail.py"]["stdout"])
|
|
self.assertIn("failure stderr", results["fail.py"]["stderr"])
|
|
self.assertEqual(results["sleep.py"]["status"], "timeout")
|
|
self.assertIsNone(results["sleep.py"]["returncode"])
|
|
self.assertEqual(results["sleep.py"]["error"]["code"], "timeout")
|
|
|
|
def test_no_matches_and_missing_path_are_not_silent(self) -> None:
|
|
_, manifest = self.make_project(
|
|
[
|
|
self.entry("missing.py", create=False),
|
|
self.entry("present.py"),
|
|
]
|
|
)
|
|
root = manifest.parent
|
|
|
|
return_code, no_match = self.invoke(root, manifest, "--kind", "does_not_exist")
|
|
self.assertEqual(return_code, 1)
|
|
self.assertEqual(no_match["status"], "no_matches")
|
|
self.assertEqual(no_match["entries"], [])
|
|
self.assertEqual(no_match["issues"][0]["code"], "no_matches")
|
|
|
|
return_code, missing = self.invoke(root, manifest, "--path", "missing.py")
|
|
self.assertEqual(return_code, 1)
|
|
self.assertEqual(missing["entries"][0]["status"], "not_found")
|
|
self.assertEqual(missing["entries"][0]["error"]["code"], "path_not_found")
|
|
|
|
def test_invalid_manifest_and_selector_requirement_are_structured(self) -> None:
|
|
temporary = tempfile.TemporaryDirectory()
|
|
self.addCleanup(temporary.cleanup)
|
|
root = Path(temporary.name)
|
|
manifest = root / "manifest.json"
|
|
manifest.write_text("{broken", encoding="utf-8")
|
|
|
|
return_code, invalid_manifest = self.invoke(root, manifest, "--path", "x.py")
|
|
self.assertEqual(return_code, 1)
|
|
self.assertEqual(invalid_manifest["status"], "manifest_invalid")
|
|
self.assertEqual(invalid_manifest["issues"][0]["code"], "manifest_invalid_json")
|
|
|
|
valid_root, valid_manifest = self.make_project([self.entry("pass.py")])
|
|
return_code, invalid_selector = self.invoke(valid_root, valid_manifest)
|
|
self.assertEqual(return_code, 1)
|
|
self.assertEqual(invalid_selector["status"], "invalid_selector")
|
|
self.assertEqual(invalid_selector["issues"][0]["code"], "selector_required")
|
|
|
|
def test_all_offline_is_explicit_opt_in_and_output_is_summary_only(self) -> None:
|
|
_, manifest = self.make_project(
|
|
[
|
|
self.entry("pass.py"),
|
|
self.entry("db.py", requires=["postgresql"]),
|
|
self.entry("noisy.py"),
|
|
]
|
|
)
|
|
root = manifest.parent
|
|
|
|
return_code, report = self.invoke(root, manifest, "--all-offline")
|
|
|
|
self.assertEqual(return_code, 1)
|
|
self.assertEqual(report["selected_count"], 3)
|
|
statuses = {entry["path"]: entry["status"] for entry in report["entries"]}
|
|
self.assertEqual(statuses["pass.py"], "passed")
|
|
self.assertEqual(statuses["db.py"], "blocked_dependency")
|
|
self.assertEqual(statuses["noisy.py"], "passed")
|
|
noisy = next(entry for entry in report["entries"] if entry["path"] == "noisy.py")
|
|
self.assertLess(len(noisy["stdout"]), 2001)
|
|
self.assertIn("output summary truncated", noisy["stdout"])
|
|
|
|
def test_generated_scope_rejects_unregistered_disk_asset(self) -> None:
|
|
_, manifest = self.make_project(
|
|
[self.entry("tests/skills/registered/test_registered.py")],
|
|
generated_scope="temporary test asset inventory",
|
|
)
|
|
root = manifest.parent
|
|
unregistered = root / "tests" / "skills" / "new" / "test_unregistered.py"
|
|
unregistered.parent.mkdir(parents=True)
|
|
unregistered.write_text("# unregistered test asset\n", encoding="utf-8")
|
|
|
|
return_code, report = self.invoke(root, manifest, "--all-offline")
|
|
|
|
self.assertEqual(return_code, 1)
|
|
self.assertEqual(report["status"], "manifest_invalid")
|
|
self.assertEqual(report["entries"], [])
|
|
self.assertEqual(
|
|
[
|
|
issue["path"]
|
|
for issue in report["issues"]
|
|
if issue["code"] == "manifest_test_asset_missing"
|
|
],
|
|
["tests/skills/new/test_unregistered.py"],
|
|
)
|
|
self.assertFalse((root / "invocations.log").exists())
|
|
|
|
def test_generated_scope_uses_manifest_scan_roots(self) -> None:
|
|
_, manifest = self.make_project(
|
|
[self.entry("custom-tests/test_registered.py")],
|
|
generated_scope="temporary test asset inventory",
|
|
test_asset_scan_roots=["custom-tests"],
|
|
)
|
|
root = manifest.parent
|
|
(root / "custom-tests" / "test_registered.py").write_text(
|
|
"def test_registered():\n assert True\n", encoding="utf-8"
|
|
)
|
|
|
|
return_code, report = self.invoke(root, manifest, "--all-offline")
|
|
|
|
self.assertEqual(return_code, 0)
|
|
self.assertEqual(
|
|
report["manifest"]["test_asset_scan_roots"], ["custom-tests"]
|
|
)
|
|
|
|
def test_generated_scope_rejects_manifest_extra_asset(self) -> None:
|
|
_, manifest = self.make_project(
|
|
[
|
|
self.entry(
|
|
"tests/skills/removed/test_removed.py",
|
|
create=False,
|
|
)
|
|
],
|
|
generated_scope="temporary test asset inventory",
|
|
)
|
|
root = manifest.parent
|
|
|
|
return_code, report = self.invoke(root, manifest, "--all-offline")
|
|
|
|
self.assertEqual(return_code, 1)
|
|
self.assertEqual(report["status"], "manifest_invalid")
|
|
self.assertEqual(report["entries"], [])
|
|
self.assertEqual(
|
|
[
|
|
issue["path"]
|
|
for issue in report["issues"]
|
|
if issue["code"] == "manifest_test_asset_extra"
|
|
],
|
|
["tests/skills/removed/test_removed.py"],
|
|
)
|
|
self.assertFalse((root / "invocations.log").exists())
|
|
|
|
def test_generated_scope_accepts_existing_behavior_eval_entry(self) -> None:
|
|
eval_entry = self.entry(
|
|
"muse/lifecycle/quality/harness/evals/skills/demo/run_eval.py",
|
|
kind="skill_behavior_eval",
|
|
requires=["model"],
|
|
)
|
|
eval_entry["skill_behavior_eval"] = True
|
|
_, manifest = self.make_project(
|
|
[
|
|
self.entry("tests/skills/registered/test_registered.py"),
|
|
eval_entry,
|
|
],
|
|
generated_scope="temporary test asset inventory",
|
|
)
|
|
root = manifest.parent
|
|
registered_test = root / "tests" / "skills" / "registered" / "test_registered.py"
|
|
registered_test.write_text("def test_registered():\n assert True\n", encoding="utf-8")
|
|
(root / "muse" / "lifecycle" / "quality" / "harness" / "evals" / "skills" / "demo" / "run_eval.py").write_text(
|
|
"# eval entry\n", encoding="utf-8"
|
|
)
|
|
|
|
return_code, report = self.invoke(root, manifest, "--kind", "tool_unit")
|
|
|
|
self.assertEqual(return_code, 0)
|
|
self.assertEqual(report["status"], "passed")
|
|
self.assertEqual(report["selected_count"], 1)
|
|
self.assertEqual(
|
|
[issue for issue in report["issues"] if "eval" in issue["code"]], []
|
|
)
|
|
|
|
def test_generated_scope_rejects_missing_behavior_eval_entry(self) -> None:
|
|
eval_entry = self.entry(
|
|
"muse/lifecycle/quality/harness/evals/skills/demo/run_eval.py",
|
|
kind="skill_behavior_eval",
|
|
requires=["model"],
|
|
create=False,
|
|
)
|
|
eval_entry["skill_behavior_eval"] = True
|
|
_, manifest = self.make_project(
|
|
[
|
|
self.entry("tests/skills/registered/test_registered.py"),
|
|
eval_entry,
|
|
],
|
|
generated_scope="temporary test asset inventory",
|
|
)
|
|
root = manifest.parent
|
|
registered_test = root / "tests" / "skills" / "registered" / "test_registered.py"
|
|
registered_test.write_text("def test_registered():\n assert True\n", encoding="utf-8")
|
|
|
|
return_code, report = self.invoke(root, manifest, "--kind", "tool_unit")
|
|
|
|
self.assertEqual(return_code, 1)
|
|
self.assertEqual(report["status"], "manifest_invalid")
|
|
self.assertEqual(
|
|
[
|
|
issue["path"]
|
|
for issue in report["issues"]
|
|
if issue["code"] == "manifest_eval_entry_missing"
|
|
],
|
|
["muse/lifecycle/quality/harness/evals/skills/demo/run_eval.py"],
|
|
)
|
|
|
|
def test_generated_scope_ignores_non_test_helpers_under_test_roots(self) -> None:
|
|
_, manifest = self.make_project(
|
|
[self.entry("tests/skills/registered/test_registered.py")],
|
|
generated_scope="temporary test asset inventory",
|
|
)
|
|
root = manifest.parent
|
|
registered_test = root / "tests" / "skills" / "registered" / "test_registered.py"
|
|
registered_test.write_text("def test_registered():\n pass\n", encoding="utf-8")
|
|
(root / "tests" / "skills" / "registered" / "helper.py").write_text(
|
|
"VALUE = 1\n", encoding="utf-8"
|
|
)
|
|
(root / "humanization" / "tests" / "helper.py").parent.mkdir(
|
|
parents=True, exist_ok=True
|
|
)
|
|
(root / "humanization" / "tests" / "helper.py").write_text(
|
|
"VALUE = 2\n", encoding="utf-8"
|
|
)
|
|
|
|
return_code, report = self.invoke(root, manifest, "--all-offline")
|
|
|
|
self.assertEqual(return_code, 0)
|
|
self.assertEqual(report["status"], "passed")
|
|
self.assertEqual(report["manifest"]["test_assets_scanned"], 1)
|
|
self.assertEqual(report["entries"][0]["path"], "tests/skills/registered/test_registered.py")
|
|
|
|
def test_generated_scope_blocks_empty_and_print_only_scripts_before_child(self) -> None:
|
|
entries = [
|
|
self.entry("tests/skills/empty/test_empty.py"),
|
|
self.entry("tests/skills/print_only/test_print_only.py"),
|
|
]
|
|
_, manifest = self.make_project(
|
|
entries,
|
|
generated_scope="temporary test asset inventory",
|
|
)
|
|
root = manifest.parent
|
|
(root / "tests" / "skills" / "empty" / "test_empty.py").write_text(
|
|
"", encoding="utf-8"
|
|
)
|
|
(root / "tests" / "skills" / "print_only" / "test_print_only.py").write_text(
|
|
"print('not a test')\n", encoding="utf-8"
|
|
)
|
|
|
|
return_code, report = self.invoke(root, manifest, "--all-offline")
|
|
|
|
self.assertEqual(return_code, 1)
|
|
self.assertEqual(report["status"], "failed")
|
|
results = {entry["path"]: entry for entry in report["entries"]}
|
|
for path in (
|
|
"tests/skills/empty/test_empty.py",
|
|
"tests/skills/print_only/test_print_only.py",
|
|
):
|
|
self.assertEqual(results[path]["status"], "failed")
|
|
self.assertIsNone(results[path]["returncode"])
|
|
self.assertEqual(results[path]["error"]["code"], "test_shape_missing")
|
|
self.assertFalse((root / "invocations.log").exists())
|
|
|
|
def test_generated_scope_accepts_supported_python_shapes(self) -> None:
|
|
entries = [
|
|
self.entry("tests/skills/function/test_function.py"),
|
|
self.entry("tests/skills/class/test_class.py"),
|
|
self.entry("muse/lifecycle/quality/harness/test_main_entry.py"),
|
|
]
|
|
_, manifest = self.make_project(
|
|
entries,
|
|
generated_scope="temporary test asset inventory",
|
|
)
|
|
root = manifest.parent
|
|
(root / "tests" / "skills" / "function" / "test_function.py").write_text(
|
|
"def test_function():\n pass\n", encoding="utf-8"
|
|
)
|
|
(root / "tests" / "skills" / "class" / "test_class.py").write_text(
|
|
"import unittest\n\nclass Fixture(unittest.TestCase):\n pass\n",
|
|
encoding="utf-8",
|
|
)
|
|
(root / "muse" / "lifecycle" / "quality" / "harness" / "test_main_entry.py").write_text(
|
|
"if __name__ == \"__main__\":\n print(\"script\")\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
return_code, report = self.invoke(root, manifest, "--all-offline")
|
|
|
|
self.assertEqual(return_code, 0)
|
|
self.assertEqual(report["status"], "passed")
|
|
self.assertEqual(
|
|
[entry["status"] for entry in report["entries"]],
|
|
["passed", "passed", "passed"],
|
|
)
|
|
self.assertEqual(
|
|
(root / "invocations.log").read_text(encoding="utf-8").splitlines(),
|
|
["test_function.py", "test_class.py", "test_main_entry.py"],
|
|
)
|
|
|
|
def test_zero_exit_without_execution_evidence_fails_closed(self) -> None:
|
|
_, manifest = self.make_project(
|
|
[self.entry("empty.py"), self.entry("skip.py")]
|
|
)
|
|
root = manifest.parent
|
|
|
|
return_code, report = self.invoke(
|
|
root,
|
|
manifest,
|
|
"--path",
|
|
"empty.py",
|
|
"--path",
|
|
"skip.py",
|
|
)
|
|
|
|
self.assertEqual(return_code, 1)
|
|
results = {entry["path"]: entry for entry in report["entries"]}
|
|
for path in ("empty.py", "skip.py"):
|
|
self.assertEqual(results[path]["status"], "failed")
|
|
self.assertEqual(results[path]["returncode"], 0)
|
|
self.assertEqual(results[path]["error"]["code"], "no_execution_evidence")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|