轻量拓展分析支持上游样本继承:同job内成功运行产物自动预填下游输入槽
- 新增 upstream.py:枚举 workspace/<job>/extension_runs/*/runtime_*/outputs 的成功产物,
按 basename 精确匹配优先、role 关键词辅助(多命中取最新 run)收敛为每槽一个候选
- 新增 GET /api/lightweight-extension-stages/{id}/upstream-candidates?job_id= 端点,
返回 workspace 相对形式的候选与来源信息;仅对契约 patch_targets 可补丁槽位启用改名匹配,
不可补丁槽位留空由用户手填,避免 PATCH_TARGET_UNAVAILABLE
- 前端运行向导:LLM slots 之后对空槽填充上游候选并显示绿色来源标记(可修改),
用户手动上传/导入时清除来源标记
- runtime_binding 支持目录输入:_resolve_inputs/_job_path 放宽目录,
物化走 copy_tree 整树拷贝 + _sha256_tree 目录摘要校验,isolate 跳过独立副本
- nl_intent prompt 规则 4 追加 extension_runs 上游产物引用指引
- 新增 20+ 单测覆盖上游发现/端点/目录输入/前端契约
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Fable 5
parent
abdd486ed8
commit
cb3732dc71
@@ -70,6 +70,96 @@ def _client(tmp_path: Path, service: _FakeService | None = None):
|
||||
return TestClient(app), queued
|
||||
|
||||
|
||||
def test_upstream_candidates_endpoint_returns_job_relative_paths(
|
||||
tmp_path: Path, monkeypatch
|
||||
) -> None:
|
||||
from gfa_core.extension_stage_lightweight.api import upstream
|
||||
|
||||
workspace_root = tmp_path / "workspace"
|
||||
job_root = workspace_root / "job_x" / "extension_runs"
|
||||
output = job_root / "ext_up" / "runtime_r1" / "outputs" / "samples.tsv"
|
||||
output.parent.mkdir(parents=True)
|
||||
output.write_text("x\n", encoding="utf-8")
|
||||
(job_root / "ext_up" / "runtime_r1" / "runtime_execution_result.json").write_text(
|
||||
'{"success": true}', encoding="utf-8"
|
||||
)
|
||||
|
||||
monkeypatch.setattr(
|
||||
upstream,
|
||||
"resolve_upstream_slots",
|
||||
lambda workspace_root, job_id, contract_inputs, exclude_extension_id="", patchable_paths=None: {
|
||||
"samples.tsv": upstream.UpstreamCandidate(
|
||||
path=output,
|
||||
relative="samples.tsv",
|
||||
source_extension_id="ext_up",
|
||||
run_id="runtime_r1",
|
||||
is_dir=False,
|
||||
run_time=0.0,
|
||||
)
|
||||
},
|
||||
)
|
||||
|
||||
service = _FakeService()
|
||||
service.help = lambda extension_id: {
|
||||
"extension_id": extension_id,
|
||||
"inputs": [
|
||||
{"path": "samples.tsv", "replaceable": True},
|
||||
{"path": "locked.tsv", "replaceable": False},
|
||||
],
|
||||
}
|
||||
client, _queued = _client(tmp_path, service)
|
||||
|
||||
response = client.get(
|
||||
"/api/lightweight-extension-stages/count_fasta/upstream-candidates",
|
||||
params={"job_id": "job_x"},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["extension_id"] == "count_fasta"
|
||||
assert data["job_id"] == "job_x"
|
||||
assert data["candidates"] == {
|
||||
"samples.tsv": {
|
||||
"path": "workspace/job_x/extension_runs/ext_up/runtime_r1/outputs/samples.tsv",
|
||||
"source_extension_id": "ext_up",
|
||||
"run_id": "runtime_r1",
|
||||
"is_dir": False,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
def test_upstream_candidates_endpoint_empty_job_id_returns_empty(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
client, _queued = _client(tmp_path, _FakeService())
|
||||
|
||||
response = client.get(
|
||||
"/api/lightweight-extension-stages/count_fasta/upstream-candidates"
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["candidates"] == {}
|
||||
assert data["job_id"] == ""
|
||||
|
||||
|
||||
def test_upstream_candidates_endpoint_unknown_extension_404(tmp_path: Path) -> None:
|
||||
service = _FakeService()
|
||||
|
||||
def _help_missing(extension_id: str) -> dict:
|
||||
raise ValueError("extension not found")
|
||||
|
||||
service.help = _help_missing
|
||||
client, _queued = _client(tmp_path, service)
|
||||
|
||||
response = client.get(
|
||||
"/api/lightweight-extension-stages/missing_ext/upstream-candidates",
|
||||
params={"job_id": "job_x"},
|
||||
)
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
def _receipt(tmp_path: Path, *, token: str = "validate-1") -> None:
|
||||
path = tmp_path / "store" / "operations" / f"{token}.json"
|
||||
path.parent.mkdir(parents=True)
|
||||
|
||||
@@ -300,3 +300,14 @@ def test_lightweight_nl_run_prefill_flag_wired() -> None:
|
||||
|
||||
assert "prefill" in source
|
||||
assert "已识别为轻量拓展分析" not in source # 消息文案由后端生成,前端按 taskType 路由
|
||||
|
||||
def test_lightweight_nl_run_wizard_loads_upstream_candidates() -> None:
|
||||
source = _source()
|
||||
|
||||
assert "loadUpstreamCandidates" in source
|
||||
assert "/upstream-candidates?job_id=" in source
|
||||
assert "source_extension_id" in source
|
||||
assert "上游样本" in source
|
||||
assert "input.source" in source
|
||||
assert "source: ''" in source
|
||||
|
||||
|
||||
@@ -242,6 +242,28 @@ def test_job_file_hints_capped_at_max(monkeypatch, tmp_path: Path) -> None:
|
||||
assert len(hints) == nl_intent.MAX_JOB_FILE_HINTS
|
||||
|
||||
|
||||
def test_job_file_hints_include_extension_runs_outputs(
|
||||
monkeypatch, tmp_path: Path
|
||||
) -> None:
|
||||
workspace_root = tmp_path / "workspace"
|
||||
_write(
|
||||
workspace_root,
|
||||
"job-h1/extension_runs/ext_a/runtime_r1/outputs/out.tsv",
|
||||
"a\tb\n",
|
||||
)
|
||||
monkeypatch.setattr(WEB_MAIN, "INPUT_ROOT", tmp_path / "input")
|
||||
monkeypatch.setattr(WEB_MAIN, "WORKSPACE_ROOT", workspace_root)
|
||||
|
||||
hints = WEB_MAIN._lightweight_job_file_hints("job-h1")
|
||||
|
||||
assert "workspace/job-h1/extension_runs/ext_a/runtime_r1/outputs/out.tsv" in hints
|
||||
|
||||
|
||||
def test_nl_intent_prompt_mentions_upstream_extension_runs() -> None:
|
||||
assert "extension_runs" in nl_intent._RESPONSE_SYSTEM_PROMPT
|
||||
assert "runtime_*/outputs" in nl_intent._RESPONSE_SYSTEM_PROMPT
|
||||
|
||||
|
||||
# ---------- 4. import-paths 端点 ----------
|
||||
|
||||
def _call_nl_import_paths(job_id: str, paths: list[str]):
|
||||
|
||||
@@ -76,6 +76,44 @@ def _outputs(result: dict) -> dict[str, dict]:
|
||||
return {item["fact_id"]: item for item in result["output_bindings"]}
|
||||
|
||||
|
||||
def test_resolves_directory_input(tmp_path: Path) -> None:
|
||||
directory = tmp_path / "proteomes"
|
||||
(directory / "01.fa").parent.mkdir(parents=True, exist_ok=True)
|
||||
(directory / "01.fa").write_text(">a\nAAAA\n", encoding="utf-8")
|
||||
(directory / "02.fa").write_text(">b\nBBBB\n", encoding="utf-8")
|
||||
contract = _contract(
|
||||
inputs=[_input("input:1", "proteomes/", "input directory")],
|
||||
outputs=[("output:1", "result.txt")],
|
||||
)
|
||||
|
||||
result = resolve_runtime_bindings(
|
||||
contract, {"input:1": directory}, {}
|
||||
)
|
||||
|
||||
assert result["status"] == "resolved"
|
||||
assert result["input_bindings"] == [
|
||||
{
|
||||
"original_path": "proteomes/",
|
||||
"runtime_path": str(directory),
|
||||
"fact_ids": ["input:1"],
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
def test_resolves_missing_directory_reports_input_path_not_found(tmp_path: Path) -> None:
|
||||
contract = _contract(
|
||||
inputs=[_input("input:1", "proteomes/", "input directory")],
|
||||
outputs=[("output:1", "result.txt")],
|
||||
)
|
||||
|
||||
result = resolve_runtime_bindings(
|
||||
contract, {"input:1": tmp_path / "missing_dir"}, {}
|
||||
)
|
||||
|
||||
assert result["status"] != "resolved"
|
||||
assert result["error"]["code"] == "INPUT_PATH_NOT_FOUND"
|
||||
|
||||
|
||||
def test_resolves_single_input_and_output_prefix(tmp_path: Path) -> None:
|
||||
replacement = _file(tmp_path, "Runtime.Sample.pep.fasta")
|
||||
contract = _contract(
|
||||
|
||||
@@ -1530,6 +1530,200 @@ def test_replay_isolates_symlink_input_before_source_overwrite(
|
||||
assert result["execution_work_tree"]["Ft.input.txt"]["type"] == "regular"
|
||||
|
||||
|
||||
def _contract_and_resolved_dir(
|
||||
raw_code: Path,
|
||||
runtime_dir: Path,
|
||||
*,
|
||||
script: str,
|
||||
) -> tuple[dict, dict]:
|
||||
"""目录输入版的契约+解析构造:input 为目录,走 copy_tree 物化。"""
|
||||
import shlex
|
||||
|
||||
from gfa_core.extension_stage_lightweight.runtime_binding import (
|
||||
_sha256_tree,
|
||||
)
|
||||
|
||||
script_path = _write(raw_code, "run.sh", script)
|
||||
script_hash = _sha256_bytes(script_path.read_bytes())
|
||||
commands = _prepare_script_commands(script, [])
|
||||
targets = []
|
||||
for index, command in enumerate(commands, start=1):
|
||||
for token_index, token in enumerate(shlex.split(command)):
|
||||
if token in {"proteomes", "proteomes/"}:
|
||||
targets.append(
|
||||
{
|
||||
"binding_key": "path:proteomes",
|
||||
"original_literal": token,
|
||||
"kind": "input_path",
|
||||
"script_path": "run.sh",
|
||||
"script_hash": script_hash,
|
||||
"expected_exact_match_count": 1,
|
||||
"fact_ids": ["input:1"],
|
||||
"anchors": [
|
||||
{
|
||||
"fact_id": "input:1",
|
||||
"step_id": f"step_{index:03d}",
|
||||
"command_digest": _sha256_bytes(command.encode()),
|
||||
"token_index": token_index,
|
||||
}
|
||||
],
|
||||
}
|
||||
)
|
||||
contract = {
|
||||
"schema_version": "lightweight-runtime-binding/v1",
|
||||
"facts_sha256": "facts-locked",
|
||||
"package_digest": "",
|
||||
"original_command": "bash run.sh",
|
||||
"source_scripts": [
|
||||
{"path": "run.sh", "sha256": script_hash, "locked": True}
|
||||
],
|
||||
"package_files": [
|
||||
{
|
||||
"path": "run.sh",
|
||||
"source": "raw_code",
|
||||
"kind": "file",
|
||||
"size": script_path.stat().st_size,
|
||||
"sha256": script_hash,
|
||||
"locked": True,
|
||||
"exposed_as_input": False,
|
||||
},
|
||||
],
|
||||
"inputs": [
|
||||
{
|
||||
"fact_id": "input:1",
|
||||
"path": "proteomes",
|
||||
"source": "baseline_input",
|
||||
"replaceable": True,
|
||||
"role": "input directory",
|
||||
}
|
||||
],
|
||||
"parameters": [],
|
||||
"path_references": [],
|
||||
"outputs": {
|
||||
"files": [],
|
||||
"stdout": {"present": True},
|
||||
},
|
||||
"patch_targets": targets,
|
||||
"unpatchable_targets": [],
|
||||
}
|
||||
inventory = {
|
||||
item["path"]: {
|
||||
key: item[key]
|
||||
for key in ("kind", "size", "sha256", "source")
|
||||
}
|
||||
for item in contract["package_files"]
|
||||
}
|
||||
contract["package_digest"] = _sha256_json(inventory)
|
||||
_refresh_digest(contract, "contract_digest")
|
||||
resolved = resolve_runtime_bindings(
|
||||
contract,
|
||||
{"input:1": runtime_dir},
|
||||
{},
|
||||
)
|
||||
assert resolved["status"] == "resolved", resolved
|
||||
return contract, resolved
|
||||
|
||||
|
||||
def test_render_materializes_directory_input_as_copy_tree(tmp_path: Path) -> None:
|
||||
raw_code = tmp_path / "raw_code"
|
||||
runtime_dir = tmp_path / "runtime_proteomes"
|
||||
_write(runtime_dir, "01.fa", ">a\nAAAA\n")
|
||||
_write(runtime_dir, "02.fa", ">b\nBBBB\n")
|
||||
contract, resolved = _contract_and_resolved_dir(
|
||||
raw_code,
|
||||
runtime_dir,
|
||||
script="ls proteomes\n",
|
||||
)
|
||||
|
||||
rendered = render_runtime_package(
|
||||
contract, raw_code, resolved, tmp_path / "run"
|
||||
)
|
||||
|
||||
materialization = rendered["input_materialization"]
|
||||
assert len(materialization) == 1
|
||||
assert materialization[0]["method"] == "copy_tree"
|
||||
assert materialization[0]["materialized_path"] == "runtime_proteomes"
|
||||
assert len(materialization[0]["sha256"]) == 64
|
||||
materialized = tmp_path / "run/work/runtime_proteomes"
|
||||
assert materialized.is_dir()
|
||||
assert not materialized.is_symlink()
|
||||
assert (materialized / "01.fa").read_text(encoding="utf-8") == ">a\nAAAA\n"
|
||||
patched = (tmp_path / "run/work/run.sh").read_text(encoding="utf-8")
|
||||
assert "ls runtime_proteomes" in patched
|
||||
|
||||
|
||||
def test_replay_executes_directory_input_and_isolates_copy(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
import json
|
||||
|
||||
raw_code = tmp_path / "raw_code"
|
||||
runtime_dir = tmp_path / "runtime_proteomes"
|
||||
_write(runtime_dir, "01.fa", ">a\nAAAA\n")
|
||||
_write(runtime_dir, "02.fa", ">b\nBBBB\n")
|
||||
contract, resolved = _contract_and_resolved_dir(
|
||||
raw_code,
|
||||
runtime_dir,
|
||||
script="ls proteomes\n",
|
||||
)
|
||||
|
||||
rendered = render_runtime_package(
|
||||
contract, raw_code, resolved, tmp_path / "run"
|
||||
)
|
||||
result = replay_runtime_binding(rendered)
|
||||
|
||||
assert result["contract_passed"] is True
|
||||
assert result["stdout_contract_passed"] is True
|
||||
# 执行期间工具写改 work 内副本不影响 source
|
||||
assert (runtime_dir / "01.fa").read_text(encoding="utf-8") == ">a\nAAAA\n"
|
||||
events = json.loads(
|
||||
(tmp_path / "run/logs/execution_events.jsonl")
|
||||
.read_text(encoding="utf-8")
|
||||
.splitlines()[0]
|
||||
)
|
||||
assert events["execution_inputs"][0]["mode"] == "isolated_copy"
|
||||
assert events["execution_inputs"][0]["materialized_path"] == "runtime_proteomes"
|
||||
|
||||
|
||||
def test_replay_rejects_directory_input_content_drift(tmp_path: Path) -> None:
|
||||
raw_code = tmp_path / "raw_code"
|
||||
runtime_dir = tmp_path / "runtime_proteomes"
|
||||
_write(runtime_dir, "01.fa", ">a\nAAAA\n")
|
||||
contract, resolved = _contract_and_resolved_dir(
|
||||
raw_code,
|
||||
runtime_dir,
|
||||
script="ls proteomes\n",
|
||||
)
|
||||
|
||||
rendered = render_runtime_package(
|
||||
contract, raw_code, resolved, tmp_path / "run"
|
||||
)
|
||||
_write(runtime_dir, "03.fa", ">c\nCCCC\n") # 渲染后 source 被改动
|
||||
|
||||
with pytest.raises(RuntimeBindingError) as caught:
|
||||
replay_runtime_binding(rendered)
|
||||
|
||||
assert caught.value.code == "INPUT_CONTENT_CHANGED"
|
||||
|
||||
|
||||
def test_sha256_tree_is_stable_and_skips_symlinks(tmp_path: Path) -> None:
|
||||
from gfa_core.extension_stage_lightweight.runtime_binding import (
|
||||
_sha256_tree,
|
||||
)
|
||||
|
||||
source = tmp_path / "tree"
|
||||
_write(source, "a.txt", "one\n")
|
||||
_write(source, "sub/b.txt", "two\n")
|
||||
first = _sha256_tree(source)
|
||||
assert first == _sha256_tree(source)
|
||||
|
||||
(source / "link.txt").symlink_to(source / "a.txt")
|
||||
assert _sha256_tree(source) == first # symlink 不参与摘要
|
||||
|
||||
_write(source, "sub/c.txt", "three\n")
|
||||
assert _sha256_tree(source) != first # 新增文件改变摘要
|
||||
|
||||
|
||||
def test_shell_word_replacement_preserves_python_string_literal_quotes() -> None:
|
||||
"""zh01 回归:python -c 里带引号的输出路径被前缀替换后必须保留引号。
|
||||
|
||||
|
||||
@@ -168,6 +168,31 @@ def test_published_run_requires_job_context_and_projects_into_job_workspace(
|
||||
)
|
||||
|
||||
|
||||
def test_job_path_accepts_directory_inside_job_and_rejects_outside(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
from gfa_core.extension_stage_lightweight.service import LightweightExtensionService
|
||||
|
||||
input_root = tmp_path / "input"
|
||||
workspace_root = tmp_path / "workspace"
|
||||
job_input = input_root / "job_1"
|
||||
job_workspace = workspace_root / "job_1"
|
||||
(job_input / "proteomes").mkdir(parents=True)
|
||||
(job_workspace / ".keep").mkdir(parents=True)
|
||||
|
||||
inside = LightweightExtensionService._job_path(
|
||||
job_input / "proteomes", (job_input, job_workspace)
|
||||
)
|
||||
assert inside.is_dir()
|
||||
|
||||
outside_dir = tmp_path / "outside"
|
||||
outside_dir.mkdir()
|
||||
with pytest.raises(ValueError, match="selected job"):
|
||||
LightweightExtensionService._job_path(
|
||||
outside_dir, (job_input, job_workspace)
|
||||
)
|
||||
|
||||
|
||||
def test_published_run_rejects_symlink_escape_from_selected_job(tmp_path: Path) -> None:
|
||||
service = _published_copy_service(tmp_path)
|
||||
input_root = tmp_path / "input"
|
||||
|
||||
@@ -0,0 +1,329 @@
|
||||
"""轻量拓展分析"上游样本继承"发现模块单元测试。
|
||||
|
||||
用 tmp_path 构造 workspace/<job>/extension_runs/ 假树(写入 runtime_execution_result.json
|
||||
的 success 字段与 outputs/ 产物),验证候选发现与槽位匹配的保守语义。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
from gfa_core.extension_stage_lightweight import upstream
|
||||
|
||||
|
||||
def _write(root: Path, relative: str, content: str) -> Path:
|
||||
path = root / relative
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(content, encoding="utf-8")
|
||||
return path
|
||||
|
||||
|
||||
def _result(root: Path, success: bool) -> Path:
|
||||
return _write(
|
||||
root,
|
||||
"runtime_execution_result.json",
|
||||
json.dumps({"success": success, "contract_passed": success, "returncode": 0}),
|
||||
)
|
||||
|
||||
|
||||
def _run_tree(
|
||||
runs: Path,
|
||||
ext_id: str,
|
||||
run_id: str,
|
||||
*,
|
||||
outputs: list[str],
|
||||
success: bool = True,
|
||||
) -> Path:
|
||||
"""建一个 run 目录并返回 run_dir;outputs 为相对 outputs/ 的文件路径。"""
|
||||
run_dir = runs / ext_id / run_id
|
||||
for relative in outputs:
|
||||
_write(run_dir / "outputs", relative, "content:" + relative)
|
||||
_result(run_dir, success)
|
||||
return run_dir
|
||||
|
||||
|
||||
def _job_root(tmp_path: Path) -> Path:
|
||||
workspace = tmp_path / "workspace"
|
||||
_write(workspace, "job_1/.keep", "")
|
||||
return workspace / "job_1" / "extension_runs"
|
||||
|
||||
|
||||
# ---------- 候选发现 ----------
|
||||
|
||||
|
||||
def test_collect_skips_failed_and_missing_result_runs(tmp_path: Path) -> None:
|
||||
runs = _job_root(tmp_path)
|
||||
_run_tree(runs, "ext_a", "runtime_good", outputs=["a.tsv"])
|
||||
_run_tree(runs, "ext_a", "runtime_failed", outputs=["b.tsv"], success=False)
|
||||
missing = _run_tree(runs, "ext_a", "runtime_missing", outputs=["c.tsv"])
|
||||
(missing / "runtime_execution_result.json").unlink()
|
||||
|
||||
pool = upstream.collect_job_upstream_outputs(tmp_path / "workspace", "job_1")
|
||||
|
||||
names = {candidate.path.name for candidate in pool}
|
||||
assert names == {"a.tsv"}
|
||||
|
||||
|
||||
def test_collect_skips_excluded_extension(tmp_path: Path) -> None:
|
||||
runs = _job_root(tmp_path)
|
||||
_run_tree(runs, "ext_self", "runtime_1", outputs=["a.tsv"])
|
||||
_run_tree(runs, "ext_other", "runtime_2", outputs=["b.tsv"])
|
||||
|
||||
pool = upstream.collect_job_upstream_outputs(
|
||||
tmp_path / "workspace", "job_1", exclude_extension_id="ext_self"
|
||||
)
|
||||
|
||||
sources = {candidate.source_extension_id for candidate in pool}
|
||||
assert sources == {"ext_other"}
|
||||
|
||||
|
||||
def test_collect_ignores_non_runtime_run_dirs(tmp_path: Path) -> None:
|
||||
runs = _job_root(tmp_path)
|
||||
_run_tree(runs, "ext_a", "runtime_1", outputs=["a.tsv"])
|
||||
# 标准拓展快照风格的 run_* 目录:glob("runtime_*") 天然不命中
|
||||
snapshot = runs / "ext_b" / "run_20260826_120000"
|
||||
_write(snapshot / "outputs", "legacy.tsv", "x")
|
||||
_write(snapshot, "extension_run.json", json.dumps({"status": "success"}))
|
||||
|
||||
pool = upstream.collect_job_upstream_outputs(tmp_path / "workspace", "job_1")
|
||||
|
||||
names = {candidate.path.name for candidate in pool}
|
||||
assert names == {"a.tsv"}
|
||||
|
||||
|
||||
def test_collect_empty_workspace_returns_empty(tmp_path: Path) -> None:
|
||||
assert upstream.collect_job_upstream_outputs(tmp_path / "workspace", "job_none") == []
|
||||
missing_ws = tmp_path / "no_workspace"
|
||||
assert upstream.collect_job_upstream_outputs(missing_ws, "job_none") == []
|
||||
|
||||
|
||||
def test_collect_symlinks_are_skipped(tmp_path: Path) -> None:
|
||||
runs = _job_root(tmp_path)
|
||||
run_dir = _run_tree(runs, "ext_a", "runtime_1", outputs=["real.tsv"])
|
||||
(run_dir / "outputs" / "link.tsv").symlink_to(run_dir / "outputs" / "real.tsv")
|
||||
|
||||
pool = upstream.collect_job_upstream_outputs(tmp_path / "workspace", "job_1")
|
||||
|
||||
names = {candidate.path.name for candidate in pool}
|
||||
assert names == {"real.tsv"}
|
||||
|
||||
|
||||
# ---------- 槽位匹配 ----------
|
||||
|
||||
|
||||
def test_exact_basename_match(tmp_path: Path) -> None:
|
||||
runs = _job_root(tmp_path)
|
||||
_run_tree(runs, "ext_a", "runtime_1", outputs=["samples.tsv"])
|
||||
_run_tree(runs, "ext_b", "runtime_2", outputs=["other.tsv"])
|
||||
|
||||
found = upstream.find_upstream_candidates(
|
||||
tmp_path / "workspace",
|
||||
"job_1",
|
||||
[{"path": "samples.tsv", "replaceable": True}],
|
||||
)
|
||||
|
||||
assert "samples.tsv" in found
|
||||
(candidate,) = found["samples.tsv"]
|
||||
assert candidate.path.name == "samples.tsv"
|
||||
assert candidate.source_extension_id == "ext_a"
|
||||
|
||||
|
||||
def test_latest_run_wins(tmp_path: Path) -> None:
|
||||
runs = _job_root(tmp_path)
|
||||
newer = _run_tree(runs, "ext_a", "runtime_new", outputs=["dup.tsv"])
|
||||
older = _run_tree(runs, "ext_b", "runtime_old", outputs=["dup.tsv"])
|
||||
stamp = time.time()
|
||||
os.utime(newer, (stamp, stamp))
|
||||
os.utime(older, (stamp - 100, stamp - 100))
|
||||
|
||||
selected = upstream.resolve_upstream_slots(
|
||||
tmp_path / "workspace",
|
||||
"job_1",
|
||||
[{"path": "dup.tsv", "replaceable": True}],
|
||||
)
|
||||
|
||||
assert selected["dup.tsv"].run_id == "runtime_new"
|
||||
|
||||
|
||||
def test_shortest_path_wins_within_same_run(tmp_path: Path) -> None:
|
||||
runs = _job_root(tmp_path)
|
||||
_run_tree(runs, "ext_a", "runtime_1", outputs=["deep/nested/dup.tsv", "dup.tsv"])
|
||||
|
||||
selected = upstream.resolve_upstream_slots(
|
||||
tmp_path / "workspace",
|
||||
"job_1",
|
||||
[{"path": "dup.tsv", "replaceable": True}],
|
||||
)
|
||||
|
||||
assert selected["dup.tsv"].relative == "dup.tsv"
|
||||
|
||||
|
||||
def test_ambiguous_role_match_yields_no_candidate(tmp_path: Path) -> None:
|
||||
runs = _job_root(tmp_path)
|
||||
_run_tree(runs, "ext_a", "runtime_1", outputs=["apple_output.tsv"])
|
||||
_run_tree(runs, "ext_b", "runtime_2", outputs=["banana_output.tsv"])
|
||||
|
||||
found = upstream.find_upstream_candidates(
|
||||
tmp_path / "workspace",
|
||||
"job_1",
|
||||
[{"path": "wanted.tsv", "role": "output", "replaceable": True}],
|
||||
)
|
||||
|
||||
assert "wanted.tsv" not in found
|
||||
|
||||
|
||||
def test_unique_role_match_yields_candidate(tmp_path: Path) -> None:
|
||||
runs = _job_root(tmp_path)
|
||||
_run_tree(runs, "ext_a", "runtime_1", outputs=["q1.tsv"])
|
||||
_run_tree(runs, "ext_b", "runtime_2", outputs=["homology_unique_name.tsv"])
|
||||
|
||||
found = upstream.find_upstream_candidates(
|
||||
tmp_path / "workspace",
|
||||
"job_1",
|
||||
[{"path": "wanted.tsv", "role": "homology", "replaceable": True}],
|
||||
)
|
||||
|
||||
assert "wanted.tsv" in found
|
||||
assert found["wanted.tsv"][0].path.name == "homology_unique_name.tsv"
|
||||
|
||||
|
||||
def test_multi_token_role_requires_all_tokens(tmp_path: Path) -> None:
|
||||
runs = _job_root(tmp_path)
|
||||
_run_tree(runs, "ext_a", "runtime_1", outputs=["homology_hits.tsv"])
|
||||
|
||||
found = upstream.find_upstream_candidates(
|
||||
tmp_path / "workspace",
|
||||
"job_1",
|
||||
# "homology" 唯一命中,但 "absentword" 零命中 → 不全命中,不采纳
|
||||
[{"path": "wanted.tsv", "role": "homology absentword", "replaceable": True}],
|
||||
)
|
||||
|
||||
assert "wanted.tsv" not in found
|
||||
|
||||
|
||||
def test_role_multi_hit_takes_latest_run(tmp_path: Path) -> None:
|
||||
runs = _job_root(tmp_path)
|
||||
newer = _run_tree(runs, "ext_a", "runtime_new", outputs=["fresh.hits.tsv"])
|
||||
older = _run_tree(runs, "ext_b", "runtime_old", outputs=["stale.hits.tsv"])
|
||||
stamp = time.time()
|
||||
os.utime(newer, (stamp, stamp))
|
||||
os.utime(older, (stamp - 100, stamp - 100))
|
||||
|
||||
selected = upstream.resolve_upstream_slots(
|
||||
tmp_path / "workspace",
|
||||
"job_1",
|
||||
[{"path": "wanted.tsv", "role": "hmm hits", "replaceable": True}],
|
||||
)
|
||||
|
||||
assert "wanted.tsv" in selected
|
||||
assert selected["wanted.tsv"].path.name == "fresh.hits.tsv"
|
||||
assert selected["wanted.tsv"].run_id == "runtime_new"
|
||||
|
||||
|
||||
def test_unpatchable_slot_skips_role_match(tmp_path: Path) -> None:
|
||||
runs = _job_root(tmp_path)
|
||||
_run_tree(runs, "ext_a", "runtime_1", outputs=["fresh.hits.tsv"])
|
||||
|
||||
found = upstream.find_upstream_candidates(
|
||||
tmp_path / "workspace",
|
||||
"job_1",
|
||||
[{"path": "hmm_hits.tsv", "role": "hmm hits", "replaceable": True}],
|
||||
patchable_paths={"01.col.primary.pep.fa"}, # 槽位不在可补丁集
|
||||
)
|
||||
|
||||
assert "hmm_hits.tsv" not in found
|
||||
|
||||
|
||||
def test_exact_name_match_ignores_patchable_filter(tmp_path: Path) -> None:
|
||||
runs = _job_root(tmp_path)
|
||||
_run_tree(runs, "ext_a", "runtime_1", outputs=["samples.tsv"])
|
||||
|
||||
found = upstream.find_upstream_candidates(
|
||||
tmp_path / "workspace",
|
||||
"job_1",
|
||||
[{"path": "samples.tsv", "role": "manifest", "replaceable": True}],
|
||||
patchable_paths=set(), # 精确同名不受可补丁过滤影响
|
||||
)
|
||||
|
||||
assert "samples.tsv" in found
|
||||
|
||||
|
||||
def test_directory_slot_matches_directory_not_file(tmp_path: Path) -> None:
|
||||
runs = _job_root(tmp_path)
|
||||
_run_tree(runs, "ext_a", "runtime_1", outputs=["proteomes/01.fa", "proteomes/02.fa"])
|
||||
|
||||
found = upstream.find_upstream_candidates(
|
||||
tmp_path / "workspace",
|
||||
"job_1",
|
||||
[{"path": "proteomes/", "replaceable": True, "role": "input directory"}],
|
||||
)
|
||||
|
||||
assert "proteomes/" in found
|
||||
(candidate,) = found["proteomes/"]
|
||||
assert candidate.is_dir is True
|
||||
assert candidate.path.name == "proteomes"
|
||||
|
||||
|
||||
def test_file_slot_ignores_same_named_directory(tmp_path: Path) -> None:
|
||||
runs = _job_root(tmp_path)
|
||||
_run_tree(runs, "ext_a", "runtime_1", outputs=["03_homology/x.tsv"])
|
||||
|
||||
found = upstream.find_upstream_candidates(
|
||||
tmp_path / "workspace",
|
||||
"job_1",
|
||||
[{"path": "03_homology", "replaceable": True, "role": "homology table"}],
|
||||
)
|
||||
|
||||
assert "03_homology" not in found
|
||||
|
||||
|
||||
def test_replaceable_filter_and_candidate_exclusivity(tmp_path: Path) -> None:
|
||||
runs = _job_root(tmp_path)
|
||||
_run_tree(runs, "ext_a", "runtime_1", outputs=["shared.tsv"])
|
||||
|
||||
selected = upstream.resolve_upstream_slots(
|
||||
tmp_path / "workspace",
|
||||
"job_1",
|
||||
[
|
||||
{"path": "shared.tsv", "replaceable": True},
|
||||
{"path": "shared.tsv", "replaceable": True},
|
||||
{"path": "locked.tsv", "replaceable": False},
|
||||
],
|
||||
)
|
||||
|
||||
assert set(selected) == {"shared.tsv"} # 同 path 第二个槽被占用后不再填;locked 不参与
|
||||
|
||||
|
||||
def test_duplicate_slot_path_keeps_latest_run(tmp_path: Path) -> None:
|
||||
runs = _job_root(tmp_path)
|
||||
newer = _run_tree(runs, "ext_a", "runtime_new", outputs=["hit.tsv"])
|
||||
older = _run_tree(runs, "ext_a", "runtime_old", outputs=["hit.tsv"])
|
||||
stamp = time.time()
|
||||
os.utime(newer, (stamp, stamp))
|
||||
os.utime(older, (stamp - 100, stamp - 100))
|
||||
|
||||
selected = upstream.resolve_upstream_slots(
|
||||
tmp_path / "workspace",
|
||||
"job_1",
|
||||
[
|
||||
{"path": "hit.tsv", "replaceable": True},
|
||||
{"path": "hit.tsv", "replaceable": True}, # 契约中同名不同 fact_id 的重复槽
|
||||
],
|
||||
)
|
||||
|
||||
assert selected["hit.tsv"].run_id == "runtime_new"
|
||||
|
||||
|
||||
def test_no_upstream_returns_empty(tmp_path: Path) -> None:
|
||||
runs = _job_root(tmp_path) # extension_runs 目录不存在
|
||||
assert (
|
||||
upstream.resolve_upstream_slots(
|
||||
tmp_path / "workspace",
|
||||
"job_1",
|
||||
[{"path": "samples.tsv", "replaceable": True}],
|
||||
)
|
||||
== {}
|
||||
)
|
||||
assert not runs.exists()
|
||||
Reference in New Issue
Block a user