wgs_sv_export_external: tests: patch pathlib with fakefs, convert Mis…

…singConfiguration to ValidationErrors

snappy_pipeline/workflows/wgs_sv_export_external/model.py

@@ -1,6 +1,6 @@
 from typing import Annotated
 
-from pydantic import Field
+from pydantic import Field, FilePath, DirectoryPath
 
 from snappy_pipeline.models import SnappyStepModel
 
@@ -18,7 +18,7 @@ class WgsSvExportExternal(SnappyStepModel):
     merge_option: str = "id"
     """How to merge VCF, used in `bcftools --merge` call."""
 
-    search_paths: Annotated[list[str], Field(min_length=1)]
+    search_paths: Annotated[list[DirectoryPath], Field(min_length=1)]
     """path to all VCF files."""
 
     search_patterns: Annotated[
@@ -28,13 +28,13 @@ class WgsSvExportExternal(SnappyStepModel):
 
     release: str = "GRCh37"
 
-    path_refseq_ser: str
+    path_refseq_ser: FilePath
     """path to RefSeq .ser file"""
 
-    path_ensembl_ser: str
+    path_ensembl_ser: FilePath
     """path to ENSEMBL .ser file"""
 
-    path_db: str
+    path_db: FilePath
     """path to annotator DB file to use"""
 
     varfish_server_compatibility: bool = False

tests/snappy_pipeline/workflows/test_workflows_wgs_sv_export_external.py

@@ -5,9 +5,9 @@
 
 import pytest
 import ruamel.yaml as ruamel_yaml
+from pydantic import ValidationError
 from snakemake.io import Wildcards
 
-from snappy_pipeline.base import MissingConfiguration
 from snappy_pipeline.workflows.wgs_sv_export_external import WgsSvExportExternalWorkflow
 
 from .common import get_expected_log_files_dict, get_expected_output_vcf_files_dict
@@ -72,6 +72,7 @@ def wgs_sv_export_external_workflow(
     # Create search path
     germline_sheet_fake_fs.fs.makedirs("/search_path")
     # Patch out file-system related things in abstract (the crawling link in step is defined there)
+    patch_module_fs("pathlib", germline_sheet_fake_fs, mocker)
     patch_module_fs("snappy_pipeline.workflows.abstract", germline_sheet_fake_fs, mocker)
     patch_module_fs(
         "snappy_pipeline.workflows.wgs_sv_export_external", germline_sheet_fake_fs, mocker
@@ -99,22 +100,29 @@ def test_workflow_check_config_invalid_annotator_files(
     # Create search path
     germline_sheet_fake_fs.fs.makedirs("/search_path")
     # Patch out file-system related things in abstract (the crawling link in step is defined there)
+    patch_module_fs("pathlib", germline_sheet_fake_fs, mocker)
     patch_module_fs("snappy_pipeline.workflows.abstract", germline_sheet_fake_fs, mocker)
     patch_module_fs(
         "snappy_pipeline.workflows.wgs_sv_export_external", germline_sheet_fake_fs, mocker
     )
     # Construct the workflow object
-    with pytest.raises(MissingConfiguration) as exec_info:
+    with pytest.raises(ValidationError) as exec_info:
         WgsSvExportExternalWorkflow(
             dummy_workflow,
             minimal_config,
             config_lookup_paths,
             config_paths,
             work_dir,
         )
-    assert "path_refseq_ser" in exec_info.value.args[0]
-    assert "path_ensembl_ser" in exec_info.value.args[0]
-    assert "path_db" in exec_info.value.args[0]
+    errors = exec_info.value.errors()
+    assert len(errors) == 3
+    assert (
+        len(
+            {"path_refseq_ser", "path_ensembl_ser", "path_db"}
+            & set(s for e in errors for s in e["loc"])
+        )
+        == 3
+    )
 
 
 def test_workflow_check_config_invalid_search_directory(
@@ -135,20 +143,25 @@ def test_workflow_check_config_invalid_search_directory(
             create_missing_dirs=True,
         )
     # Patch out file-system related things in abstract (the crawling link in step is defined there)
+    patch_module_fs("pathlib", germline_sheet_fake_fs, mocker)
     patch_module_fs("snappy_pipeline.workflows.abstract", germline_sheet_fake_fs, mocker)
     patch_module_fs(
         "snappy_pipeline.workflows.wgs_sv_export_external", germline_sheet_fake_fs, mocker
     )
     # Construct the workflow object
-    with pytest.raises(MissingConfiguration) as exec_info:
+    with pytest.raises(ValidationError) as exec_info:
         WgsSvExportExternalWorkflow(
             dummy_workflow,
             minimal_config,
             config_lookup_paths,
             config_paths,
             work_dir,
         )
-    assert " is not a directory: /search_path" in exec_info.value.args[0]
+
+    errors = exec_info.value.errors()
+    assert len(errors) == 1
+
+    assert "path_not_directory" in {e["type"] for e in errors}
 
 
 def test_workflow_check_config_invalid_search_pattern(
@@ -171,6 +184,7 @@ def test_workflow_check_config_invalid_search_pattern(
     # Create search path
     germline_sheet_fake_fs.fs.makedirs("/search_path")
     # Patch out file-system related things in abstract (the crawling link in step is defined there)
+    patch_module_fs("pathlib", germline_sheet_fake_fs, mocker)
     patch_module_fs("snappy_pipeline.workflows.abstract", germline_sheet_fake_fs, mocker)
     patch_module_fs(
         "snappy_pipeline.workflows.wgs_sv_export_external", germline_sheet_fake_fs, mocker
@@ -182,15 +196,33 @@ def test_workflow_check_config_invalid_search_pattern(
         "*/*.vcf.gz",
     ]
     # Construct the workflow object
-    with pytest.raises(MissingConfiguration) as exec_info:
+    with pytest.raises(ValidationError) as exec_info:
         WgsSvExportExternalWorkflow(
             dummy_workflow,
             modified_config,
             config_lookup_paths,
             config_paths,
             work_dir,
         )
-    assert "Value in 'search_patterns' is not a dictionary" in exec_info.value.args[0]
+
+    errors = exec_info.value.errors()
+
+    # there is 1 incorrectly defined search_patterns entry
+    # which is incorrectly defined as a list instead of a dict/key-value pairs,
+    # so pydantic tries to parse *2* dicts from the list and fails
+    assert len(errors) == 2
+
+    expected_errors = [
+        {
+            "input": input_str,
+            "loc": ("step_config", "wgs_sv_export_external", "search_patterns", i),
+            "msg": "Input should be a valid dictionary",
+            "type": "dict_type",
+            "url": "https://errors.pydantic.dev/2.7/v/dict_type",
+        }
+        for i, input_str in enumerate(["vcf", "*/*.vcf.gz"])
+    ]
+    assert expected_errors == errors
 
 
 # Tests for VarfishAnnotatorExternalStepPart (merge_vcf)   -----------------------------------------