split subworkflow/test_lint into separate files

CHANGELOG.md

@@ -35,6 +35,7 @@
 ### Subworkflows
 
 - Update to new topic version handling ([#3929](https://github.com/nf-core/tools/pull/3929))
+- split subworkflow/test_lint into separate files ([#3965](https://github.com/nf-core/tools/pull/3965))
 
 ## [v3.5.1 - Terne Topi](https://github.com/nf-core/tools/releases/tag/3.5.1) - [2025-11-19]
 

tests/subworkflows/lint/test_integration.py

@@ -0,0 +1,69 @@
+import nf_core.subworkflows
+
+from ...test_subworkflows import TestSubworkflows
+from ...utils import GITLAB_SUBWORKFLOWS_BRANCH, GITLAB_URL
+
+
+class TestSubworkflowsLintIntegration(TestSubworkflows):
+    """Test general subworkflow linting functionality"""
+
+    def test_subworkflows_lint(self):
+        """Test linting the fastq_align_bowtie2 subworkflow"""
+        self.subworkflow_install.install("fastq_align_bowtie2")
+        subworkflow_lint = nf_core.subworkflows.SubworkflowLint(directory=self.pipeline_dir)
+        subworkflow_lint.lint(print_results=False, subworkflow="fastq_align_bowtie2")
+        assert len(subworkflow_lint.failed) == 0, f"Linting failed with {[x.__dict__ for x in subworkflow_lint.failed]}"
+        assert len(subworkflow_lint.passed) > 0
+        assert len(subworkflow_lint.warned) >= 0
+
+    def test_subworkflows_lint_empty(self):
+        """Test linting a pipeline with no subworkflows installed"""
+        self.subworkflow_remove.remove("utils_nextflow_pipeline", force=True)
+        self.subworkflow_remove.remove("utils_nfcore_pipeline", force=True)
+        self.subworkflow_remove.remove("utils_nfschema_plugin", force=True)
+        nf_core.subworkflows.SubworkflowLint(directory=self.pipeline_dir)
+        assert "No subworkflows from https://github.com/nf-core/modules.git installed in pipeline" in self.caplog.text
+
+    def test_subworkflows_lint_new_subworkflow(self):
+        """lint a new subworkflow"""
+        subworkflow_lint = nf_core.subworkflows.SubworkflowLint(directory=self.nfcore_modules)
+        subworkflow_lint.lint(print_results=True, all_subworkflows=True)
+        assert len(subworkflow_lint.failed) == 0
+        assert len(subworkflow_lint.passed) > 0
+        assert len(subworkflow_lint.warned) >= 0
+
+    def test_subworkflows_lint_no_gitlab(self):
+        """Test linting a pipeline with no subworkflows installed"""
+        nf_core.subworkflows.SubworkflowLint(directory=self.pipeline_dir, remote_url=GITLAB_URL)
+        assert f"No subworkflows from {GITLAB_URL} installed in pipeline" in self.caplog.text
+
+    def test_subworkflows_lint_gitlab_subworkflows(self):
+        """Lint subworkflows from a different remote"""
+        self.subworkflow_install_gitlab.install("bam_stats_samtools")
+        subworkflow_lint = nf_core.subworkflows.SubworkflowLint(
+            directory=self.pipeline_dir, remote_url=GITLAB_URL, branch=GITLAB_SUBWORKFLOWS_BRANCH
+        )
+        subworkflow_lint.lint(print_results=False, all_subworkflows=True)
+        assert len(subworkflow_lint.failed) == 0
+        assert len(subworkflow_lint.passed) > 0
+        assert len(subworkflow_lint.warned) >= 0
+
+    def test_subworkflows_lint_multiple_remotes(self):
+        """Lint subworkflows from a different remote"""
+        self.subworkflow_install_gitlab.install("bam_stats_samtools")
+        self.subworkflow_install.install("fastq_align_bowtie2")
+        subworkflow_lint = nf_core.subworkflows.SubworkflowLint(
+            directory=self.pipeline_dir, remote_url=GITLAB_URL, branch=GITLAB_SUBWORKFLOWS_BRANCH
+        )
+        subworkflow_lint.lint(print_results=False, all_subworkflows=True)
+        assert len(subworkflow_lint.failed) == 0
+        assert len(subworkflow_lint.passed) > 0
+        assert len(subworkflow_lint.warned) >= 0
+
+    def test_subworkflows_lint_fix(self):
+        """update the meta.yml of a subworkflow"""
+        subworkflow_lint = nf_core.subworkflows.SubworkflowLint(directory=self.nfcore_modules, fix=True)
+        subworkflow_lint.lint(print_results=False, subworkflow="test_subworkflow")
+        assert len(subworkflow_lint.failed) == 0, f"Linting failed with {[x.__dict__ for x in subworkflow_lint.failed]}"
+        assert len(subworkflow_lint.passed) > 0
+        assert len(subworkflow_lint.warned) >= 0

tests/subworkflows/lint/test_local.py

@@ -0,0 +1,46 @@
+import shutil
+from pathlib import Path
+
+import nf_core.subworkflows
+
+from ...test_subworkflows import TestSubworkflows
+
+
+class TestSubworkflowsLintLocal(TestSubworkflows):
+    """Test linting local subworkflows"""
+
+    def setUp(self) -> None:
+        super().setUp()
+        assert self.subworkflow_install.install("fastq_align_bowtie2")
+        self.install_path = Path(self.pipeline_dir, "subworkflows", "nf-core", "fastq_align_bowtie2")
+        self.local_path = Path(self.pipeline_dir, "subworkflows", "local", "fastq_align_bowtie2")
+
+    def test_subworkflows_lint_local(self):
+        shutil.move(self.install_path, self.local_path)
+        subworkflow_lint = nf_core.subworkflows.SubworkflowLint(directory=self.pipeline_dir)
+        subworkflow_lint.lint(print_results=False, local=True)
+        assert len(subworkflow_lint.failed) == 0, f"Linting failed with {[x.__dict__ for x in subworkflow_lint.failed]}"
+        assert len(subworkflow_lint.passed) > 0
+        assert len(subworkflow_lint.warned) >= 0
+
+    def test_subworkflows_lint_local_missing_files(self):
+        shutil.move(self.install_path, self.local_path)
+        Path(self.local_path, "meta.yml").unlink()
+        subworkflow_lint = nf_core.subworkflows.SubworkflowLint(directory=self.pipeline_dir)
+        subworkflow_lint.lint(print_results=False, local=True)
+        assert len(subworkflow_lint.failed) == 0, f"Linting failed with {[x.__dict__ for x in subworkflow_lint.failed]}"
+        assert len(subworkflow_lint.passed) > 0
+        assert len(subworkflow_lint.warned) >= 0
+        warnings = [x.message for x in subworkflow_lint.warned]
+        assert "Subworkflow `meta.yml` does not exist" in warnings
+
+    def test_subworkflows_lint_local_old_format(self):
+        Path(self.pipeline_dir, "subworkflows", "local").mkdir(exist_ok=True)
+        local = Path(self.pipeline_dir, "subworkflows", "local", "fastq_align_bowtie2.nf")
+        shutil.copy(Path(self.install_path, "main.nf"), local)
+        self.subworkflow_remove.remove("fastq_align_bowtie2", force=True)
+        subworkflow_lint = nf_core.subworkflows.SubworkflowLint(directory=self.pipeline_dir)
+        subworkflow_lint.lint(print_results=False, local=True)
+        assert len(subworkflow_lint.failed) == 0, f"Linting failed with {[x.__dict__ for x in subworkflow_lint.failed]}"
+        assert len(subworkflow_lint.passed) > 0
+        assert len(subworkflow_lint.warned) >= 0

tests/subworkflows/lint/test_main_nf.py

@@ -0,0 +1,103 @@
+from pathlib import Path
+
+import nf_core.subworkflows
+
+from ...test_subworkflows import TestSubworkflows
+
+
+class TestMainNf(TestSubworkflows):
+    """Test main.nf functionality in subworkflows"""
+
+    def setUp(self) -> None:
+        super().setUp()
+
+        self.subworkflow_install.install("bam_stats_samtools")
+        self.main_nf = Path(
+            self.pipeline_dir,
+            "subworkflows",
+            "nf-core",
+            "bam_stats_samtools",
+            "main.nf",
+        )
+
+    def test_subworkflows_lint_less_than_two_modules_warning(self):
+        """Test linting a subworkflow with less than two modules"""
+        # Remove two modules
+        with open(self.main_nf) as fh:
+            content = fh.read()
+            new_content = content.replace(
+                "include { SAMTOOLS_IDXSTATS } from '../../../modules/nf-core/samtools/idxstats/main'",
+                "",
+            )
+            new_content = new_content.replace(
+                "include { SAMTOOLS_FLAGSTAT } from '../../../modules/nf-core/samtools/flagstat/main'",
+                "",
+            )
+        with open(
+            self.main_nf,
+            "w",
+        ) as fh:
+            fh.write(new_content)
+        subworkflow_lint = nf_core.subworkflows.SubworkflowLint(directory=self.pipeline_dir)
+        subworkflow_lint.lint(print_results=False, subworkflow="bam_stats_samtools")
+        assert len(subworkflow_lint.failed) >= 0, f"Linting failed with {[x.__dict__ for x in subworkflow_lint.failed]}"
+        assert len(subworkflow_lint.passed) > 0
+        assert len(subworkflow_lint.warned) > 0
+        assert subworkflow_lint.warned[0].lint_test == "main_nf_include"
+
+    def test_subworkflows_lint_include_multiple_alias(self):
+        """Test linting a subworkflow with multiple include methods"""
+        with open(self.main_nf) as fh:
+            content = fh.read()
+            new_content = content.replace("SAMTOOLS_STATS", "SAMTOOLS_STATS_1")
+            new_content = new_content.replace(
+                "include { SAMTOOLS_STATS_1 ",
+                "include { SAMTOOLS_STATS as SAMTOOLS_STATS_1; SAMTOOLS_STATS as SAMTOOLS_STATS_2 ",
+            )
+        with open(
+            self.main_nf,
+            "w",
+        ) as fh:
+            fh.write(new_content)
+
+        subworkflow_lint = nf_core.subworkflows.SubworkflowLint(directory=self.pipeline_dir)
+        subworkflow_lint.lint(print_results=False, subworkflow="bam_stats_samtools")
+        assert len(subworkflow_lint.failed) >= 0, f"Linting failed with {[x.__dict__ for x in subworkflow_lint.failed]}"
+        assert len(subworkflow_lint.passed) > 0
+        assert len(subworkflow_lint.warned) == 3
+        assert any(
+            [x.message == "Included component 'SAMTOOLS_STATS_1' used in main.nf" for x in subworkflow_lint.passed]
+        )
+        assert any(
+            [x.message == "Included component 'SAMTOOLS_STATS_2' not used in main.nf" for x in subworkflow_lint.warned]
+        )
+        assert any(
+            [
+                x.message.endswith("Can be ignored if the module is using topic channels")
+                for x in subworkflow_lint.warned
+            ]
+        )
+
+        # cleanup
+        self.subworkflow_remove.remove("bam_stats_samtools", force=True)
+
+    def test_subworkflows_lint_capitalization_fail(self):
+        """Test linting a subworkflow with a capitalization fail"""
+        # change workflow name to lowercase
+        with open(self.main_nf) as fh:
+            content = fh.read()
+            new_content = content.replace("workflow BAM_STATS_SAMTOOLS {", "workflow bam_stats_samtools {")
+        with open(
+            self.main_nf,
+            "w",
+        ) as fh:
+            fh.write(new_content)
+        subworkflow_lint = nf_core.subworkflows.SubworkflowLint(directory=self.pipeline_dir)
+        subworkflow_lint.lint(print_results=False, subworkflow="bam_stats_samtools")
+        assert len(subworkflow_lint.failed) >= 1, f"Linting failed with {[x.__dict__ for x in subworkflow_lint.failed]}"
+        assert len(subworkflow_lint.passed) > 0
+        assert len(subworkflow_lint.warned) >= 0
+        assert any([x.lint_test == "workflow_capitals" for x in subworkflow_lint.failed])
+
+        # cleanup
+        self.subworkflow_remove.remove("bam_stats_samtools", force=True)

tests/subworkflows/lint/test_nf_test.py

@@ -0,0 +1,142 @@
+import json
+import shutil
+from pathlib import Path
+
+import nf_core.subworkflows
+
+from ...test_subworkflows import TestSubworkflows
+
+
+class TestSubworkflowsNfTest(TestSubworkflows):
+    """Test subworkflow nf-test and snapshot functionality"""
+
+    def setUp(self):
+        super().setUp()
+        self.snap_file = Path(
+            self.nfcore_modules,
+            "subworkflows",
+            "nf-core",
+            "test_subworkflow",
+            "tests",
+            "main.nf.test.snap",
+        )
+
+    def test_subworkflows_lint_snapshot_file(self):
+        """Test linting a subworkflow with a snapshot file"""
+        subworkflow_lint = nf_core.subworkflows.SubworkflowLint(directory=self.nfcore_modules)
+        subworkflow_lint.lint(print_results=False, subworkflow="test_subworkflow")
+        assert len(subworkflow_lint.failed) == 0, f"Linting failed with {[x.__dict__ for x in subworkflow_lint.failed]}"
+        assert len(subworkflow_lint.passed) > 0
+        assert len(subworkflow_lint.warned) >= 0
+
+    def test_subworkflows_lint_snapshot_file_missing_fail(self):
+        """Test linting a subworkflow with a snapshot file missing, which should fail"""
+        self.snap_file.unlink()
+        subworkflow_lint = nf_core.subworkflows.SubworkflowLint(directory=self.nfcore_modules)
+        subworkflow_lint.lint(print_results=False, subworkflow="test_subworkflow")
+        self.snap_file.touch()
+        assert len(subworkflow_lint.failed) == 1, f"Linting failed with {[x.__dict__ for x in subworkflow_lint.failed]}"
+        assert len(subworkflow_lint.passed) > 0
+        assert len(subworkflow_lint.warned) >= 0
+
+    def test_subworkflows_lint_snapshot_file_not_needed(self):
+        """Test linting a subworkflow which doesn't need a snapshot file by removing the snapshot keyword in the main.nf.test file"""
+        with open(
+            Path(
+                self.nfcore_modules,
+                "subworkflows",
+                "nf-core",
+                "test_subworkflow",
+                "tests",
+                "main.nf.test",
+            )
+        ) as fh:
+            content = fh.read()
+            new_content = content.replace("snapshot(", "snap (")
+        with open(
+            Path(
+                self.nfcore_modules,
+                "subworkflows",
+                "nf-core",
+                "test_subworkflow",
+                "tests",
+                "main.nf.test",
+            ),
+            "w",
+        ) as fh:
+            fh.write(new_content)
+
+        self.snap_file.unlink()
+        subworkflow_lint = nf_core.subworkflows.SubworkflowLint(directory=self.nfcore_modules)
+        subworkflow_lint.lint(print_results=False, subworkflow="test_subworkflow")
+        Path(
+            self.nfcore_modules,
+            "subworkflows",
+            "nf-core",
+            "test_subworkflow",
+            "tests",
+            "main.nf.test.snap",
+        ).touch()
+        assert len(subworkflow_lint.failed) == 0, f"Linting failed with {[x.__dict__ for x in subworkflow_lint.failed]}"
+        assert len(subworkflow_lint.passed) > 0
+        assert len(subworkflow_lint.warned) >= 0
+
+    def test_subworkflows_absent_version(self):
+        """Test linting a nf-test subworkflow if the versions is absent in the snapshot file `"""
+
+        with open(self.snap_file) as fh:
+            content = fh.read()
+            new_content = content.replace("versions", "foo")
+        with open(self.snap_file, "w") as fh:
+            fh.write(new_content)
+
+        subworkflow_lint = nf_core.subworkflows.SubworkflowLint(directory=self.nfcore_modules)
+        subworkflow_lint.lint(print_results=False, subworkflow="test_subworkflow")
+        assert len(subworkflow_lint.failed) == 0
+        assert len(subworkflow_lint.passed) > 0
+        assert len(subworkflow_lint.warned) >= 0, f"Linting warned with {[x.__dict__ for x in subworkflow_lint.warned]}"
+        assert any([x.lint_test == "test_snap_versions" for x in subworkflow_lint.warned])
+
+    def test_subworkflows_empty_file_in_snapshot(self):
+        """Test linting a nf-test subworkflow with an empty file sha sum in the test snapshot, which should make it fail (if it is not a stub)"""
+
+        snap = json.load(self.snap_file.open())
+        snap["my test"]["content"][0]["0"] = "test:md5,d41d8cd98f00b204e9800998ecf8427e"
+
+        with open(self.snap_file, "w") as fh:
+            json.dump(snap, fh)
+
+        subworkflow_lint = nf_core.subworkflows.SubworkflowLint(directory=self.nfcore_modules)
+        subworkflow_lint.lint(print_results=False, subworkflow="test_subworkflow")
+        assert len(subworkflow_lint.failed) == 1, f"Linting failed with {[x.__dict__ for x in subworkflow_lint.failed]}"
+        assert len(subworkflow_lint.passed) > 0
+        assert len(subworkflow_lint.warned) >= 0
+        assert subworkflow_lint.failed[0].lint_test == "test_snap_md5sum"
+
+    def test_subworkflows_empty_file_in_stub_snapshot(self):
+        """Test linting a nf-test subworkflow with an empty file sha sum in the stub test snapshot, which should make it not fail"""
+
+        content = json.load(self.snap_file.open())
+        content["my_test_stub"] = {"content": [{"0": "test:md5,d41d8cd98f00b204e9800998ecf8427e", "versions": {}}]}
+
+        with open(self.snap_file, "w") as fh:
+            json.dump(content, fh)
+
+        subworkflow_lint = nf_core.subworkflows.SubworkflowLint(directory=self.nfcore_modules)
+        subworkflow_lint.lint(print_results=False, subworkflow="test_subworkflow")
+        assert len(subworkflow_lint.failed) == 0, f"Linting failed with {[x.__dict__ for x in subworkflow_lint.failed]}"
+        assert len(subworkflow_lint.passed) > 0
+        assert len(subworkflow_lint.warned) >= 0
+        assert any(x.lint_test == "test_snap_md5sum" for x in subworkflow_lint.passed)
+
+    def test_subworkflows_missing_test_dir(self):
+        """Test linting a nf-test subworkflow if the tests directory is missing"""
+        test_dir = self.snap_file.parent
+        shutil.rmtree(test_dir)
+
+        subworkflow_lint = nf_core.subworkflows.SubworkflowLint(self.nfcore_modules)
+        subworkflow_lint.lint(print_results=False, subworkflow="test_subworkflow")
+        assert len(subworkflow_lint.failed) == 1
+        assert len(subworkflow_lint.passed) > 0
+        assert len(subworkflow_lint.warned) >= 0, f"Linting warned with {[x.__dict__ for x in subworkflow_lint.warned]}"
+        assert any([x.lint_test == "test_dir_exists" for x in subworkflow_lint.failed])