From bfdbaf2c8232431b0d922f69e7f3bc0b4cff777d Mon Sep 17 00:00:00 2001 From: hyperpolymath <6759885+hyperpolymath@users.noreply.github.com> Date: Wed, 14 Oct 2026 18:08:27 +0000 Subject: [PATCH] test(pipeline): run the vsearch stage through the server's entry points The refactor left 3 lines in src/server/routes/pipeline.jl uncovered, and said so in its own message: the call into _vsearch_stage and the OTU read, inside _run_full_pipeline and _run_stage, which no test runs. This closes that. The reason no test ran them is that none of them called the server's own entry points. test/integration/test_pipeline.jl built its run directories by calling the stages by hand, which is not what pressing a stage in the UI does: the UI calls _run_full_pipeline and _run_stage, and those two calls are the only callers of _vsearch_stage outside the occupancy test. The new testset re-runs the vsearch stage both ways, on the runs the loop left behind: - run_A through _run_stage(..., "vsearch", ...): the serial shape; - run_B through _run_full_pipeline(...): the concurrent shape. Each vsearch output is removed first. A stage whose outputs are current skips before its work begins, and a stage that skips records nothing, so a re-run that skipped would test the skip guard instead of the stage. The stages before vsearch still skip on their own guards, so neither call redoes the DADA2 work. The probe is switched on for both, at the narrowest level of the config cascade - the run's own pipeline.yml, which is the way the defaults file says to switch it on, the config editor offering only keys that have a factory default. occupancy_probe sits in no stage's hashed section, so turning it on marks nothing stale. What the test asserts is what each run records: the vsearch stage, and that stage's occupancy, one labelled task per search. Every line the refactor added to src/server/routes/pipeline.jl is now executed by the suite: 19 of 19, where it was 16. Co-authored-by: arena-agent <297053741+arena-agent@users.noreply.github.com> --- test/integration/test_pipeline.jl | 70 +++++++++++++++++++++++++++++++ 1 file changed, 70 insertions(+) diff --git a/test/integration/test_pipeline.jl b/test/integration/test_pipeline.jl index 00d5fca8..bd3b5d6f 100644 --- a/test/integration/test_pipeline.jl +++ b/test/integration/test_pipeline.jl @@ -235,4 +235,74 @@ end end + ## The server's own entry points, on the run directories the loop above built. + # Nothing up to here has called them: the loop drives the stages by hand, which + # is not what pressing a stage in the UI does. The full pipeline runs the + # vsearch stage's two searches concurrently and a single-stage re-run runs them + # one after the other (_vsearch_searches) - the two shapes whose peaks the + # occupancy probe exists to tell apart - and these calls are the only callers of + # that helper outside test/occupancy_disagreement_test.jl. + # + # Each vsearch output is removed first. A stage whose outputs are current skips + # before doing any work, and a stage that skips records nothing: a re-run that + # skipped would exercise the skip guard rather than the stage. The stages before + # vsearch still skip on their own guards, so neither call redoes the DADA2 work. + # + # The probe is switched on for both, at the narrowest level of the config + # cascade (the run's own pipeline.yml) - the way the factory defaults file says + # to switch it on, since the config editor offers only keys that have a factory + # default. It sits in no stage's hashed section, so turning it on marks nothing + # stale, and it is on only here: every other call in the suite records exactly + # the stage keys it always did. + @testset "Pipeline and stage routes (vsearch re-run)" begin + SV = MetaManifold.Server + PV = MetaManifold.Provenance + MP = MetaManifold.MemoryProbe + run_dir(run) = joinpath(PROJECT_ROOT, "projects", PROJECT_NAME, run) + probe_cfg = Dict("occupancy_probe" => Dict("enabled" => true, "interval_ms" => 10)) + old_root = SV.ServerState._root[] + written = String[] + try + SV.ServerState.set_root!(PROJECT_ROOT) + + # run_A: one stage on its own - the serial shape. + rm(joinpath(run_dir("run_A"), "vsearch"); recursive=true, force=true) + rm(joinpath(run_dir("run_A"), "swarm", "vsearch"); recursive=true, force=true) + yaml_a = joinpath(PROJECT_ROOT, "data", PROJECT_NAME, "run_A", "pipeline.yml") + YAML.write_file(yaml_a, probe_cfg) + push!(written, yaml_a) + SV._run_stage(PROJECT_NAME, "run_A", "vsearch", db_config, dbs_loaded) + @test isfile(joinpath(run_dir("run_A"), "vsearch", "taxonomy.tsv")) + @test isfile(joinpath(run_dir("run_A"), "swarm", "vsearch", "taxonomy.tsv")) + + # run_B: the whole pipeline - the concurrent shape. + rm(joinpath(run_dir("run_B"), "vsearch"); recursive=true, force=true) + rm(joinpath(run_dir("run_B"), "swarm", "vsearch"); recursive=true, force=true) + yaml_b = joinpath(PROJECT_ROOT, "data", PROJECT_NAME, "run_B", "pipeline.yml") + YAML.write_file(yaml_b, probe_cfg) + push!(written, yaml_b) + merged = SV._run_full_pipeline(PROJECT_NAME, "run_B", db_config, dbs_loaded) + @test haskey(merged.tables, "merged") + @test isfile(joinpath(run_dir("run_B"), "vsearch", "taxonomy.tsv")) + @test isfile(joinpath(run_dir("run_B"), "swarm", "vsearch", "taxonomy.tsv")) + + # What each run recorded: its vsearch stage, and - the probe being on - + # that stage's occupancy, one labelled task per search. + for run in ("run_A", "run_B") + att = PV.read_attestation(joinpath(run_dir(run), "attestation.yml")) + stage = only([s for s in att.doc["stages"] if s["name"] == "vsearch"]) + @test haskey(stage, "occupancy") + occ = get(stage, "occupancy", nothing) + isnothing(occ) && continue + @test occ["probe"] == MP.PROBE + @test occ["interval_ms"] == 10 + labels = sort([String(t["label"]) for t in occ["tasks"]]) + @test labels == ["vsearch:asv", "vsearch:otu"] + end + finally + foreach(p -> rm(p; force=true), written) + SV.ServerState._root[] = old_root + end + end + end