Add e2e tests and logic for indexing .vcf

Turns out that bcftools requires bgzipped VCFs for indexing. Currently,
DivBase enforces .vcf.gz but if this were to change, the dimensions
update logic would suddenly break for .vcf files. This adds bgzipping
steps to the VCFDimensionCalculator.

Author: brinkdp

packages/divbase-api/src/divbase_api/worker/vcf_dimension_indexing.py

@@ -29,15 +29,36 @@ def calculate_dimensions(self, vcf_path: Path) -> VCFDimensions | None:
 
         Returns None if this is a DivBase-generated result file (should be skipped).
         This is checked for when the header is parsed in _extract_sample_names_from_vcf_header.
+
+        Note: bcftools index --csi requires bgzipped input. Plain .vcf files are bgzipped to a
+        temporary .vcf.gz file for indexing, then the temporary file is cleaned up internally.
         """
         logger.debug(f"Reading: {vcf_path} ...")
 
         sample_names = self._extract_sample_names_from_vcf_header(vcf_path)
         if sample_names is None:
             return None
 
-        csi_index_path = self._index_vcf_with_csi(vcf_path)
-        scaffold_names, variant_count = self._extract_scaffold_names_and_variant_count_from_csi_index(csi_index_path)
+        indexing_path = vcf_path
+
+        if vcf_path.suffix == ".vcf":
+            bgzipped_temp = Path(str(vcf_path) + ".gz")
+            self._bgzip_vcf(vcf_path, bgzipped_temp)
+            indexing_path = bgzipped_temp
+
+        try:
+            csi_index_path = self._index_vcf_with_csi(vcf_path=indexing_path)
+            scaffold_names, variant_count = self._extract_scaffold_names_and_variant_count_from_csi_index(
+                csi_index_path=csi_index_path
+            )
+        finally:
+            # Clean up temp bgzipped file and its CSI index; these are not tracked by tasks.py
+            if bgzipped_temp is not None:
+                bgzipped_temp_csi = Path(str(bgzipped_temp) + ".csi")
+                if bgzipped_temp.exists():
+                    bgzipped_temp.unlink()
+                if bgzipped_temp_csi.exists():
+                    bgzipped_temp_csi.unlink()
 
         return VCFDimensions(
             variants=variant_count,
@@ -73,10 +94,27 @@ def _extract_sample_names_from_vcf_header(self, vcf_path: Path) -> List[str] | N
             logger.error(f"Error extracting sample names from the VCF header {vcf_path}: {e}")
             raise
 
+    def _bgzip_vcf(self, vcf_path: Path, output_path: Path) -> None:
+        """
+        Bgzip a plain .vcf file to a bgzipped .vcf.gz file using bcftools view.
+        Required because bcftools index --csi only accepts bgzipped input.
+        """
+        try:
+            subprocess.run(
+                ["bcftools", "view", "-Oz", "-o", str(output_path), str(vcf_path)],
+                check=True,
+                stderr=subprocess.DEVNULL,
+            )
+            logger.info(f"Bgzipped {vcf_path} to {output_path} for CSI indexing.")
+        except subprocess.CalledProcessError as e:
+            logger.error(f"Error bgzipping {vcf_path}: {e}")
+            raise
+
     def _index_vcf_with_csi(self, vcf_path: Path) -> Path:
         """
         Index the VCF file with CSI index using bcftools.
         The CSI index is then used to extract dimensions information.
+        Input must be bgzipped (.vcf.gz); use _bgzip_vcf first for plain .vcf files.
         """
         csi_index_path = vcf_path.with_suffix(vcf_path.suffix + ".csi")
         try:

tests/e2e_integration/cli_commands/test_dimensions_cli.py

@@ -890,3 +890,106 @@ def test_validate_metadata_file_nonexistent(
 
     assert cli_result.exit_code == 2, "Expected exit code 2 for nonexistent file (Typer path validation)"
     assert "does not exist" in cli_result.output.lower(), "Expected error message about file not existing"
+
+
+@patch("divbase_api.worker.tasks.create_s3_file_manager")
+def test_update_dimensions_cleans_up_csi_index_files_from_worker(
+    mock_create_s3_manager,
+    CONSTANTS,
+    project_map,
+    tmp_path,
+    monkeypatch,
+):
+    """
+    Test that CSI index files created during dimension calculation are deleted from the worker
+    filesystem after update_vcf_dimensions_task completes.
+    """
+    mock_create_s3_manager.side_effect = lambda url=None: create_s3_file_manager(url=CONSTANTS["MINIO_URL"])
+    monkeypatch.chdir(tmp_path)
+
+    project_name = CONSTANTS["SPLIT_SCAFFOLD_PROJECT"]
+    bucket_name = CONSTANTS["PROJECT_TO_BUCKET_MAP"][project_name]
+    project_id = project_map[project_name]
+    user_id = 1
+
+    result = update_vcf_dimensions_task(
+        bucket_name=bucket_name, project_id=project_id, project_name=project_name, user_id=user_id
+    )
+
+    assert result["status"] == "completed"
+
+    vcf_files_remaining = list(tmp_path.glob("*.vcf")) + list(tmp_path.glob("*.vcf.gz"))
+    assert vcf_files_remaining == [], (
+        f"Expected all VCF files to be deleted from worker after task, but found: {vcf_files_remaining}"
+    )
+
+    csi_files_remaining = list(tmp_path.glob("*.csi"))
+    assert csi_files_remaining == [], (
+        f"Expected all CSI index files to be deleted from worker after task, but found: {csi_files_remaining}"
+    )
+
+
+@patch("divbase_api.worker.tasks.create_s3_file_manager")
+def test_update_dimensions_indexes_uncompressed_vcf(
+    mock_create_s3_manager,
+    CONSTANTS,
+    db_session_sync,
+    project_map,
+    tmp_path,
+):
+    """
+    Test that update_vcf_dimensions_task correctly indexes a plain uncompressed .vcf file.
+
+    bcftools index --csi requires bgzipped input, so calculate_dimensions bgzips plain .vcf
+    files to a temp .vcf.gz internally before indexing, then cleans up the temp files.
+    This tests that the full indexing pipeline works end-to-end for uncompressed VCFs.
+    """
+    mock_create_s3_manager.side_effect = lambda url=None: create_s3_file_manager(url=CONSTANTS["MINIO_URL"])
+
+    project_name = CONSTANTS["SPLIT_SCAFFOLD_PROJECT"]
+    bucket_name = CONSTANTS["PROJECT_TO_BUCKET_MAP"][project_name]
+    project_id = project_map[project_name]
+    user_id = 1
+
+    plain_vcf_name = "test_uncompressed_plain.vcf"
+    vcf_path = tmp_path / plain_vcf_name
+    vcf_content = (
+        "##fileformat=VCFv4.2\n"
+        "##contig=<ID=1,length=22053058>\n"
+        "#CHROM\tPOS\tID\tREF\tALT\tQUAL\tFILTER\tINFO\tFORMAT\tsample1\tsample2\n"
+        "1\t17504018\t.\tC\tA\t.\tPASS\t.\tGT\t0/1\t0/0\n"
+        "1\t22053057\t.\tG\tA\t.\tPASS\t.\tGT\t1/1\t0/1\n"
+    )
+    vcf_path.write_text(vcf_content)
+
+    s3_file_manager = create_s3_file_manager(url=CONSTANTS["MINIO_URL"])
+    s3_file_manager.upload_files(
+        to_upload={plain_vcf_name: vcf_path},
+        bucket_name=bucket_name,
+    )
+
+    try:
+        result = update_vcf_dimensions_task(
+            bucket_name=bucket_name, project_id=project_id, project_name=project_name, user_id=user_id
+        )
+
+        assert result["status"] == "completed"
+        indexed_files = result.get("VCF_files_added") or []
+        assert plain_vcf_name in indexed_files, f"Expected plain .vcf file to be indexed, got: {indexed_files}"
+
+        # Verify the dimensions were stored correctly in the DB
+        db_session_sync.expire_all()
+        vcf_dimensions = get_vcf_metadata_by_project(project_id=project_id, db=db_session_sync)
+        all_indexed_keys = [entry["vcf_file_s3_key"] for entry in vcf_dimensions.get("vcf_files", [])]
+        assert plain_vcf_name in all_indexed_keys
+
+        entry = next(e for e in vcf_dimensions["vcf_files"] if e["vcf_file_s3_key"] == plain_vcf_name)
+        assert entry["sample_count"] == 2
+        assert entry["variant_count"] == 2
+        assert "1" in entry["scaffolds"]
+        assert "sample1" in entry["samples"]
+        assert "sample2" in entry["samples"]
+    finally:
+        # Remove the uploaded file from the bucket to avoid polluting subsequent tests.
+        # auto_clean_dimensions_entries_for_all_projects only cleans the DB, not the bucket.
+        s3_file_manager.soft_delete_objects(objects=[plain_vcf_name], bucket_name=bucket_name)