somatic_purity_ploidy_estimate: model: fix missing 'somatic_targeted_…

…seq_cnv_calling' entrypoint and add validator, base: remove todo because of validator, tests: fix minimal config, fakefs

snappy_pipeline/workflows/somatic_purity_ploidy_estimate/__init__.py

@@ -312,8 +312,6 @@ def __init__(self, workflow, config, config_lookup_paths, config_paths, workdir)
         self.register_sub_step_classes((AscatStepPart, LinkOutStepPart))
         # Initialize sub-workflows
         self.register_sub_workflow("ngs_mapping", self.config.path_ngs_mapping)
-        # TODO: potential bug here as this step requires an entry that is not available
-        #  in DEFAULT_CONFIG.
         if self.config.tool_cnv_calling == "copywriter":
             self.register_sub_workflow(
                 "somatic_targeted_seq_cnv_calling",

snappy_pipeline/workflows/somatic_purity_ploidy_estimate/model.py

@@ -1,6 +1,8 @@
 import enum
 from typing import Annotated
 
+from pydantic import model_validator
+
 from snappy_pipeline.models import EnumField, SnappyModel, SnappyStepModel, validators
 
 
@@ -28,4 +30,15 @@ class SomaticPurityPloidyEstimate(SnappyStepModel, validators.ToolsMixin):
 
     path_ngs_mapping: str = "../ngs_mapping"
 
+    path_somatic_targeted_seq_cnv_calling: str = ""
+
     ascat: Ascat | None = None
+
+    @model_validator(mode="after")
+    def check_tool_cnv_calling(self):
+        if self.tool_cnv_calling == "copywriter" and not self.path_somatic_targeted_seq_cnv_calling:
+            raise ValueError(
+                "When using 'copywriter' as tool_cnv_calling, "
+                "path_somatic_targeted_seq_cnv_calling must be set"
+            )
+        return self

tests/snappy_pipeline/workflows/test_workflows_somatic_purity_ploidy_estimate.py

@@ -36,6 +36,8 @@ def minimal_config():
             tools: ['ascat']
             tool_cnv_calling: cnvetti
             path_somatic_targeted_seq_cnv_calling: ../somatic_targeted_seq_cnv_calling
+            ascat:
+              b_af_loci: DUMMY
 
         data_sets:
           first_batch:
@@ -87,18 +89,20 @@ def somatic_purity_ploidy_estimate_workflow(
 
 
 @pytest.fixture
-def somatic_purity_ploidy_estimate_workflow_w_copywritter(
+def somatic_purity_ploidy_estimate_workflow_w_copywriter(
     dummy_workflow,
     minimal_config_copywritter,
     config_lookup_paths,
     work_dir,
     config_paths,
     cancer_sheet_fake_fs,
+    aligner_indices_fake_fs,
     mocker,
 ):
     """Return SomaticPurityPloidyEstimateWorkflow object pre-configured with cancer sheet"""
     # Patch out file-system related things in abstract (the crawling link in step is defined there)
     patch_module_fs("snappy_pipeline.workflows.abstract", cancer_sheet_fake_fs, mocker)
+    patch_module_fs("snappy_pipeline.workflows.ngs_mapping", aligner_indices_fake_fs, mocker)
     dummy_workflow.globals = {
         "ngs_mapping": lambda x: "NGS_MAPPING/" + x,
         "somatic_targeted_seq_cnv_calling": lambda x: "SOMATIC_CNV_CALLING/" + x,
@@ -169,7 +173,7 @@ def test_ascat_step_part_get_input_files_cnv_normal(somatic_purity_ploidy_estima
 
 
 def test_ascat_step_part_get_input_files_cnv_tumor_wes(
-    somatic_purity_ploidy_estimate_workflow_w_copywritter,
+    somatic_purity_ploidy_estimate_workflow_w_copywriter,
 ):
     """Tests AscatStepPart._get_input_files_cnv_tumor_wes()"""
     wildcards = Wildcards(fromdict={"tumor_library_name": "P001-T1-DNA1-WGS1", "mapper": "bwa"})
@@ -179,14 +183,14 @@ def test_ascat_step_part_get_input_files_cnv_tumor_wes(
             "bwa.copywriter.P001-T1-DNA1-WGS1_bins.txt"
         )
     }
-    actual = somatic_purity_ploidy_estimate_workflow_w_copywritter.get_input_files(
+    actual = somatic_purity_ploidy_estimate_workflow_w_copywriter.get_input_files(
         "ascat", "cnv_tumor_wes"
     )(wildcards)
     assert actual == expected
 
 
 def test_ascat_step_part_get_input_files_cnv_normal_wes(
-    somatic_purity_ploidy_estimate_workflow_w_copywritter,
+    somatic_purity_ploidy_estimate_workflow_w_copywriter,
 ):
     """Tests AscatStepPart._get_input_files_cnv_normal_wes()"""
     wildcards = Wildcards(fromdict={"normal_library_name": "P001-N1-DNA1-WGS1", "mapper": "bwa"})
@@ -196,7 +200,7 @@ def test_ascat_step_part_get_input_files_cnv_normal_wes(
             "bwa.copywriter.P001-T1-DNA1-WGS1_bins.txt"
         )
     }
-    actual = somatic_purity_ploidy_estimate_workflow_w_copywritter.get_input_files(
+    actual = somatic_purity_ploidy_estimate_workflow_w_copywriter.get_input_files(
         "ascat", "cnv_normal_wes"
     )(wildcards)
     assert actual == expected