Add version check in snapshot file

CHANGELOG.md

@@ -16,6 +16,7 @@
 ### Linting
 
 - Only match assignments of params in `main.nf` and not references like `params.aligner == <something>` ([#2833](https://github.com/nf-core/tools/pull/2833))
+- Include test for presence of versions in snapshot ([#2888](https://github.com/nf-core/tools/pull/2888))
 
 ### Download
 

nf_core/modules/lint/module_tests.py

@@ -107,6 +107,22 @@ def module_tests(_, module: NFCoreComponent):
                                             snap_file,
                                         )
                                     )
+                            if "versions" in str(snap_content[test_name]) or "versions" in str(snap_content.keys()):
+                                module.passed.append(
+                                    (
+                                        "test_snap_versions",
+                                        "versions found in snapshot file",
+                                        snap_file,
+                                    )
+                                )
+                            else:
+                                module.failed.append(
+                                    (
+                                        "test_snap_versions",
+                                        "versions not found in snapshot file",
+                                        snap_file,
+                                    )
+                                )
                         except json.decoder.JSONDecodeError as e:
                             module.failed.append(
                                 (

nf_core/subworkflows/lint/subworkflow_tests.py

@@ -114,6 +114,22 @@ def subworkflow_tests(_, subworkflow: NFCoreComponent):
                                             snap_file,
                                         )
                                     )
+                            if "versions" in str(snap_content[test_name]) or "versions" in str(snap_content.keys()):
+                                subworkflow.passed.append(
+                                    (
+                                        "test_snap_versions",
+                                        "versions found in snapshot file",
+                                        snap_file,
+                                    )
+                                )
+                            else:
+                                subworkflow.warned.append(
+                                    (
+                                        "test_snap_versions",
+                                        "versions not found in snapshot file",
+                                        snap_file,
+                                    )
+                                )
                         except json.decoder.JSONDecodeError as e:
                             subworkflow.failed.append(
                                 (

tests/components/generate_snapshot.py

@@ -120,9 +120,12 @@ def test_test_not_found(self):
             remote_url=GITLAB_URL,
             branch=GITLAB_NFTEST_BRANCH,
         )
+        test_file = Path("modules", "nf-core-test", "fastp", "tests", "main.nf.test")
+        test_file.rename(test_file.parent / "main.nf.test.bak")
         with pytest.raises(UserWarning) as e:
             snap_generator.run()
         assert "Test file 'main.nf.test' not found" in str(e.value)
+        Path(test_file.parent / "main.nf.test.bak").rename(test_file)
 
 
 def test_unstable_snapshot(self):

tests/modules/lint.py

@@ -1,3 +1,4 @@
+import json
 from pathlib import Path
 
 import pytest
@@ -563,30 +564,29 @@ def test_modules_missing_required_tag(self):
 
 def test_modules_missing_tags_yml(self):
     """Test linting a module with a missing tags.yml file"""
-    Path(self.nfcore_modules, "modules", "nf-core", "bpipe", "test", "tests", "tags.yml").rename(
-        Path(self.nfcore_modules, "modules", "nf-core", "bpipe", "test", "tests", "tags.yml.bak")
-    )
+    tags_path = Path(self.nfcore_modules, "modules", "nf-core", "bpipe", "test", "tests", "tags.yml")
+    tags_path.rename(tags_path.parent / "tags.yml.bak")
     module_lint = nf_core.modules.ModuleLint(dir=self.nfcore_modules)
     module_lint.lint(print_results=False, module="bpipe/test")
-    Path(self.nfcore_modules, "modules", "nf-core", "bpipe", "test", "tests", "tags.yml.bak").rename(
-        Path(self.nfcore_modules, "modules", "nf-core", "bpipe", "test", "tests", "tags.yml")
-    )
     assert len(module_lint.failed) == 1, f"Linting failed with {[x.__dict__ for x in module_lint.failed]}"
     assert len(module_lint.passed) >= 0
     assert len(module_lint.warned) >= 0
     assert module_lint.failed[0].lint_test == "test_tags_yml_exists"
+    # cleanup
+    Path(tags_path.parent / "tags.yml.bak").rename(tags_path.parent / "tags.yml")
 
 
 def test_modules_incorrect_tags_yml_key(self):
     """Test linting a module with an incorrect key in tags.yml file"""
-    with open(Path(self.nfcore_modules, "modules", "nf-core", "bpipe", "test", "tests", "tags.yml")) as fh:
+    tags_path = Path(self.nfcore_modules, "modules", "nf-core", "bpipe", "test", "tests", "tags.yml")
+    with open(tags_path) as fh:
         content = fh.read()
         new_content = content.replace("bpipe/test:", "bpipe_test:")
-    with open(Path(self.nfcore_modules, "modules", "nf-core", "bpipe", "test", "tests", "tags.yml"), "w") as fh:
+    with open(tags_path, "w") as fh:
         fh.write(new_content)
     module_lint = nf_core.modules.ModuleLint(dir=self.nfcore_modules)
     module_lint.lint(print_results=True, module="bpipe/test")
-    with open(Path(self.nfcore_modules, "modules", "nf-core", "bpipe", "test", "tests", "tags.yml"), "w") as fh:
+    with open(tags_path, "w") as fh:
         fh.write(content)
     assert len(module_lint.failed) == 1, f"Linting failed with {[x.__dict__ for x in module_lint.failed]}"
     assert len(module_lint.passed) >= 0
@@ -596,14 +596,15 @@ def test_modules_incorrect_tags_yml_key(self):
 
 def test_modules_incorrect_tags_yml_values(self):
     """Test linting a module with an incorrect path in tags.yml file"""
-    with open(Path(self.nfcore_modules, "modules", "nf-core", "bpipe", "test", "tests", "tags.yml")) as fh:
+    tags_path = Path(self.nfcore_modules, "modules", "nf-core", "bpipe", "test", "tests", "tags.yml")
+    with open(tags_path) as fh:
         content = fh.read()
         new_content = content.replace("modules/nf-core/bpipe/test/**", "foo")
-    with open(Path(self.nfcore_modules, "modules", "nf-core", "bpipe", "test", "tests", "tags.yml"), "w") as fh:
+    with open(tags_path, "w") as fh:
         fh.write(new_content)
     module_lint = nf_core.modules.ModuleLint(dir=self.nfcore_modules)
     module_lint.lint(print_results=False, module="bpipe/test")
-    with open(Path(self.nfcore_modules, "modules", "nf-core", "bpipe", "test", "tests", "tags.yml"), "w") as fh:
+    with open(tags_path, "w") as fh:
         fh.write(content)
     assert len(module_lint.failed) == 1, f"Linting failed with {[x.__dict__ for x in module_lint.failed]}"
     assert len(module_lint.passed) >= 0
@@ -642,3 +643,68 @@ def test_nftest_failing_linting(self):
     assert module_lint.failed[2].lint_test == "test_main_tags"
     assert "kallisto/index" in module_lint.failed[2].message
     assert module_lint.failed[3].lint_test == "test_tags_yml"
+
+
+def test_modules_absent_version(self):
+    """Test linting a nf-test module if the versions is absent in the snapshot file `"""
+    with open(Path(self.nfcore_modules, "modules", "nf-core", "bpipe", "test", "tests", "main.nf.test.snap")) as fh:
+        content = fh.read()
+        new_content = content.replace("versions", "foo")
+    with open(
+        Path(self.nfcore_modules, "modules", "nf-core", "bpipe", "test", "tests", "main.nf.test.snap"), "w"
+    ) as fh:
+        fh.write(new_content)
+    module_lint = nf_core.modules.ModuleLint(dir=self.nfcore_modules)
+    module_lint.lint(print_results=False, module="bpipe/test")
+    with open(
+        Path(self.nfcore_modules, "modules", "nf-core", "bpipe", "test", "tests", "main.nf.test.snap"), "w"
+    ) as fh:
+        fh.write(content)
+    assert len(module_lint.failed) == 1, f"Linting failed with {[x.__dict__ for x in module_lint.failed]}"
+    assert len(module_lint.passed) >= 0
+    assert len(module_lint.warned) >= 0
+    assert module_lint.failed[0].lint_test == "test_snap_versions"
+
+
+def test_modules_empty_file_in_snapshot(self):
+    """Test linting a nf-test module with an empty file sha sum in the test snapshot, which should make it fail (if it is not a stub)"""
+    snap_file = Path(self.nfcore_modules, "modules", "nf-core", "bpipe", "test", "tests", "main.nf.test.snap")
+    snap = json.load(snap_file.open())
+    content = snap_file.read_text()
+    snap["my test"]["content"][0]["0"] = "test:md5,d41d8cd98f00b204e9800998ecf8427e"
+
+    with open(snap_file, "w") as fh:
+        json.dump(snap, fh)
+
+    module_lint = nf_core.modules.ModuleLint(dir=self.nfcore_modules)
+    module_lint.lint(print_results=False, module="bpipe/test")
+    assert len(module_lint.failed) == 1, f"Linting failed with {[x.__dict__ for x in module_lint.failed]}"
+    assert len(module_lint.passed) > 0
+    assert len(module_lint.warned) >= 0
+    assert module_lint.failed[0].lint_test == "test_snap_md5sum"
+
+    # reset the file
+    with open(snap_file, "w") as fh:
+        fh.write(content)
+
+
+def test_modules_empty_file_in_stub_snapshot(self):
+    """Test linting a nf-test module with an empty file sha sum in the stub test snapshot, which should make it not fail"""
+    snap_file = Path(self.nfcore_modules, "modules", "nf-core", "bpipe", "test", "tests", "main.nf.test.snap")
+    snap = json.load(snap_file.open())
+    content = snap_file.read_text()
+    snap["my_test_stub"] = {"content": [{"0": "test:md5,d41d8cd98f00b204e9800998ecf8427e", "versions": {}}]}
+
+    with open(snap_file, "w") as fh:
+        json.dump(snap, fh)
+
+    module_lint = nf_core.modules.ModuleLint(dir=self.nfcore_modules)
+    module_lint.lint(print_results=False, module="bpipe/test")
+    assert len(module_lint.failed) == 0, f"Linting failed with {[x.__dict__ for x in module_lint.failed]}"
+    assert len(module_lint.passed) > 0
+    assert len(module_lint.warned) >= 0
+    assert any(x.lint_test == "test_snap_md5sum" for x in module_lint.passed)
+
+    # reset the file
+    with open(snap_file, "w") as fh:
+        fh.write(content)

tests/subworkflows/lint.py

@@ -1,3 +1,5 @@
+import json
+import shutil
 from pathlib import Path
 
 import pytest
@@ -181,3 +183,160 @@ def test_subworkflows_lint_capitalization_fail(self):
 
     # cleanup
     self.subworkflow_remove.remove("bam_stats_samtools", force=True)
+
+
+def test_subworkflows_absent_version(self):
+    """Test linting a nf-test module if the versions is absent in the snapshot file `"""
+    snap_file = Path(self.nfcore_modules, "subworkflows", "nf-core", "test_subworkflow", "tests", "main.nf.test.snap")
+    with open(snap_file) as fh:
+        content = fh.read()
+        new_content = content.replace("versions", "foo")
+    with open(snap_file, "w") as fh:
+        fh.write(new_content)
+
+    subworkflow_lint = nf_core.subworkflows.SubworkflowLint(dir=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])
+
+    # cleanup
+    with open(snap_file, "w") as fh:
+        fh.write(content)
+
+
+def test_subworkflows_missing_test_dir(self):
+    """Test linting a nf-test subworkflow if the tests directory is missing"""
+    test_dir = Path(self.nfcore_modules, "subworkflows", "nf-core", "test_subworkflow", "tests")
+    test_dir_copy = shutil.copytree(test_dir, test_dir.parent / "tests_copy")
+    shutil.rmtree(test_dir)
+
+    subworkflow_lint = nf_core.subworkflows.SubworkflowLint(dir=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_dir_versions" for x in subworkflow_lint.warned])
+
+    # cleanup
+    shutil.copytree(test_dir_copy, test_dir)
+
+
+def test_subworkflows_missing_main_nf(self):
+    """Test linting a nf-test subworkflow if the main.nf file is missing"""
+    main_nf = Path(self.nfcore_modules, "subworkflows", "nf-core", "test_subworkflow", "main.nf")
+    main_nf_copy = shutil.copy(main_nf, main_nf.parent / "main_nf_copy")
+    main_nf.unlink()
+
+    subworkflow_lint = nf_core.subworkflows.SubworkflowLint(dir=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 == "main_nf_exists"
+
+    # cleanup
+    shutil.copy(main_nf_copy, main_nf)
+
+
+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_file = Path(self.nfcore_modules, "subworkflows", "nf-core", "test_subworkflow", "tests", "main.nf.test.snap")
+    snap = json.load(snap_file.open())
+    content = snap_file.read_text()
+    snap["my test"]["content"][0]["0"] = "test:md5,d41d8cd98f00b204e9800998ecf8427e"
+
+    with open(snap_file, "w") as fh:
+        json.dump(snap, fh)
+
+    subworkflow_lint = nf_core.subworkflows.SubworkflowLint(dir=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"
+
+    # reset the file
+    with open(snap_file, "w") as fh:
+        fh.write(content)
+
+
+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"""
+    snap_file = Path(self.nfcore_modules, "subworkflows", "nf-core", "test_subworkflow", "tests", "main.nf.test.snap")
+    snap = json.load(snap_file.open())
+    content = snap_file.read_text()
+    snap["my_test_stub"] = {"content": [{"0": "test:md5,d41d8cd98f00b204e9800998ecf8427e", "versions": {}}]}
+
+    with open(snap_file, "w") as fh:
+        json.dump(snap, fh)
+
+    subworkflow_lint = nf_core.subworkflows.SubworkflowLint(dir=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)
+
+    # reset the file
+    with open(snap_file, "w") as fh:
+        fh.write(content)
+
+
+def test_subworkflows_missing_tags_yml(self):
+    """Test linting a subworkflow with a missing tags.yml file"""
+    tags_path = Path(self.nfcore_modules, "subworkflows", "nf-core", "test_subworkflow", "tests", "tags.yml")
+    tags_path.rename(tags_path.parent / "tags.yml.bak")
+    subworkflow_lint = nf_core.subworkflows.SubworkflowLint(dir=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_tags_yml_exists"
+
+    # cleanup
+    Path(tags_path.parent / "tags.yml.bak").rename(tags_path.parent / "tags.yml")
+
+
+def test_subworkflows_incorrect_tags_yml_key(self):
+    """Test linting a subworkflow with an incorrect key in tags.yml file"""
+    tags_path = Path(self.nfcore_modules, "subworkflows", "nf-core", "test_subworkflow", "tests", "tags.yml")
+    with open(tags_path) as fh:
+        content = fh.read()
+        new_content = content.replace("test_subworkflow:", "subworkflow:")
+    with open(tags_path, "w") as fh:
+        fh.write(new_content)
+    module_lint = nf_core.subworkflows.SubworkflowLint(dir=self.nfcore_modules)
+    module_lint.lint(print_results=True, subworkflow="test_subworkflow")
+    with open(tags_path, "w") as fh:
+        fh.write(content)
+    assert len(module_lint.failed) == 1, f"Linting failed with {[x.__dict__ for x in module_lint.failed]}"
+    assert len(module_lint.passed) >= 0
+    assert len(module_lint.warned) >= 0
+    assert module_lint.failed[0].lint_test == "test_tags_yml"
+    # cleanup
+    with open(tags_path, "w") as fh:
+        fh.write(content)
+
+
+def test_subworkflows_incorrect_tags_yml_values(self):
+    """Test linting a subworkflow with an incorrect path in tags.yml file"""
+    tags_path = Path(self.nfcore_modules, "subworkflows", "nf-core", "test_subworkflow", "tests", "tags.yml")
+    with open(tags_path) as fh:
+        content = fh.read()
+        new_content = content.replace("subworkflows/nf-core/test_subworkflow/**", "foo")
+    with open(tags_path, "w") as fh:
+        fh.write(new_content)
+    module_lint = nf_core.subworkflows.SubworkflowLint(dir=self.nfcore_modules)
+    module_lint.lint(print_results=False, subworkflow="test_subworkflow")
+    with open(tags_path, "w") as fh:
+        fh.write(content)
+    assert len(module_lint.failed) == 1, f"Linting failed with {[x.__dict__ for x in module_lint.failed]}"
+    assert len(module_lint.passed) >= 0
+    assert len(module_lint.warned) >= 0
+    assert module_lint.failed[0].lint_test == "test_tags_yml"
+    # cleanup
+    with open(tags_path, "w") as fh:
+        fh.write(content)