Add a bunch of optionals, move tests to dedicated file

Chiara Rasi · Chiara Rasi · commit 0133711e88df · 2026-03-04T09:18:40.000+01:00
diff --git a/scout/commands/delete/variants.py b/scout/commands/delete/variants.py
@@ -47,95 +47,109 @@ def _log_case(cases: List[Dict], cid: str, deleted_n: int) -> None:
 
 
 def _process_batch(
-    cases: List[dict],
-    user_obj: dict,
-    rank_threshold: int,
-    variants_threshold: int,
+    cases: List[Dict[str, Any]],
+    user_obj: Dict[str, Any],
+    rank_threshold: int | None,
+    variants_threshold: int | None,
     keep_ctg: Iterable[str],
     dry_run: bool,
 ) -> int:
-    """Process a batch of cases by performing bulk deletion of variants.
-    Returns total number of deleted (or estimated deleted) variants.
     """
-    total_deleted = 0
-    case_ids = [case["_id"] for case in cases]
+    Process a batch of cases by performing bulk deletion of variants.
+
+    Creates events with threshold info, respecting optional thresholds.
+    """
 
-    # ---- OPTIONAL: pre-count variants for threshold ----
-    counts = {}
-    if variants_threshold:
+    total_deleted: int = 0
+    case_ids: List[Any] = [case["_id"] for case in cases]
+
+    # Optional variant count pre-computation
+    counts: Dict[Any, int] = {}
+    if variants_threshold is not None:
         pipeline = [
             {"$match": {"case_id": {"$in": case_ids}}},
             {"$group": {"_id": "$case_id", "n": {"$sum": 1}}},
         ]
-
         for doc in store.variant_collection.aggregate(pipeline, allowDiskUse=True):
             counts[doc["_id"]] = doc["n"]
 
-    delete_ops = []
-    case_delete_map = {}
+    delete_ops: List[DeleteMany] = []
+    case_delete_map: Dict[Any, Dict[str, Any]] = {}
 
     for case in cases:
         cid = case["_id"]
 
-        if variants_threshold and counts.get(cid, 0) < variants_threshold:
+        # Skip case if variants_threshold is set
+        if variants_threshold is not None and counts.get(cid, 0) < variants_threshold:
             continue
 
-        # Positive delete filter only
-        delete_query = {
+        delete_query: Dict[str, Any] = {
             "case_id": cid,
-            "rank_score": {"$lt": rank_threshold},
-            "category": {"$nin": keep_ctg},
+            "category": {"$nin": list(keep_ctg)},
         }
+        if rank_threshold is not None:
+            delete_query["rank_score"] = {"$lt": rank_threshold}
 
         delete_ops.append(DeleteMany(delete_query))
         case_delete_map[cid] = delete_query
 
     if not delete_ops:
         return 0
 
-    # ---- DRY RUN ----
+    # Dry-run mode
     if dry_run:
         for cid, query in case_delete_map.items():
-            n = store.variant_collection.count_documents(query)
+            n: int = store.variant_collection.count_documents(query)
             total_deleted += n
-            _log_case(cases, cid, n)
         return total_deleted
 
-    # ---- REAL DELETE ----
-    result = store.variant_collection.bulk_write(delete_ops, ordered=False)
-
+    # Execute bulk delete
+    result: BulkWriteResult = store.variant_collection.bulk_write(
+        delete_ops,
+        ordered=False,
+    )
     total_deleted += result.deleted_count
 
-    # ---- Post-delete: events + count update ----
+    # Post-delete operations: events + variant count updates
     for case in cases:
         cid = case["_id"]
-
         if cid not in case_delete_map:
             continue
 
-        institute_obj = store.institute(case["owner"])
+        institute_obj: Dict[str, Any] = store.institute(case["owner"])
 
         with current_app.test_request_context("/cases"):
-            url = url_for(
+            url: str = url_for(
                 "cases.case",
                 institute_id=institute_obj["_id"],
                 case_name=case["display_name"],
             )
 
+            # -----------------------------
+            # Build event content string
+            # -----------------------------
+            threshold_parts: List[str] = []
+            if rank_threshold is not None:
+                threshold_parts.append(f"Rank-score threshold:{rank_threshold}")
+            if variants_threshold is not None:
+                threshold_parts.append(f"case n. variants threshold:{variants_threshold}")
+
+            content_str: str = ", ".join(threshold_parts)
+
             store.remove_variants_event(
                 institute=institute_obj,
                 case=case,
                 user=user_obj,
                 link=url,
-                content=f"Rank-score threshold:{rank_threshold}",
+                content=content_str,
             )
 
         store.case_variants_count(cid, institute_obj["_id"], True)
 
     return total_deleted
 
 
-def get_case_ids(case_file: Optional[str], case_id: List[str]) -> List[str]:
+def get_case_ids(case_file: Optional[str], case_id: Optional[List[str]]) -> List[str]:
     """Fetch the _id of the cases to remove variants from."""
     if case_file and case_id:
         click.echo(
@@ -149,7 +163,9 @@ def get_case_ids(case_file: Optional[str], case_id: List[str]) -> List[str]:
     )
 
 
-def _set_keep_ctg(keep_ctg: Tuple[str], rm_ctg: Tuple[str]) -> List[str]:
+def _set_keep_ctg(
+    keep_ctg: Optional[Tuple[str, ...]], rm_ctg: Optional[Tuple[str, ...]]
+) -> List[str]:
     """Define the categories of variants that should not be removed."""
     if keep_ctg and rm_ctg:
         raise click.UsageError("Please use either '--keep-ctg' or '--rm-ctg', not both.")
@@ -208,17 +224,17 @@ def _set_keep_ctg(keep_ctg: Tuple[str], rm_ctg: Tuple[str]) -> List[str]:
 @with_appcontext
 def variants(
     user: str,
-    case_id: tuple,
-    case_file: str,
-    institute: str,
-    status: tuple,
-    older_than: int,
-    analysis_type: tuple,
-    rank_threshold: int,
-    variants_threshold: int,
-    rm_ctg: tuple,
-    keep_ctg: tuple,
-    dry_run: bool,
+    case_id: Optional[Tuple[str, ...]] = None,
+    case_file: Optional[str] = None,
+    institute: Optional[str] = None,
+    status: Optional[Tuple[str, ...]] = None,
+    older_than: Optional[int] = None,
+    analysis_type: Optional[Tuple[str, ...]] = None,
+    rank_threshold: Optional[int] = None,
+    variants_threshold: Optional[int] = None,
+    rm_ctg: Optional[Tuple[str, ...]] = None,
+    keep_ctg: Optional[Tuple[str, ...]] = None,
+    dry_run: bool = False,
 ) -> None:
     """Delete variants for one or more cases"""
 
diff --git a/tests/commands/delete/test_delete_cmd.py b/tests/commands/delete/test_delete_cmd.py
@@ -3,151 +3,6 @@
 from scout.commands.delete.delete_command import CASE_RNA_KEYS
 from scout.server.extensions import store
 
-VARIANTS_QUERY = {"rank_score": {"$lt": 0}}
-RANK_THRESHOLD = 0
-VARIANTS_THRESHOLD = 10
-
-
-def test_delete_variants_dry_run(mock_app, case_obj, user_obj):
-    """test command for cleaning variants collection - simulate deletion"""
-
-    assert store.user_collection.find_one()
-
-    # Given a database with SNV variants
-    runner = mock_app.test_cli_runner()
-    result = runner.invoke(
-        cli, ["load", "variants", case_obj["_id"], "--snv", "--rank-threshold", 5]
-    )
-    assert result.exit_code == 0
-    n_initial_vars = sum(1 for _ in store.variant_collection.find())
-
-    # Then the function that delete variants in dry run should run without error
-    cmd_params = [
-        "delete",
-        "variants",
-        "-u",
-        user_obj["email"],
-        "--status",
-        "inactive",
-        "--older-than",
-        2,
-        "--analysis-type",
-        "wes",
-        "--rank-threshold",
-        RANK_THRESHOLD,
-        "--variants-threshold",
-        VARIANTS_THRESHOLD,
-        "--keep-ctg",
-        "str",
-        "--dry-run",
-    ]
-    result = runner.invoke(cli, cmd_params)
-    assert result.exit_code == 0
-    assert "estimated deleted variants" in result.output
-
-    # And no variants should be deleted
-    assert sum(1 for _ in store.variant_collection.find()) == n_initial_vars
-
-
-def test_delete_variants(mock_app, case_obj, user_obj):
-    """Test deleting variants using the delete variants command line"""
-
-    # Given a case with with SNV variants
-    runner = mock_app.test_cli_runner()
-    result = runner.invoke(
-        cli, ["load", "variants", "--snv", "--rank-threshold", 0, case_obj["_id"]]
-    )
-    assert result.exit_code == 0
-    nr_snvs = sum(1 for _ in store.variant_collection.find())
-
-    # AND WTS outliers
-    result = runner.invoke(
-        cli, ["load", "variants", "--outlier-research", case_obj["_id"], "--force"]
-    )
-    assert result.exit_code == 0
-    nr_outliers = sum(1 for _ in store.omics_variant_collection.find())
-
-    n_initial_vars = nr_snvs + nr_outliers
-
-    # Then the function that delete variants should run without error
-    cmd_params = [
-        "delete",
-        "variants",
-        "-u",
-        user_obj["email"],
-        "--status",
-        "inactive",
-        "--keep-ctg",
-        "outlier",
-        "--older-than",
-        2,
-        "--analysis-type",
-        "wes",
-        "--rank-threshold",
-        RANK_THRESHOLD,
-        "--variants-threshold",
-        VARIANTS_THRESHOLD,
-    ]
-    result = runner.invoke(cli, cmd_params, input="y")
-
-    assert result.exit_code == 0
-    assert "estimated deleted variants" not in result.output
-
-    # variants should be deleted
-    n_current_vars = sum(1 for _ in store.variant_collection.find())
-    assert n_current_vars < n_initial_vars
-
-    # and a relative event should be created
-    event = store.event_collection.find_one({"verb": "remove_variants"})
-    assert event["case"] == case_obj["_id"]
-    assert (
-        event["content"]
-        == f"Rank-score threshold:0, case n. variants threshold:{VARIANTS_THRESHOLD}."
-    )
-    # SNV variants should be gone
-    assert sum(1 for _ in store.variant_collection.find()) == 0
-    # WHILE outliers should still be available
-    assert sum(1 for _ in store.omics_variant_collection.find()) == nr_outliers
-
-
-def test_delete_outlier_variants(mock_app, case_obj, user_obj):
-    """Test the delete variants command's ability to remove omics variants."""
-
-    # Given a case with with (research) outlier variants
-    runner = mock_app.test_cli_runner()
-    result = runner.invoke(
-        cli, ["load", "variants", "--outlier-research", case_obj["_id"], "--force"]
-    )
-    assert result.exit_code == 0
-    n_initial_vars = sum(1 for _ in store.omics_variant_collection.find())
-    assert n_initial_vars
-    n_variants_to_delete = store.omics_variant_collection.count_documents({})
-    assert n_variants_to_delete
-
-    # WHEN variants are removed using the command line
-    cmd_params = [
-        "delete",
-        "variants",
-        "-u",
-        user_obj["email"],
-        "--rank-threshold",
-        0,
-        "--rm-ctg",
-        "outlier",
-    ]
-    result = runner.invoke(cli, cmd_params, input="y")
-    assert result.exit_code == 0
-    assert "estimated deleted variants" not in result.output
-
-    # THEN the variants should be gone
-    n_current_vars = sum(1 for _ in store.variant_collection.find())
-    assert n_current_vars == 0
-    assert n_current_vars + n_variants_to_delete == n_initial_vars
-    # and a relative event should be created
-    event = store.event_collection.find_one({"verb": "remove_variants"})
-    assert event["case"] == case_obj["_id"]
-    assert "Rank-score threshold:0" in event["content"]
-
 
 def test_delete_panel_non_existing(empty_mock_app, testpanel_obj):
     "Test the CLI command that deletes a gene panel"
diff --git a/tests/commands/delete/test_delete_variants_cmd.py b/tests/commands/delete/test_delete_variants_cmd.py
