e2e mirror of GeiserX/OwnGenome#319

docs/output-contract.md

@@ -10,7 +10,7 @@ The full list is one file, [`assets/output_contract.tsv`](https://github.com/Gei
 - **`--tools` names:** every name `main.nf` accepts (`pharmcat`, `clinvar`, `vcfanno`, ...).
 - **`run-all.sh`:** its two arguments (`<sample> <male|female>`) and every switch in its `SWITCHES` list (`TOOLS`, `THREADS`, `SKIP_TRIM`, `INTERVALS`, `REF_FASTA`, ...).
 - **Result files:** every file the Nextflow pipeline publishes, which is also what `run-all.sh` writes, at its path under `GENOME_DIR/<sample>/` (or `--outdir`). For each one the contract gives its step, its format and the fields a reader can count on: the columns of a table, the INFO and FORMAT fields of a VCF header, the top-level keys of a JSON file (`summary.json`'s keys are the ones its schema, `tests/schema/summary.schema.json`, requires).
-- **Single-step scripts:** the files listed with leg `bash` or `both`. Some single steps write to other folders than the pipeline does (step 7 writes PharmCAT's reports to `vcf/`, the pipeline to `pharmcat/`; step 16b writes to `mosdepth/`, the pipeline to `coverage/`). Both names are kept. A single-step output that is not in the file is not part of the promise yet.
+- **Single-step scripts:** the files listed with leg `bash` or `both`. A row is `both` when the step's script (`scripts/<step>-*.sh`) writes the same path under `GENOME_DIR`. Some single steps write to other folders than the pipeline does (step 7 writes PharmCAT's reports to `vcf/`, the pipeline to `pharmcat/`; step 16b writes to `mosdepth/`, the pipeline to `coverage/`). Both names are kept. A single-step output that is not in the file is not part of the promise yet.
 
 ## What counts as a break
 
@@ -40,7 +40,7 @@ The engines that run by default, and every step that existed before 1.0, are `st
 
 Two checks read the contract, both with [`scripts/ci/check-output-contract.py`](https://github.com/GeiserX/OwnGenome/blob/main/scripts/ci/check-output-contract.py):
 
-- **On every pull request, with no run** (`tests/test_output_contract.py`, in the `unit-tests` job of the Guard workflow): every param of `nextflow_schema.json`, every `--tools` name of `main.nf` and every switch of `run-all.sh` has a row, and every row still names one. Every file row of a Nextflow process still matches a file that process writes and publishes, and every file a module publishes has a row. A removed or renamed stable item fails as `BREAKING`. A new one with no row fails too, with the message to add it, so the contract is always the full list. The test also plants each kind of break on a copy of the code (a renamed output, a removed param, a removed switch, a new tool and a new output with no row) and requires each to fail.
+- **On every pull request, with no run** (`tests/test_output_contract.py`, in the `unit-tests` job of the Guard workflow): every param of `nextflow_schema.json`, every `--tools` name of `main.nf` and every switch of `run-all.sh` has a row, and every row still names one. Every file row of a Nextflow process still matches a file that process writes and publishes, in the folder of a `publishDir` that keeps it (a `saveAs` that returns `null` for the file does not count), and every file a module publishes has a row. A published row with leg `nextflow` fails when its step's script writes the same path too, since its leg is then `both`; the check reads the script with its shell variables filled in, so a path it cannot follow is not flagged, and a `both` row it does not find is not refused. Every default in `nextflow_schema.json` must equal the value in `nextflow.config`. A removed or renamed stable item fails as `BREAKING`. A new one with no row fails too, with the message to add it, so the contract is always the full list. The test also plants each kind of break on a copy of the code (a renamed output, a removed param, a removed switch, a new tool and a new output with no row, a schema default that differs from `nextflow.config`, a row in the folder a `saveAs` drops the file from, and a `nextflow` row a step script also writes) and requires each to fail.
 - **On the HG002 end-to-end run** (`tests/e2e/output-contract-1-hg002.sh`): every stable file whose `e2e` column names a leg must be in the results that leg wrote, under its contracted name, with its required fields. The bash leg is the single-step scripts the base cases run; the Nextflow leg is case `60-nextflow`. Rows whose step the fixture run does not reach have `e2e` set to `-` and are held by the static check only. The two files `run-all.sh` writes itself, `run_manifest.tsv` and `logs/run_status.tsv`, are listed but no check holds them yet: neither the fixture run nor a module writes them. The same case copies the Nextflow results, drops the `clnsig` column of the ClinVar hits table and renames the ROH summary, and requires the check to fail on both. It also prints every file each leg wrote with its fields, which is what to read when the contract is written again.
 
 ## Changing the contract

main.nf

@@ -316,9 +316,8 @@ workflow {
     // leading family column: one family runs, the one --family_name names
     // (needed when the file holds more than one). Here only the family and
     // sample columns are read; FAMILY_PREPARE (bin/family_report.py prepare)
-    // checks the rest before the joint call starts. Read with containsKey,
-    // so a run without the option is not told about an undefined parameter.
-    def pedigree   = params.containsKey('pedigree') ? params.pedigree : null
+    // checks the rest before the joint call starts.
+    def pedigree   = params.pedigree
     // The family's folder under --outdir: --family_name, else 'family'
     def family_dir = params.family_name ?: 'family'
     def family_ids = []

nextflow.config

@@ -121,6 +121,7 @@ params {
     viral_ref          = null       // viral: the viral genomes of step 47 (setup.sh --viral-data: reference/viral-<version>/viral_genomes.fa)
     hpo                = null       // exomiser: the symptoms as HPO ids, comma separated (HP:0001250,HP:0001263); setting it runs exomiser (step 48)
     exomiser_data      = null       // exomiser: Exomiser's data folder (setup.sh --exomiser-data: reference/exomiser-<release>)
+    pedigree           = null       // opt-in: a pedigree or families file (assets/pedigree.example.tsv, assets/families.example.tsv); runs the family workflow (steps 38, 39)
     family_name        = null       // --pedigree: the family to run of a families file (assets/families.example.tsv) that holds more than one; also its folder under --outdir (default family)
     exomiser_proband   = null       // exomiser: the sample the --hpo terms describe; needed when the samplesheet has more than one
     exomiser_assembly  = 'hg38'     // exomiser: the assembly of its data; hg19 only for Exomiser's cut-down test release (CI)

nextflow_schema.json

@@ -105,7 +105,7 @@
                 "tools": {
                     "type": "string",
                     "description": "Comma-separated list of tools to run; an unknown name stops the run. DB-heavy tools (vep, slivar, clinical_filter, cpsr, clinvar, expansion_hunter, sample_qc) and extras (stranger, ancestry (needs --ancestry_ref and prs), ancestry_fine (needs --ancestry_fine_panel), ancient (needs --ancient_panel; samples with a gVCF), pypgx, SV callers, cram_archive, cyrius, parascopy, paraphase (PacBio HiFi BAMs only), y_haplogroup (male samples), y_str (Y-STR repeat counts, male samples), consumer_export (samples with a gVCF), bloodtype (blood group antigens; RhD needs the BAM), lpa (the LPA KIV-2 repeat count; needs the BAM), vntyper (MUC1 VNTR frameshifts behind ADTKD-MUC1; needs the BAM), viral (viral DNA in the unmapped reads; needs --viral_ref and a BAM), exomiser (runs when --hpo is set), chip (clonal haematopoiesis from the BAM; needs --chip_gff), archaic (needs --archaic_data), mei (Alu, L1 and SVA insertions with MEGAnE; needs --megane_kmers and a BAM), mocha (mosaic chromosomal alterations; needs --mocha_data), locityper (a second opinion at HLA class II, KIR, CYP2D6 and FCGR; samples with a BAM; needs --locityper_data)) are opt-in. prs and vcfanno are skipped with a warning until --pgs_scoring or a score file is set. With pharmcat, PharmCAT also gets the HLA types of hla_typing and a CYP2D6 call pypgx and cyrius agree on (PGX_CONSENSUS).",
-                    "default": "pharmcat,cpic,vcfanno,roh,prs,mito_haplogroup,telomere_hunter,mosdepth,mito_variants,html_report,multiqc",
+                    "default": "pharmcat,cpic,vcfanno,roh,prs,mito_haplogroup,telomere_hunter,mosdepth,mito_variants,traits,html_report,multiqc",
                     "fa_icon": "fas fa-toolbox"
                 }
             }
@@ -344,6 +344,12 @@
                     "description": "Exomiser's data folder, which `scripts/setup.sh --exomiser-data` installs as reference/exomiser-<release> (its <release>_hg38 and <release>_phenotype folders). Required with --hpo.",
                     "fa_icon": "fas fa-folder"
                 },
+                "pedigree": {
+                    "type": "string",
+                    "format": "file-path",
+                    "description": "Opt-in: run the family workflow (steps 38 and 39) on the samples of the samplesheet this file names: a pedigree (assets/pedigree.example.tsv), or a families file with a leading family column (assets/families.example.tsv). A families file that holds more than one family needs --family_name, and only that family runs. The family that runs needs two or more of those samples, each with a gVCF, and --somalier_sites.",
+                    "fa_icon": "fas fa-sitemap"
+                },
                 "family_name": {
                     "type": "string",
                     "pattern": "^[a-zA-Z0-9._][a-zA-Z0-9._-]*$",

scripts/ci/check-output-contract.py

@@ -8,10 +8,14 @@
       run: every param of nextflow_schema.json, every --tools name of main.nf
       (known_tools) and every run-all.sh switch (its SWITCHES list) has a row,
       and every row still names one; every file row of a process names a file
-      that process still writes and publishes; and every file a module
-      publishes has a row. A stable row that no longer matches is a break. A
-      new param, tool, switch or published file with no row fails too, so the
-      contract stays the full list.
+      that process still writes and publishes, in the folder of a publishDir
+      that keeps it; every file a module publishes has a row; a nextflow row
+      (not internal) whose step script writes the same path is refused (its
+      legs are both);
+      and every default of nextflow_schema.json equals nextflow.config's. A
+      stable row that no longer matches is a break. A new param, tool, switch
+      or published file with no row fails too, so the contract stays the full
+      list.
 
   --outputs DIR --leg bash|nextflow --sample S [--observe]
       The files a run wrote. DIR is the folder that holds <sample>/ (GENOME_DIR
@@ -109,8 +113,49 @@ def concrete(pattern, sample="SAMPLE"):
     return p.strip("/")
 
 
+SAVE_AS = re.compile(r"^\s*(\w+)\s*->\s*\1\s*(?:\.(endsWith|startsWith)\(\s*'([^']*)'\s*\)|==\s*'([^']*)')"
+                     r"\s*\?\s*([^?:]+?)\s*:\s*([^?:]+?)\s*$", re.S)
+
+
+def closure_body(stmt):
+    """The body of a publishDir's saveAs closure, up to its own closing brace (an option
+    after it is not part of it), or None when the statement has no saveAs."""
+    m = re.search(r"saveAs:\s*\{", stmt)
+    if not m:
+        return None
+    depth, i = 1, m.end()
+    while i < len(stmt):
+        depth += {"{": 1, "}": -1}.get(stmt[i], 0)
+        if depth == 0:
+            return stmt[m.end():i]
+        i += 1
+    return stmt[m.end():]
+
+
+def save_as(closure, name):
+    """What a publishDir's saveAs closure does to a task file name (an output glob):
+    'keep' (published under its own name), 'drop' (returns null) or 'rename'.
+    A closure this reader does not follow counts as 'rename', which the static
+    check matches by basename only."""
+    if closure is None:
+        return "keep"
+    m = SAVE_AS.match(closure)
+    if not m:
+        return "rename"
+    var, method, arg, eq, yes, no = m.groups()
+    if method == "endsWith":
+        cond = name.endswith(arg)
+    elif method == "startsWith":
+        cond = name.startswith(arg)
+    else:
+        cond = name == eq
+    expr = (yes if cond else no).strip()
+    return "drop" if expr == "null" else "keep" if expr == var else "rename"
+
+
 def module_processes(repo):
-    """{process: {'publish': [(dir, [patterns] or None, has_saveas)], 'outputs': [glob]}} of modules/local."""
+    """{process: {'publish': [(dir, [patterns] or None, saveAs closure body or None)], 'outputs': [glob]}}
+    of modules/local."""
     procs = {}
     for f in sorted(glob.glob(os.path.join(repo, "modules", "local", "*", "main.nf"))):
         src = open(f).read()
@@ -134,7 +179,7 @@ def module_processes(repo):
                 pat = re.search(r"pattern:\s*[\"']([^\"']+)[\"']", stmt)
                 pubs.append((concrete(d.group(1)) if d else "",
                              braces(pat.group(1)) if pat else None,
-                             "saveAs" in stmt))
+                             closure_body(stmt)))
             out = re.search(r"^\s*output:\s*$(.*?)(?:^\s*(?:when|topic):|\Z)", head, re.M | re.S)
             outputs = []
             if out:
@@ -147,15 +192,19 @@ def module_processes(repo):
 
 
 def published(proc):
-    """(contracted-path glob, basename-only) pairs a process publishes; basename-only when a saveAs renames."""
+    """(output glob, contracted-path glob, basename-only) of each file a process publishes. A publishDir
+    whose saveAs drops the file does not count; basename-only when its saveAs renames the file."""
     items = []
-    for d, pats, saveas in proc["publish"]:
+    for d, pats, closure in proc["publish"]:
         for g in proc["outputs"]:
             c = concrete(g)
             if pats is not None and not any(fnmatch.fnmatch(c.replace("*", "X"), p) or fnmatch.fnmatch(c, p)
                                             for p in pats):
                 continue
-            items.append((g, (d + "/" + c) if d else c, saveas))
+            action = save_as(closure, c)
+            if action == "drop":
+                continue
+            items.append((g, (d + "/" + c) if d else c, action == "rename"))
     return items
 
 
@@ -165,6 +214,82 @@ def schema_params(repo):
     return {p for group in defs.values() for p in group.get("properties", {})}
 
 
+def schema_defaults(repo):
+    s = json.load(open(os.path.join(repo, "nextflow_schema.json")))
+    defs = s.get("definitions") or s.get("$defs") or {}
+    return {p: v["default"] for group in defs.values() for p, v in group.get("properties", {}).items()
+            if "default" in v}
+
+
+def config_params(repo):
+    """{name: value} of nextflow.config's params block, for the plain literals (strings, numbers,
+    true, false, null); a value this reader cannot parse is left out."""
+    src = open(os.path.join(repo, "nextflow.config")).read()
+    m = re.search(r"^params\s*\{(.*?)^\}", src, re.M | re.S)
+    out = {}
+    if not m:
+        return out
+    lit = {"true": True, "false": False, "null": None}
+    for name, val in re.findall(r"^\s*(\w+)\s*=\s*('[^']*'|\"[^\"]*\"|[-\w.]+)", m.group(1), re.M):
+        if val[0] in "'\"":
+            out[name] = val[1:-1]
+        elif val in lit:
+            out[name] = lit[val]
+        else:
+            try:
+                out[name] = int(val)
+            except ValueError:
+                try:
+                    out[name] = float(val)
+                except ValueError:
+                    pass
+    return out
+
+
+VAR = re.compile(r"\$\{(\w+)\}|\$(\w+)")
+ASSIGN = re.compile(r"^\s*(?:local\s+|export\s+|readonly\s+)?([A-Za-z_]\w*)=(\"[^\"\n]*\"|'[^'\n]*'|[^\s;#()]+)", re.M)
+
+
+def bash_paths(src):
+    """The lines of a step script with its shell variables expanded: SAMPLE as {sample},
+    GENOME_DIR as G, every other variable as each value it is assigned (a few at most)."""
+    assign = {}
+    for v, val in ASSIGN.findall(src):
+        assign.setdefault(v, []).append(val.strip("\"'"))
+    fixed = {"SAMPLE": "{sample}", "GENOME_DIR": "G", "1": "{sample}"}
+
+    # depth counts every variable filled in on the line, not the nesting: a line with
+    # ${WORK}, ${P} and P's own chain of OUTDIR, S, GENOME_DIR and SAMPLE needs seven.
+    # The search stops at 64 expansions of a line, so a deep line stays cheap.
+    def expand(s, depth=0):
+        m = VAR.search(s)
+        if not m or depth > 24:
+            return [s]
+        name = m.group(1) or m.group(2)
+        vals = [fixed[name]] if name in fixed else assign.get(name) or ["\0"]
+        out = []
+        for val in vals[:4]:
+            out.extend(expand(s[:m.start()] + val + s[m.end():], depth + 1))
+            if len(out) >= 64:
+                break
+        return out[:64]
+
+    return {e for line in src.split("\n") if not line.lstrip().startswith("#") for e in expand(line)}
+
+
+def bash_writes(repo, step, name, cache):
+    """The step script that names this per-sample path (scripts/<step>-*.sh), or None."""
+    if not name.startswith("{sample}/") or "*" in name or step == "-":
+        return None
+    if step not in cache:
+        files = sorted(glob.glob(os.path.join(repo, "scripts", f"{step}-*.sh")))
+        cache[step] = (files, "\n".join(bash_paths("".join(open(f).read() for f in files))) if files else "")
+    files, text = cache[step]
+    # the whole path: a longer name that starts with it (x.tsv.gz for x.tsv) is not it
+    hit = files and re.search(re.escape("G/" + name) + r"(?![\w.-])", text)
+    return os.path.relpath(files[0], repo) if hit else None
+
+
 def known_tools(repo):
     src = open(os.path.join(repo, "main.nf")).read()
     m = re.search(r"def known_tools = \[(.*?)\]", src, re.S)
@@ -193,6 +318,14 @@ def check_static(repo, rows):
         for name in sorted(have[kind] - set(listed)):
             problems.append(f"{kind} {name} of {where[kind]} has no row in the contract: add one")
 
+    defaults, config = schema_defaults(repo), config_params(repo)
+    if not config:
+        problems.append("found no params in nextflow.config: the reader no longer matches the code")
+    for name, val in sorted(defaults.items()):
+        if name in config and config[name] != val:
+            problems.append(f"param {name}: nextflow_schema.json's default {val!r} is not nextflow.config's "
+                            f"{config[name]!r}")
+
     procs = module_processes(repo)
     if not procs:
         problems.append("found no processes under modules/local")
@@ -214,6 +347,13 @@ def check_static(repo, rows):
             what = "BREAKING: the stable file" if r["stability"] == "stable" else "the file"
             problems.append(f"{what} {r['name']} (contract line {r['line']}) is not what {r['process']} "
                             f"publishes ({p['file']}): renamed, moved or no longer written")
+    cache = {}
+    for r in rows:
+        if r["kind"] == "file" and r["legs"] == "nextflow" and r["stability"] != "internal":
+            script = bash_writes(repo, r["step"], r["name"], cache)
+            if script:
+                problems.append(f"{r['name']} (contract line {r['line']}) has legs nextflow, but {script} "
+                                "writes it under the same name too: its legs are both")
     for name, p in sorted(procs.items()):
         rows_p = [concrete(r["name"]) for r in file_rows if r["process"] == name]
         if not p["publish"]: