add pytest for nf-test linting

Author: mirpedrol

nf_core/pipelines/lint/nf_test_content.py

@@ -91,8 +91,10 @@ def nf_test_content(self) -> Dict[str, List[str]]:
     }
 
     for test_fn in test_fns:
-        if nf_test_content_conf is not None and (not nf_test_content_conf or str(test_fn) in nf_test_content_conf):
-            ignored.append(f"'{test_fn.name}' checking ignored")
+        if nf_test_content_conf is not None and (
+            not nf_test_content_conf or str(test_fn.relative_to(self.wf_path)) in nf_test_content_conf
+        ):
+            ignored.append(f"'{test_fn.relative_to(self.wf_path)}' checking ignored")
             continue
 
         checks_passed = {check: False for check in test_checks}
@@ -103,16 +105,18 @@ def nf_test_content(self) -> Dict[str, List[str]]:
                         while "}\n" not in line:
                             line = next(fh)
                             if re.search(str(check_info["pattern"]), line):
-                                passed.append(f"'{test_fn}' contains {check_info['description']}")
+                                passed.append(
+                                    f"'{test_fn.relative_to(self.wf_path)}' contains {check_info['description']}"
+                                )
                                 checks_passed[check_name] = True
                                 break
                     elif not check_info["when_block"] and re.search(str(check_info["pattern"]), line):
-                        passed.append(f"'{test_fn}' {check_info['description']}")
+                        passed.append(f"'{test_fn.relative_to(self.wf_path)}' {check_info['description']}")
                         checks_passed[check_name] = True
 
         for check_name, check_info in test_checks.items():
             if not checks_passed[check_name]:
-                failed.append(f"'{test_fn}' {check_info['failure_msg']}")
+                failed.append(f"'{test_fn.relative_to(self.wf_path)}' {check_info['failure_msg']}")
 
     # Content of nextflow.config file
     conf_fn = Path(self.wf_path, "tests", "nextflow.config")
@@ -127,38 +131,32 @@ def nf_test_content(self) -> Dict[str, List[str]]:
         },
         "cpus": {
             "pattern": "cpus: *[\"']?4[\"']?",
-            "description": "correct CPU resource limits",
-            "failure_msg": "correct CPU resource limits. Should be 4",
+            "description": "correct CPU resource limits. Should be 4",
         },
         "memory": {
             "pattern": "memory: *[\"']?15\.GB[\"']?",
-            "description": "correct memory resource limits",
-            "failure_msg": "correct memory resource limits. Should be 15.GB",
+            "description": "correct memory resource limits. Should be 15.GB",
         },
         "time": {
             "pattern": "time: *[\"']?1\.h[\"']?",
-            "description": "correct time resource limits",
-            "failure_msg": "correct time resource limits. Should be 1.h",
+            "description": "correct time resource limits. Should be 1.h",
         },
     }
 
-    if nf_test_content_conf is None or str(test_fn) not in nf_test_content_conf:
+    if nf_test_content_conf is None or str(conf_fn.relative_to(self.wf_path)) not in nf_test_content_conf:
         checks_passed = {check: False for check in config_checks}
         with open(conf_fn) as fh:
             for line in fh:
                 line = line.strip()
                 for check_name, config_check_info in config_checks.items():
                     if re.search(str(config_check_info["pattern"]), line):
-                        passed.append(f"'{conf_fn}' contains {config_check_info['description']}")
+                        passed.append(f"'{conf_fn.relative_to(self.wf_path)}' contains {config_check_info['description']}")
                         checks_passed[check_name] = True
         for check_name, config_check_info in config_checks.items():
             if not checks_passed[check_name]:
-                failure_msg = config_check_info.get(
-                    "failure_msg", f"does not contain {config_check_info['description']}"
-                )
-                failed.append(f"'{conf_fn}' {failure_msg}")
+                failed.append(f"'{conf_fn.relative_to(self.wf_path)}' does not contain {config_check_info['description']}")
     else:
-        ignored.append(f"'{conf_fn}' checking ignored")
+        ignored.append(f"'{conf_fn.relative_to(self.wf_path)}' checking ignored")
 
     # Content of nf-test.config file
     nf_test_conf_fn = Path(self.wf_path, "nf-test.config")
@@ -180,19 +178,19 @@ def nf_test_content(self) -> Dict[str, List[str]]:
         },
     }
 
-    if nf_test_content_conf is None or str(nf_test_conf_fn) not in nf_test_content_conf:
+    if nf_test_content_conf is None or str(nf_test_conf_fn.relative_to(self.wf_path)) not in nf_test_content_conf:
         checks_passed = {check: False for check in nf_test_checks}
         with open(nf_test_conf_fn) as fh:
             for line in fh:
                 line = line.strip()
                 for check_name, nf_test_check_info in nf_test_checks.items():
                     if re.search(str(nf_test_check_info["pattern"]), line):
-                        passed.append(f"'{nf_test_conf_fn}' {nf_test_check_info['description']}")
+                        passed.append(f"'{nf_test_conf_fn.relative_to(self.wf_path)}' {nf_test_check_info['description']}")
                         checks_passed[check_name] = True
         for check_name, nf_test_check_info in nf_test_checks.items():
             if not checks_passed[check_name]:
-                failed.append(f"'{nf_test_conf_fn}' {nf_test_check_info['failure_msg']}")
+                failed.append(f"'{nf_test_conf_fn.relative_to(self.wf_path)}' {nf_test_check_info['failure_msg']}")
     else:
-        ignored.append(f"'{nf_test_conf_fn}' checking ignored")
+        ignored.append(f"'{nf_test_conf_fn.relative_to(self.wf_path)}' checking ignored")
 
     return {"passed": passed, "failed": failed, "ignored": ignored}

tests/pipelines/lint/test_nf_test_content.py

@@ -0,0 +1,88 @@
+from pathlib import Path
+
+import yaml
+
+import nf_core.pipelines.lint
+
+from ..test_lint import TestLint
+
+
+class TestLintNfTestContent(TestLint):
+    def setUp(self) -> None:
+        super().setUp()
+        self.new_pipeline = self._make_pipeline_copy()
+
+    def test_nf_test_content_nf_test_files(self):
+        """Test failure if nf-test file does not contain outdir parameter."""
+        # Create a test file without outdir parameter or versions.yml snapshot
+        test_file = Path(self.new_pipeline) / "tests" / "test.nf.test"
+        with open(test_file, "w") as f:
+            f.write("// No outdir parameter or versions YAML snapshot")
+        lint_obj = nf_core.pipelines.lint.PipelineLint(self.new_pipeline)
+        result = lint_obj.nf_test_content()
+        assert len(result["failed"]) > 0
+        assert (
+            "'tests/test.nf.test' does not contain `outdir` parameter, it should contain `outdir = \"$outputDir\"`"
+            in result["failed"]
+        )
+        assert "'tests/test.nf.test' does not snapshot a 'versions.yml' file" in result["failed"]
+
+    def test_nf_test_content_nextflow_config_file(self):
+        """Test failure if nextflow.config does not contain correct resource limits."""
+        # Create nextflow.config without resource limits or required parameters
+        config_file = Path(self.new_pipeline) / "tests" / "nextflow.config"
+        with open(config_file, "w") as f:
+            f.write("// Missing resource limits and parameters")
+        lint_obj = nf_core.pipelines.lint.PipelineLint(self.new_pipeline)
+        result = lint_obj.nf_test_content()
+        assert len(result["failed"]) > 0
+        assert "'tests/nextflow.config' does not contain `modules_testdata_base_path`" in result["failed"]
+        assert "'tests/nextflow.config' does not contain `pipelines_testdata_base_path`" in result["failed"]
+        assert "'tests/nextflow.config' does not contain correct CPU resource limits. Should be 4" in result["failed"]
+        assert (
+            "'tests/nextflow.config' does not contain correct memory resource limits. Should be 15.GB"
+            in result["failed"]
+        )
+        assert (
+            "'tests/nextflow.config' does not contain correct time resource limits. Should be 1.h" in result["failed"]
+        )
+
+    def test_nf_test_content_missing_nf_test_config_file(self):
+        """Test failure if nf-test.config does not contain required settings."""
+        # Create nf-test.config without required settings
+        config_file = Path(self.new_pipeline) / "nf-test.config"
+        with open(config_file, "w") as f:
+            f.write("// Missing required settings")
+        lint_obj = nf_core.pipelines.lint.PipelineLint(self.new_pipeline)
+        result = lint_obj.nf_test_content()
+        assert len(result["failed"]) > 0
+        assert "'nf-test.config' does not set a `testsDir`, it should contain `testsDir \".\"`" in result["failed"]
+        assert (
+            '\'nf-test.config\' does not set a `workDir`, it should contain `workDir System.getenv("NFT_WORKDIR") ?: ".nf-test"`'
+            in result["failed"]
+        )
+        assert (
+            "'nf-test.config' does not set a `configFile`, it should contain `configFile \"tests/nextflow.config\"`"
+            in result["failed"]
+        )
+
+    def test_nf_test_content_ignored(self):
+        """Test that nf-test content checks can be ignored via .nf-core.yml."""
+        # Create .nf-core.yml to ignore checks
+        nf_core_yml = Path(self.new_pipeline) / ".nf-core.yml"
+        with open(nf_core_yml) as f:
+            yml_content = yaml.safe_load(f)
+
+        yml_content["lint"] = {"nf_test_content": ["tests/default.nf.test", "tests/nextflow.config", "nf-test.config"]}
+
+        with open(nf_core_yml, "w") as f:
+            yaml.dump(yml_content, f)
+
+        lint_obj = nf_core.pipelines.lint.PipelineLint(self.new_pipeline)
+        lint_obj._load()
+        result = lint_obj.nf_test_content()
+        print(result)
+        assert len(result["ignored"]) == 3
+        assert "'tests/default.nf.test' checking ignored" in result["ignored"]
+        assert "'tests/nextflow.config' checking ignored" in result["ignored"]
+        assert "'nf-test.config' checking ignored" in result["ignored"]