Migrate pipeline lint rules to AST-grep

CHANGELOG.md

@@ -14,6 +14,7 @@
 ### Linting
 
 - accept `process_low_memory` as a standard module label ([#4264](https://github.com/nf-core/tools/pull/4264))
+- `pipeline_if_empty_null`, `system_exit` and `pipeline_todos` (incl. module/subworkflow todos) now match structurally with ast-grep and the tree-sitter-nextflow grammar when available (skips comments/strings), falling back to regex otherwise — per file, whenever a file has parse errors, so recall never drops below regex level. Match logic lives in ast-grep rule files under `nf_core/pipelines/lint/rules/` with snippet tests in `tests/pipelines/lint/rules/`
 - improve linting for `modules.json` to add support for only installing subworkflows from a repository and provide more explicit error messages ([#4287](https://github.com/nf-core/tools/pull/4287))
 - improve singularity_tag linting check (add skip, handle exceptions correctly, check against oras) ([#4358](https://github.com/nf-core/tools/pull/4358))
 

MANIFEST.in

@@ -8,5 +8,6 @@ include nf_core/assets/logo/nf-core-repo-logo-base-lightbg.png
 include nf_core/assets/logo/nf-core-repo-logo-base-darkbg.png
 include nf_core/assets/logo/placeholder_logo.svg
 include nf_core/assets/logo/MavenPro-Bold.ttf
+graft nf_core/pipelines/lint/rules
 include nf_core/pipelines/create/create.tcss
 include nf_core/pipelines/create/template_features.yml

nf_core/pipelines/lint/pipeline_if_empty_null.py

@@ -1,11 +1,13 @@
 import logging
-import re
 from pathlib import Path
 
+from nf_core import astgrep
 from nf_core.utils import get_wf_files
 
 log = logging.getLogger(__name__)
 
+RULE_FILE = Path(__file__).parent / "rules" / "pipeline_if_empty_null.yml"
+
 
 def pipeline_if_empty_null(self, root_dir=None):
     """Check for ifEmpty(null)
@@ -16,29 +18,23 @@ def pipeline_if_empty_null(self, root_dir=None):
 
     There are multiple examples of workflows that inject null objects into channels using `ifEmpty(null)`, which can cause unhandled null pointer exceptions.
     This lint test throws warnings for those instances.
-    """
-    passed = []
-    warned = []
-    file_paths = []
-    pattern = re.compile(r"ifEmpty\s*\(\s*null\s*\)")
 
+    The match logic lives in the ast-grep rule file ``rules/pipeline_if_empty_null.yml``.
+    See ``nf_core.astgrep.find_matches`` for the structural vs regex-fallback behaviour.
+    """
     # Pipelines don't provide a path, so use the workflow path.
     # Modules run this function twice and provide a string path
     if root_dir is None:
         root_dir = self.wf_path
 
-    for file in get_wf_files(root_dir):
-        try:
-            with open(Path(file), encoding="latin1") as fh:
-                for line in fh:
-                    if re.findall(pattern, line):
-                        warned.append(f"`ifEmpty(null)` found in `{file}`: _{line}_")
-                        file_paths.append(Path(file))
-        except FileNotFoundError:
-            log.debug(f"Could not open file {file} in pipeline_if_empty_null lint test")
-
-    if len(warned) == 0:
-        passed.append("No `ifEmpty(null)` strings found")
+    rule = astgrep.load_rule(RULE_FILE)
+    warned = []
+    file_paths = []
+    for file, line_num, text in astgrep.find_matches(rule, get_wf_files(root_dir)):
+        warned.append(f"{rule['message']} in `{file}` [line {line_num}]: _{text}_")
+        file_paths.append(file)
+
+    passed = [] if warned else ["No `ifEmpty(null)` strings found"]
 
     # return file_paths for use in subworkflow lint
     return {"passed": passed, "warned": warned, "file_paths": file_paths}

nf_core/pipelines/lint/pipeline_todos.py

@@ -3,8 +3,31 @@
 import os
 from pathlib import Path
 
+from nf_core import astgrep
+
 log = logging.getLogger(__name__)
 
+RULE_FILE = Path(__file__).parent / "rules" / "pipeline_todos.yml"
+
+
+def _clean_todo(text: str) -> str:
+    """Reduce a matched comment to the TODO text itself."""
+    for line in text.splitlines():
+        if "TODO nf-core" in line:
+            text = line
+            break
+    return (
+        text.replace("<!--", "")
+        .replace("-->", "")
+        .replace("/*", "")
+        .replace("*/", "")
+        .replace("*", "")
+        .replace("# TODO nf-core: ", "")
+        .replace("// TODO nf-core: ", "")
+        .replace("TODO nf-core: ", "")
+        .strip()
+    )
+
 
 def pipeline_todos(self, root_dir=None):
     """Check for nf-core *TODO* lines.
@@ -26,6 +49,10 @@ def pipeline_todos(self, root_dir=None):
     This lint test runs through all files in the pipeline and searches for these lines.
     If any are found they will throw a warning.
 
+    Nextflow/Groovy files are matched structurally via the ast-grep rule file
+    ``rules/pipeline_todos.yml`` when the parser is available (a "TODO nf-core"
+    inside a string is not flagged); all other files are searched line by line.
+
     .. tip:: Note that many GUI code editors have plugins to list all instances of *TODO*
               in a given project directory. This is a very quick and convenient way to get
               started on your pipeline!
@@ -45,33 +72,22 @@ def pipeline_todos(self, root_dir=None):
         with open(Path(root_dir, ".gitignore"), encoding="latin1") as fh:
             for line in fh:
                 ignore.append(Path(line.strip().rstrip("/")).name)
+
+    to_check = []
     for root, dirs, files in os.walk(root_dir, topdown=True):
         # Ignore files
         for i_base in ignore:
             i = str(Path(root, i_base))
             dirs[:] = [d for d in dirs if not fnmatch.fnmatch(str(Path(root, d)), i)]
             files[:] = [f for f in files if not fnmatch.fnmatch(str(Path(root, f)), i)]
-        for fname in files:
-            try:
-                with open(Path(root, fname), encoding="latin1") as fh:
-                    for line in fh:
-                        if "TODO nf-core" in line:
-                            line = (
-                                line.replace("<!--", "")
-                                .replace("-->", "")
-                                .replace("# TODO nf-core: ", "")
-                                .replace("// TODO nf-core: ", "")
-                                .replace("TODO nf-core: ", "")
-                                .strip()
-                            )
-                            warned.append(f"TODO string in `{fname}`: _{line}_")
-                            file_paths.append(Path(root, fname))
-            except FileNotFoundError:
-                log.debug(f"Could not open file {fname} in pipeline_todos lint test")
+        to_check.extend(Path(root, fname) for fname in files)
+
+    rule = astgrep.load_rule(RULE_FILE)
+    for file, _line_num, text in astgrep.find_matches(rule, to_check, regex_unparseable=True):
+        warned.append(f"TODO string in `{file.name}`: _{_clean_todo(text)}_")
+        file_paths.append(file)
 
     if len(warned) == 0:
         passed.append("No TODO strings found")
 
-    # HACK file paths are returned to allow usage of this function in modules/lint.py
-    # Needs to be refactored!
     return {"passed": passed, "warned": warned, "file_paths": file_paths}

nf_core/pipelines/lint/rules/pipeline_if_empty_null.yml

@@ -0,0 +1,28 @@
+# ast-grep lint rule (https://ast-grep.github.io/guide/project/lint-rule.html)
+# Also usable directly with `ast-grep scan` via a sgconfig.yml ruleDirs entry.
+id: pipeline_if_empty_null
+language: nextflow
+severity: warning
+message: "`ifEmpty(null)` found"
+note: |
+  Instead of `ifEmpty(null)`, use `ifEmpty([])` to keep a process executing on
+  optional input, or `ifEmpty { error ... }` when the channel must not be empty.
+  Injecting null objects into channels causes unhandled null pointer exceptions.
+metadata:
+  fallback-regex: ifEmpty\s*\(\s*null\s*\)
+rule:
+  any:
+    # method call form: ch.ifEmpty(null)
+    - pattern: $CH.ifEmpty(null)
+    # bare / piped form: ch | ifEmpty(null)
+    - pattern: ifEmpty(null)
+    # method calls whose receiver the grammar could not fully parse; only
+    # requires direct children `ifEmpty` and `null`
+    - all:
+        - kind: method_call
+        - has:
+            kind: identifier
+            regex: ^ifEmpty$
+        - has:
+            kind: simple_expression
+            regex: ^null$

nf_core/pipelines/lint/rules/pipeline_todos.yml

@@ -0,0 +1,16 @@
+# ast-grep lint rule (https://ast-grep.github.io/guide/project/lint-rule.html)
+id: pipeline_todos
+language: nextflow
+severity: warning
+message: "TODO string"
+note: |
+  The nf-core templates contain `TODO nf-core:` comments marking places that
+  developers need to edit. They should be removed once addressed.
+metadata:
+  fallback-regex: TODO nf-core
+# a comment node containing "TODO nf-core" (strings/code are not flagged)
+rule:
+  regex: TODO nf-core
+  any:
+    - kind: line_comment
+    - kind: block_comment

nf_core/pipelines/lint/rules/system_exit.yml

@@ -0,0 +1,16 @@
+# ast-grep lint rule (https://ast-grep.github.io/guide/project/lint-rule.html)
+id: system_exit
+language: nextflow
+severity: warning
+message: "`System.exit` call found"
+note: |
+  Calls to `System.exit(1)` should be replaced by throwing errors, e.g.
+  `error "Something went wrong"`. `System.exit(0)` is allowed.
+metadata:
+  fallback-regex: System\.exit\s*\((?!\s*0\s*\))
+rule:
+  pattern: System.exit($CODE)
+constraints:
+  CODE:
+    not:
+      regex: ^0$

nf_core/pipelines/lint/system_exit.py

@@ -1,8 +1,12 @@
 import logging
 from pathlib import Path
 
+from nf_core import astgrep
+
 log = logging.getLogger(__name__)
 
+RULE_FILE = Path(__file__).parent / "rules" / "system_exit.yml"
+
 
 def system_exit(self):
     """Check for System.exit calls in groovy/nextflow code
@@ -11,27 +15,18 @@ def system_exit(self):
 
     This lint test looks for all calls to `System.exit`
     in any file with the `.nf` or `.groovy` extension
-    """
-    passed = []
-    warned = []
 
+    The match logic lives in the ast-grep rule file ``rules/system_exit.yml``.
+    See ``nf_core.astgrep.find_matches`` for the structural vs regex-fallback behaviour.
+    """
     root_dir = Path(self.wf_path)
-
-    # Get all groovy and nf files
-    groovy_files = list(root_dir.rglob("*.groovy"))
-    nf_files = list(root_dir.rglob("*.nf"))
-    to_check = nf_files + groovy_files
-
-    for file in to_check:
-        try:
-            with file.open() as fh:
-                for i, line in enumerate(fh.readlines(), start=1):
-                    if "System.exit" in line and "System.exit(0)" not in line:
-                        warned.append(f"`System.exit` in {file.name}: _{line.strip()}_  [line {i}]")
-        except FileNotFoundError:
-            log.debug(f"Could not open file {file.name} in system_exit lint test")
-
-    if len(warned) == 0:
-        passed.append("No `System.exit` calls found")
+    files = list(root_dir.rglob("*.nf")) + list(root_dir.rglob("*.groovy"))
+
+    rule = astgrep.load_rule(RULE_FILE)
+    warned = [
+        f"{rule['message']} in {file.name}: _{text}_  [line {line_num}]"
+        for file, line_num, text in astgrep.find_matches(rule, files)
+    ]
+    passed = [] if warned else ["No `System.exit` calls found"]
 
     return {"passed": passed, "warned": warned}

tests/pipelines/lint/rules/pipeline_if_empty_null-test.yml

@@ -0,0 +1,21 @@
+# ast-grep rule test (https://ast-grep.github.io/guide/test-rule.html)
+# Run by tests/pipelines/lint/test_astgrep_rules.py: every `valid` snippet
+# must produce no matches, every `invalid` snippet at least one.
+id: pipeline_if_empty_null
+valid:
+  - ch.ifEmpty([])
+  - ch.ifEmpty { error "empty samplesheet" }
+  - |
+    workflow {
+        // ifEmpty(null) in a comment is fine
+        ch_ok = ch.toList()
+    }
+invalid:
+  - ch.ifEmpty(null)
+  - ch.ifEmpty( null )
+  - Channel.fromPath(params.input).ifEmpty(null)
+  - |
+    ch.ifEmpty(
+        null
+    )
+  - ch | ifEmpty(null)

tests/pipelines/lint/rules/pipeline_todos-test.yml

@@ -0,0 +1,19 @@
+# ast-grep rule test (https://ast-grep.github.io/guide/test-rule.html)
+id: pipeline_todos
+valid:
+  - "// a regular comment"
+  - 'println "TODO nf-core: inside a string is not a template TODO"'
+invalid:
+  - "// TODO nf-core: edit this section"
+  - |
+    /*
+     * TODO nf-core: a block comment TODO
+     */
+  - |
+    process FOO {
+        script:
+        // TODO nf-core: replace the command below
+        """
+        echo hi
+        """
+    }

tests/pipelines/lint/rules/system_exit-test.yml

@@ -0,0 +1,16 @@
+# ast-grep rule test (https://ast-grep.github.io/guide/test-rule.html)
+id: system_exit
+valid:
+  - System.exit(0)
+  - "// System.exit(1) in a comment is fine"
+  - error "Something went wrong"
+invalid:
+  - System.exit(1)
+  - System.exit(status)
+  - |
+    class Utils {
+        public static void fail(msg) {
+            log.error(msg)
+            System.exit(1)
+        }
+    }

tests/pipelines/lint/test_astgrep_rules.py

@@ -0,0 +1,48 @@
+"""Generic harness for ast-grep lint rule tests.
+
+Each rule in nf_core/pipelines/lint/rules/<id>.yml can ship a test file
+tests/pipelines/lint/rules/<id>-test.yml in ast-grep's test-rule format
+(https://ast-grep.github.io/guide/test-rule.html): `valid` snippets must not
+match the rule, `invalid` snippets must. Adding a new rule needs no new
+Python — just the rule file and a test file.
+"""
+
+from pathlib import Path
+
+import pytest
+import yaml
+
+import nf_core.pipelines.lint
+from nf_core import astgrep
+
+RULES_DIR = Path(nf_core.pipelines.lint.__file__).parent / "rules"
+TESTS_DIR = Path(__file__).parent / "rules"
+
+pytestmark = pytest.mark.skipif(not astgrep.nextflow_available(), reason="tree-sitter-nextflow parser not available")
+
+
+def rule_test_files():
+    return sorted(TESTS_DIR.glob("*-test.yml"))
+
+
+def test_every_rule_has_a_test_file():
+    tested = {f.name.removesuffix("-test.yml") for f in rule_test_files()}
+    rules = {f.stem for f in RULES_DIR.glob("*.yml")}
+    assert rules == tested, f"rules without tests: {rules - tested}; tests without rules: {tested - rules}"
+
+
+@pytest.mark.parametrize("test_file", rule_test_files(), ids=lambda p: p.name.removesuffix("-test.yml"))
+def test_rule(test_file):
+    with open(test_file) as fh:
+        spec = yaml.safe_load(fh)
+    rule_config = astgrep.load_rule(RULES_DIR / f"{spec['id']}.yml")
+
+    snippets = [(s, True) for s in spec.get("valid", [])] + [(s, False) for s in spec.get("invalid", [])]
+    for snippet, should_be_clean in snippets:
+        # a snippet with an unquoted `foo: bar` parses as a YAML map, not a string
+        assert isinstance(snippet, str), f"snippet in {test_file.name} is not a string (quote it?): {snippet!r}"
+        matches = astgrep.scan(snippet, rule_config)
+        if should_be_clean:
+            assert not matches, f"valid snippet matched {spec['id']}: {snippet!r}"
+        else:
+            assert matches, f"invalid snippet did not match {spec['id']}: {snippet!r}"