From f27c1ddf49c2b09fb2d00c476d6c0553d7e671a1 Mon Sep 17 00:00:00 2001 From: Till Hartmann Date: Fri, 31 May 2024 11:09:11 +0200 Subject: [PATCH] wgs_sv_export_external: tests: patch pathlib with fakefs, convert MissingConfiguration to ValidationErrors --- .../workflows/wgs_sv_export_external/model.py | 10 ++-- .../test_workflows_wgs_sv_export_external.py | 50 +++++++++++++++---- 2 files changed, 46 insertions(+), 14 deletions(-) diff --git a/snappy_pipeline/workflows/wgs_sv_export_external/model.py b/snappy_pipeline/workflows/wgs_sv_export_external/model.py index 19152ce32..d2983501b 100644 --- a/snappy_pipeline/workflows/wgs_sv_export_external/model.py +++ b/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 diff --git a/tests/snappy_pipeline/workflows/test_workflows_wgs_sv_export_external.py b/tests/snappy_pipeline/workflows/test_workflows_wgs_sv_export_external.py index d2c5e8e97..02e4fef63 100644 --- a/tests/snappy_pipeline/workflows/test_workflows_wgs_sv_export_external.py +++ b/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,12 +100,13 @@ 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, @@ -112,9 +114,15 @@ def test_workflow_check_config_invalid_annotator_files( 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,12 +143,13 @@ 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, @@ -148,7 +157,11 @@ def test_workflow_check_config_invalid_search_directory( 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,7 +196,7 @@ 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, @@ -190,7 +204,25 @@ def test_workflow_check_config_invalid_search_pattern( 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) -----------------------------------------