Fix pytest for strict nextflow syntax

.github/workflows/pytest.yml

@@ -20,6 +20,11 @@ on:
   release:
     types: [published]
   workflow_dispatch:
+    inputs:
+      nextflow-version:
+        description: "Nextflow version to test against"
+        required: false
+        default: "latest-stable"
 
 # Cancel if a newer run with the same workflow name is queued
 concurrency:
@@ -46,16 +51,17 @@ jobs:
       tests: ${{ steps.list_tests.outputs.tests }}
 
   test:
-    name: Run ${{matrix.test}} with Python ${{ matrix.python-version }} on ubuntu-latest
+    name: Run ${{matrix.test}} with Python ${{ matrix.python-version }} and Nextflow ${{ matrix.nextflow-version }}
     needs: list_tests
     runs-on:
       - runs-on=${{ github.run_id }}-run-test
       - runner=4cpu-linux-x64
     strategy:
       matrix:
         # On main branch test with 3.10 and 3.14, otherwise just 3.10
-        python-version: ${{ github.base_ref == 'main' && fromJson('["3.10", "3.14"]') || fromJson('["3.10"]') }}
+        python-version: ${{ (github.base_ref == 'main' || github.event_name == 'push') && fromJson('["3.10", "3.14"]') || fromJson('["3.10"]') }}
         test: ${{ fromJson(needs.list_tests.outputs.tests).test }}
+        nextflow-version: ${{ (github.base_ref == 'main' || github.event_name == 'push') && fromJson('["25.10.4", "latest-stable"]') || fromJson('["25.10.4"]') }}
       fail-fast: false # run all tests even if one fails
     steps:
       - name: go to subdirectory and change nextflow workdir
@@ -84,18 +90,17 @@ jobs:
       - name: Set up Apptainer
         if: ${{ startsWith(matrix.test, 'pipelines/download/') }}
         uses: eWaterCycle/setup-apptainer@4bb22c52d4f63406c49e94c804632975787312b3 # v2.0.0
-        with:
-          apptainer-version: 1.3.4
-
-      - name: Get current date
-        id: date
-        run: echo "date=$(date +'%Y-%m')" >> $GITHUB_ENV
 
       - name: Install Nextflow
         uses: nf-core/setup-nextflow@b4ec1bc7c16a94435159de94a05253542fddf6ef # v3
+        with:
+          version: ${{ github.event.inputs.nextflow-version || matrix.nextflow-version }}
 
       - name: Install nf-test
         uses: nf-core/setup-nf-test@4069fbbaabe94c08faba4ad261bfa88225ba133f # v2
+        with:
+          version: 0.9.5
+          install-fast-diff: true
 
       - name: move coveragerc file up
         run: |

nf_core/components/components_test.py

@@ -201,7 +201,8 @@ def generate_snapshot(self) -> bool:
 
         # set verbose flag if self.verbose is True
         verbose = "--verbose --debug" if self.verbose else ""
-        update = "--update-snapshot" if self.update else ""
+        update_snapshot = self.update
+        update = "--update-snapshot" if update_snapshot else ""
         self.update = False  # reset self.update to False to test if the new snapshot is stable
         tag = f"subworkflows/{self.component_name}" if self.component_type == "subworkflows" else self.component_name
         profile = self.profile if self.profile else os.environ["PROFILE"]
@@ -221,8 +222,9 @@ def generate_snapshot(self) -> bool:
                 self.obsolete_snapshots = True
             # check if nf-test was successful
             if "Assertion failed:" in nftest_out.decode():
-                if "Different Snapshot:" not in nftest_err.decode():
-                    self.errors.append("Assertion failed.")
+                if "Different Snapshot:" in nftest_err.decode():
+                    return update_snapshot  # snapshot was updated return False only if we don't want to update the snapshot
+                self.errors.append("Assertion failed.")
                 return False
             elif "No tests to execute." in nftest_out.decode():
                 log.error("Nothing to execute. Is the file 'main.nf.test' missing?")

nf_core/pipelines/containers_utils.py

@@ -112,7 +112,13 @@ def generate_container_configs(
                 raise UserWarning("Failed to run `nextflow inspect`. Please check your Nextflow installation.")
 
             out, _ = cmd_out
-            out_json = json.loads(out)
+            out_str = out.decode("utf-8", errors="replace")
+            try:
+                # Newer Nextflow versions print [PIPELINE]/[WORKDIR] headers before and [SUCCESS] after the JSON
+                json_start = out_str.find("{")
+                out_json, _ = json.JSONDecoder().raw_decode(out_str, json_start if json_start >= 0 else 0)
+            except json.JSONDecodeError:
+                out_json = json.loads(out)
 
         except RuntimeError as e:
             log.error("Running 'nextflow inspect' failed with the following error:")

nf_core/pipelines/download/download.py

@@ -693,10 +693,25 @@ def run_nextflow_inspect(params_file: Path | None = None) -> dict[str, Any]:
         try:
             out_json = run_nextflow_inspect()
         except RuntimeError as e:
+            # Extract Nextflow stdout from the chained CalledProcessError (errors go to stdout in NF)
+            nf_stdout = getattr(e.__cause__, "output", None) or b""
+
+            # Nextflow >= 26.04 enforces strict process directive syntax and rejects old-style
+            # if/else container blocks with "Invalid process directive". Users need an older NF.
+            if b"Invalid process directive" in nf_stdout:
+                raise DownloadError(
+                    "nextflow inspect failed because the pipeline uses old-style process syntax "
+                    "that the default strict syntax of Nextflow >= 26.04 no longer accepts.\n"
+                    "Please downgrade to Nextflow 25.10 or lower to download this pipeline, "
+                    "or ask the pipeline maintainers to update their container directives."
+                ) from e
+
             # Some workflow revisions explicitly require an outdir parameter. If this is the
             # only issue, retry inspect with an ephemeral params file that provides one.
             if re.search(
-                r"missing required parameter\s*:\s*(?:--outdir|params\.outdir|outdir)\b", str(e), flags=re.IGNORECASE
+                r"missing required parameter\s*:\s*(?:--outdir|params\.outdir|outdir)\b",
+                nf_stdout.decode("utf-8", errors="replace"),
+                flags=re.IGNORECASE,
             ):
                 try:
                     with tempfile.TemporaryDirectory(prefix="nf-core-inspect-") as tmp_dir:

nf_core/pipelines/list.py

@@ -25,13 +25,44 @@ def _get_nextflow_assets_dir() -> Path:
     """Return the Nextflow assets directory used for local workflow caches."""
     nxf_assets = os.environ.get("NXF_ASSETS")
     if nxf_assets:
-        return Path(nxf_assets)
+        base = Path(nxf_assets)
+    elif nxf_home := os.environ.get("NXF_HOME"):
+        base = Path(nxf_home, "assets")
+    else:
+        base = Path(os.environ.get("HOME", ""), ".nextflow", "assets")
+
+    # Newer Nextflow versions store clones under assets/.repos/
+    repos_dir = base / ".repos"
+    if repos_dir.is_dir():
+        return repos_dir
+    return base
+
 
-    nxf_home = os.environ.get("NXF_HOME")
-    if nxf_home:
-        return Path(nxf_home, "assets")
+def _resolve_wf_path(path: Path) -> Path:
+    """Resolve the actual pipeline working tree for a given assets dir / workflow path.
 
-    return Path(os.environ.get("HOME", ""), ".nextflow", "assets")
+    Nextflow 26.04+ uses a worktree layout under .repos:
+        <org>/<pipeline>/clones/<sha>/   ← working tree
+        <org>/<pipeline>/bare/           ← bare git repo
+    Prefers the clone matching the bare repo's HEAD; falls back to the most
+    recently modified clone if HEAD's clone is missing or the bare repo is unreadable.
+    Returns path unchanged for the old (pre-26.04) flat layout.
+    """
+    clones_dir = path / "clones"
+    bare_dir = path / "bare"
+    if clones_dir.is_dir():
+        if bare_dir.is_dir():
+            try:
+                sha = git.Repo(bare_dir).head.commit.hexsha
+                clone = clones_dir / sha
+                if clone.is_dir():
+                    return clone
+            except git.GitCommandError:
+                pass
+        sha_dirs = sorted(clones_dir.iterdir(), key=lambda p: p.stat().st_mtime)
+        if sha_dirs:
+            return sha_dirs[-1]
+    return path
 
 
 def list_workflows(filter_by=None, sort_by="release", as_json=False, show_archived=False):
@@ -79,7 +110,7 @@ def get_local_wf(workflow: str | Path, revision=None) -> Path | None:
         workflow = Path("nf-core", workflow)
 
     local_wf = LocalWorkflow(str(workflow))
-    local_wf_path = _get_nextflow_assets_dir() / workflow
+    local_wf_path = _resolve_wf_path(_get_nextflow_assets_dir() / workflow)
     if local_wf_path.is_dir():
         local_wf.local_path = local_wf_path
         local_wf.get_local_nf_workflow_details()
@@ -103,7 +134,11 @@ def get_local_wf(workflow: str | Path, revision=None) -> Path | None:
     pull_cmd = f"pull {workflow}"
     if revision is not None:
         pull_cmd += f" -r {revision}"
-    nf_core.utils.run_cmd("nextflow", pull_cmd)
+    try:
+        nf_core.utils.run_cmd("nextflow", pull_cmd)
+    except RuntimeError as e:
+        log.warning(f"Could not pull workflow '{workflow}': {e}")
+        return None
     local_wf = LocalWorkflow(str(workflow))
     local_wf.get_local_nf_workflow_details()
     return local_wf.local_path
@@ -191,7 +226,7 @@ def compare_remote_local(self):
                 if rwf.full_name == lwf.full_name:
                     rwf.local_wf = lwf
                     if rwf.releases:
-                        if rwf.releases[-1]["tag_sha"] == lwf.commit_sha:
+                        if rwf.releases[0]["tag_sha"] == lwf.commit_sha:
                             rwf.local_is_latest = True
                         else:
                             rwf.local_is_latest = False
@@ -229,7 +264,7 @@ def print_summary(self):
         if not self.sort_workflows_by or self.sort_workflows_by == "release":
             filtered_workflows.sort(
                 key=lambda wf: (
-                    (wf.releases[-1].get("published_at_timestamp", 0) if len(wf.releases) > 0 else 0) * -1,
+                    (wf.releases[0].get("published_at_timestamp", 0) if len(wf.releases) > 0 else 0) * -1,
                     wf.archived,
                     wf.full_name.lower(),
                 )
@@ -365,7 +400,7 @@ def get_local_nf_workflow_details(self):
 
         if self.local_path is None:
             # Try to guess the local cache directory
-            nf_wfdir = _get_nextflow_assets_dir() / self.full_name
+            nf_wfdir = _resolve_wf_path(_get_nextflow_assets_dir() / self.full_name)
             if nf_wfdir.is_dir():
                 log.debug(f"Guessed nextflow assets workflow directory: {nf_wfdir}")
                 self.local_path = nf_wfdir
@@ -379,7 +414,10 @@ def get_local_nf_workflow_details(self):
                     for key, pattern in re_patterns.items():
                         m = re.search(pattern, str(nfinfo_raw))
                         if m:
-                            setattr(self, key, m.group(1))
+                            value = Path(m.group(1)) if key == "local_path" else m.group(1)
+                            if key == "local_path":
+                                value = _resolve_wf_path(value)
+                            setattr(self, key, value)
 
         # Pull information from the local git repository
         if self.local_path is not None:
@@ -388,7 +426,8 @@ def get_local_nf_workflow_details(self):
                 repo = git.Repo(self.local_path)
                 self.commit_sha = str(repo.head.commit.hexsha)
                 self.remote_url = str(repo.remotes.origin.url)
-                self.last_pull = (self.local_path / ".git" / "FETCH_HEAD").stat().st_mtime
+                self.last_pull = Path(repo.common_dir) / "FETCH_HEAD"
+                self.last_pull = self.last_pull.stat().st_mtime
                 self.last_pull_date = datetime.fromtimestamp(self.last_pull).strftime("%Y-%m-%d %H:%M:%S")
                 self.last_pull_pretty = pretty_date(self.last_pull)
 
@@ -410,7 +449,7 @@ def get_local_nf_workflow_details(self):
                     f"Could not fetch status of local Nextflow copy of '{self.full_name}':"
                     f"\n   [red]{type(e).__name__}:[/] {str(e)}"
                     "\n\nThis git repository looks broken. It's probably a good idea to delete this local copy and pull again:"
-                    f"\n   [magenta]rm -rf {self.local_path}"
+                    f"\n   [magenta]rm -rf {_get_nextflow_assets_dir() / self.full_name}"
                     f"\n   [magenta]nextflow pull {self.full_name}",
                 )
 

nf_core/pipelines/schema.py

@@ -67,8 +67,10 @@ def _update_validation_plugin_from_config(self) -> None:
         plugin = "nf-schema"
         if self.schema_filename:
             conf = nf_core.utils.fetch_wf_config(Path(self.schema_filename).parent)
-        else:
+        elif self.pipeline_dir is not None:
             conf = nf_core.utils.fetch_wf_config(Path(self.pipeline_dir))
+        else:
+            return
 
         plugins = str(conf.get("plugins", "")).strip("'\"").strip(" ").split(",")
         plugin_found = False
@@ -137,7 +139,7 @@ def get_schema_path(self, path: str | Path, local_only: bool = False, revision:
             self.pipeline_dir = nf_core.pipelines.list.get_local_wf(path, revision=revision)
             self.schema_filename = Path(self.pipeline_dir or "", "nextflow_schema.json")
             # check if the schema file exists
-            if not self.schema_filename.exists():
+            if self.schema_filename is not None and not self.schema_filename.exists():
                 self.schema_filename = None
         # Only looking for local paths, overwrite with None to be safe
         else:

nf_core/utils.py

@@ -479,23 +479,21 @@ def run_cmd(executable: str, cmd: str) -> tuple[bytes, bytes] | None:
     full_cmd = f"{executable} {cmd}"
     log.debug(f"Running command: {full_cmd}")
     try:
-        proc = subprocess.run(shlex.split(full_cmd), capture_output=True, check=True)
+        proc = subprocess.run(shlex.split(full_cmd), capture_output=True, check=False)
+        if proc.returncode != 0:
+            if executable == "nf-test":
+                return (proc.stdout, proc.stderr)
+            raise subprocess.CalledProcessError(proc.returncode, proc.args, output=proc.stdout, stderr=proc.stderr)
         return (proc.stdout, proc.stderr)
+    except subprocess.CalledProcessError as e:
+        raise RuntimeError(f"Command '{full_cmd}' failed: {e}") from e
     except OSError as e:
         if e.errno == errno.ENOENT:
             raise RuntimeError(
                 f"It looks like {executable} is not installed. Please ensure it is available in your PATH."
             ) from e
         else:
             return None
-    except subprocess.CalledProcessError as e:
-        log.debug(f"Command '{full_cmd}' returned non-zero error code '{e.returncode}':\n[red]> {e.stderr.decode()}")
-        if executable == "nf-test":
-            return (e.stdout, e.stderr)
-        else:
-            raise RuntimeError(
-                f"Command '{full_cmd}' returned non-zero error code '{e.returncode}':\n[red]> {e.stderr.decode()}{e.stdout.decode()}"
-            ) from e
 
 
 def setup_nfcore_dir() -> bool: