diff --git a/src/anyvlm/anyvar/base_client.py b/src/anyvlm/anyvar/base_client.py index 6ac83e5..d3fca90 100644 --- a/src/anyvlm/anyvar/base_client.py +++ b/src/anyvlm/anyvar/base_client.py @@ -3,10 +3,11 @@ import abc from collections.abc import Iterable, Sequence -from anyvar.core.objects import SupportedVrsVariation from anyvar.mapping.liftover import ReferenceAssembly from ga4gh.vrs.models import Allele +from anyvlm.anyvar.types import SupportedVrsVariation + class AnyVarClientError(Exception): """Generic client-related exception.""" diff --git a/src/anyvlm/anyvar/python_client.py b/src/anyvlm/anyvar/python_client.py index 85ea297..e10bb13 100644 --- a/src/anyvlm/anyvar/python_client.py +++ b/src/anyvlm/anyvar/python_client.py @@ -5,7 +5,6 @@ from anyvar import AnyVar from anyvar.core.metadata import VariationMapping, VariationMappingType -from anyvar.core.objects import SupportedVrsVariation from anyvar.mapping.liftover import ReferenceAssembly from anyvar.restapi.schema import SupportedVariationType from anyvar.storage.base import Storage @@ -14,6 +13,7 @@ from ga4gh.vrs.models import Allele from anyvlm.anyvar.base_client import BaseAnyVarClient +from anyvlm.anyvar.types import SupportedVrsVariation _logger = logging.getLogger(__name__) diff --git a/src/anyvlm/anyvar/types.py b/src/anyvlm/anyvar/types.py new file mode 100644 index 0000000..3a182c7 --- /dev/null +++ b/src/anyvlm/anyvar/types.py @@ -0,0 +1,13 @@ +"""Compatibility aliases for AnyVar/VRS object types. + +These names preserve AnyVLM's broader type intent even when upstream AnyVar stops +exporting its convenience aliases. +""" + +from ga4gh.vrs.models import Allele + +try: + from anyvar.core.objects import SupportedVrsObject, SupportedVrsVariation +except ImportError: + SupportedVrsObject = Allele + SupportedVrsVariation = Allele diff --git a/src/anyvlm/cli.py b/src/anyvlm/cli.py index 5d898ad..3bb9425 100644 --- a/src/anyvlm/cli.py +++ b/src/anyvlm/cli.py @@ -1,19 +1,29 @@ """CLI for interacting with AnyVLM instance""" import logging -from http import HTTPStatus from pathlib import Path from timeit import default_timer as timer import click -import requests from anyvar.mapping.liftover import ReferenceAssembly import anyvlm +from anyvlm.anyvar.base_client import BaseAnyVarClient from anyvlm.config import Settings, get_config +from anyvlm.functions.ingest_vcf import ingest_vcf as ingest_vcf_function +from anyvlm.main import create_anyvar_client, create_anyvlm_storage +from anyvlm.storage import Storage + +# Create alias for easier mocking in tests +ingest_vcf = ingest_vcf_function _logger = logging.getLogger(__name__) +# Constants +MAX_FILE_SIZE = 5 * 1024 * 1024 * 1024 # 5GB +UPLOAD_CHUNK_SIZE = 1024 * 1024 # 1MB +REQUIRED_INFO_FIELDS = {"AC", "AN", "AC_Het", "AC_Hom", "AC_Hemi"} + @click.version_option(anyvlm.__version__) @click.group() @@ -22,10 +32,10 @@ def _cli() -> None: logging.basicConfig(filename="anyvlm.log", level=logging.INFO) -@_cli.command() +@_cli.command(name="ingest-vcf") @click.option( "--file", - "vcf_path", + "vcf_file_path", type=click.Path(exists=True, dir_okay=False, path_type=Path), required=True, help="Path to a gzip-compressed VCF file (.vcf.gz)", @@ -39,44 +49,31 @@ def _cli() -> None: callback=lambda _, __, value: ReferenceAssembly(value), help="Reference genome assembly", ) -def ingest_vcf(vcf_path: Path, assembly: ReferenceAssembly) -> None: +def ingest_vcf_cli_wrapper(vcf_file_path: Path, assembly: ReferenceAssembly) -> None: """Deposit variants and allele frequencies from VCF into AnyVLM instance $ anyvlm ingest-vcf --file path/to/file.vcf.gz --assembly grch38 """ start: float = timer() - _logger.info( "Starting VCF ingestion: file='%s', assembly='%s'", - str(vcf_path), + str(vcf_file_path), assembly.value, ) config: Settings = get_config() - endpoint: str = f"{config.service_uri}/ingest_vcf" - - params = {"assembly": assembly.value} - - with vcf_path.open("rb") as fh: - files = {"file": (vcf_path.name, fh, "application/gzip")} - - try: - response: requests.Response = requests.post( - endpoint, - files=files, - params=params, - timeout=3600, # 1 hour - ) - except requests.RequestException as e: - _logger.exception("HTTP POST request to AnyVLM '/ingest_vcf' failed") - raise click.ClickException(str(e)) from e - - if response.status_code != HTTPStatus.OK: - _logger.error("Request failed with status code %s", response.status_code) - raise click.ClickException( - f"Request failed with status code: {response.status_code}" - ) + anyvar_client: BaseAnyVarClient = create_anyvar_client( + connection_string=config.anyvar_uri + ) + anyvlm_storage: Storage = create_anyvlm_storage(uri=config.storage_uri) + ingest_vcf_function( + vcf_path=vcf_file_path, + av=anyvar_client, + storage=anyvlm_storage, + assembly=assembly, + ) end: float = timer() duration: float = end - start _logger.info("Ingestion complete in %s", f"{duration:.3f} seconds") + print("✅ Ingestion complete") # noqa: T201 diff --git a/src/anyvlm/functions/ingest_vcf.py b/src/anyvlm/functions/ingest_vcf.py index bdbc245..cba482a 100644 --- a/src/anyvlm/functions/ingest_vcf.py +++ b/src/anyvlm/functions/ingest_vcf.py @@ -1,14 +1,17 @@ """Get a VCF, register its contained variants, and add cohort frequency data to storage""" import logging -from collections import namedtuple from collections.abc import Iterator +from enum import StrEnum +from logging import Logger from pathlib import Path +from typing import NamedTuple import pysam from anyvar.mapping.liftover import ReferenceAssembly from ga4gh.core.models import iriReference from ga4gh.va_spec.base import StudyGroup +from pysam.libcbcf import VariantRecordInfo from anyvlm.anyvar.base_client import BaseAnyVarClient from anyvlm.storage.base_storage import Storage @@ -18,16 +21,47 @@ QualityMeasures, ) -_logger = logging.getLogger(__name__) +_logger: Logger = logging.getLogger(__name__) -AfData = namedtuple("AfData", ("ac", "an", "ac_het", "ac_hom", "ac_hemi", "filters")) +class VcfInfoField(StrEnum): + """Represents required info fields""" + + AC = "AC" + AN = "AN" + AC_HET = "AC_Het" + AC_HOM = "AC_Hom" + AC_HEMI = "AC_Hemi" + + +REQUIRED_INFO_FIELDS: frozenset[VcfInfoField] = frozenset(VcfInfoField) + + +class AfData(NamedTuple): + """Represents Af data""" + + ac: int + an: int + ac_het: int + ac_hom: int + ac_hemi: int + filters: object class VcfAfColumnsError(Exception): """Raise for missing VCF INFO columns that are required for AF ingestion""" +def _validate_vcf_header(vcf: pysam.VariantFile) -> None: + """Validate that a VCF header includes the required INFO fields.""" + found_fields: set[str] = set[str](vcf.header.info.keys()) + missing: frozenset[VcfInfoField] = REQUIRED_INFO_FIELDS - found_fields + if missing: + raise VcfAfColumnsError( + f"VCF ingestion failed: missing required INFO fields: {', '.join(sorted(missing))}" + ) + + def _yield_expression_af_batches( vcf: pysam.VariantFile, batch_size: int = 1000 ) -> Iterator[list[tuple[str, AfData]]]: @@ -47,9 +81,9 @@ def _yield_expression_af_batches( if record.ref is None or "*" in record.ref or "*" in alt: _logger.info("Skipping missing allele at %s", record) continue - expression = f"{record.chrom}-{record.pos}-{record.ref}-{alt}" + expression: str = f"{record.chrom}-{record.pos}-{record.ref}-{alt}" try: - af = AfData( + af: AfData = AfData( ac=record.info["AC"][i], an=record.info["AN"], ac_het=record.info["AC_Het"][i], @@ -58,8 +92,8 @@ def _yield_expression_af_batches( filters=record.filter.keys(), ) except KeyError as e: - info = record.info - msg = f"One or more required INFO column is missing: {'AC' in info}, {'AN' in info}, {'AC_Het' in info}, {'AC_Hom' in info}, {'AC_Hemi' in info}" + info: VariantRecordInfo = record.info + msg: str = f"One or more required INFO column is missing: {VcfInfoField.AC in info}, {VcfInfoField.AN in info}, {VcfInfoField.AC_HET in info}, {VcfInfoField.AC_HOM in info}, {VcfInfoField.AC_HEMI in info}" _logger.exception(msg) raise VcfAfColumnsError(msg) from e if af.an == 0: @@ -101,16 +135,27 @@ def ingest_vcf( :param av: AnyVar client :param storage: AnyVLM storage instance :param assembly: reference assembly used by VCF + :raise ValueError: if VCF is unreadable or missing required INFO fields :raise VcfAfColumnsError: if VCF is missing required columns """ pysam.set_verbosity(0) # silences warning re: lack of an index for the vcf file - vcf = pysam.VariantFile(filename=vcf_path.absolute().as_uri(), mode="r") + + try: + vcf = pysam.VariantFile(filename=vcf_path.absolute().as_uri(), mode="r") + except ValueError as e: + error_message = ( + "VCF ingestion failed: Not a valid VCF file (missing format declaration)" + ) + _logger.exception(msg=error_message) + raise ValueError(error_message) from e + + _validate_vcf_header(vcf) for batch in _yield_expression_af_batches(vcf): expressions, afs = zip(*batch, strict=True) variant_ids = av.put_allele_expressions(expressions, assembly) - cafs = [] + cafs: list[AnyVlmCohortAlleleFrequencyResult] = [] for variant_id, af in zip(variant_ids, afs, strict=True): if variant_id is None: continue diff --git a/src/anyvlm/restapi/vlm.py b/src/anyvlm/restapi/vlm.py index 4c7f18e..0e8ecb5 100644 --- a/src/anyvlm/restapi/vlm.py +++ b/src/anyvlm/restapi/vlm.py @@ -1,28 +1,19 @@ """Define route(s) for the variant-level matching (VLM) protocol""" -import gzip import logging -import tempfile -import uuid from http import HTTPStatus -from pathlib import Path -from typing import Annotated, BinaryIO, Literal +from typing import Annotated -from anyvar.mapping.liftover import ReferenceAssembly from fastapi import ( APIRouter, HTTPException, Query, Request, - UploadFile, ) -from pydantic import BaseModel from anyvlm.anyvar.base_client import AnyVarClientConnectionError, BaseAnyVarClient from anyvlm.functions.build_vlm_response import build_vlm_response from anyvlm.functions.get_cafs import get_cafs -from anyvlm.functions.ingest_vcf import VcfAfColumnsError -from anyvlm.functions.ingest_vcf import ingest_vcf as ingest_vcf_function from anyvlm.schemas.vlm import VlmResponse from anyvlm.storage.base_storage import Storage from anyvlm.utils.exceptions import VariantLookupError @@ -35,251 +26,15 @@ UcscAssemblyBuild, ) -# Create alias for easier mocking in tests -ingest_vcf = ingest_vcf_function - _logger = logging.getLogger(__name__) router = APIRouter() -# Constants -MAX_FILE_SIZE = 5 * 1024 * 1024 * 1024 # 5GB -UPLOAD_CHUNK_SIZE = 1024 * 1024 # 1MB -REQUIRED_INFO_FIELDS = {"AC", "AN", "AC_Het", "AC_Hom", "AC_Hemi"} - - -# ==================== -# Response Models -# ==================== - - -class VcfIngestionResponse(BaseModel): - """Response model for VCF ingestion endpoint.""" - - status: Literal["success", "error"] - message: str - details: str | None = None - - -# ==================== -# Validation Helpers -# ==================== - - -def validate_filename_extension(filename: str) -> None: - """Validate that filename has .vcf.gz extension. - - :param filename: name of uploaded file - :raise ValueError: if extension is not .vcf.gz - """ - if not filename.endswith(".vcf.gz"): - raise ValueError("Only .vcf.gz files are accepted") - - -def validate_gzip_magic_bytes(file_obj: BinaryIO) -> None: - """Validate that file has gzip magic bytes. - - :param file_obj: file-like object to validate - :raise ValueError: if file is not gzipped - """ - header = file_obj.read(2) - file_obj.seek(0) # Reset file pointer - - if header != b"\x1f\x8b": - raise ValueError("File is not a valid gzip file") - - -def validate_file_size(size: int) -> None: - """Validate that file size is within limits. - - :param size: file size in bytes - :raise ValueError: if file exceeds maximum size - """ - if size > MAX_FILE_SIZE: - max_gb = MAX_FILE_SIZE / (1024**3) - raise ValueError(f"File too large. Maximum size: {max_gb:.1f}GB") - - -def validate_vcf_header(file_path: Path) -> None: - """Validate VCF file format and required INFO fields. - - :param file_path: path to VCF file - :raise ValueError: if VCF is malformed or missing required fields - """ - with gzip.open(file_path, "rt") as f: - # Check first line is VCF format declaration - first_line = f.readline().strip() - if not first_line.startswith("##fileformat=VCF"): - raise ValueError("Not a valid VCF file (missing format declaration)") - - # Scan headers for required INFO fields - found_fields = set() - - for line in f: - if line.startswith("##INFO= Path: - """Save uploaded file to temporary location using streaming. - - :param upload_file: FastAPI UploadFile object - :return: path to saved temporary file - :raise: Any exceptions during file operations (caller should handle cleanup) - """ - temp_dir = Path(tempfile.gettempdir()) - temp_path = temp_dir / f"anyvlm_{uuid.uuid4()}.vcf.gz" - - try: - # Stream upload to disk (memory efficient) - # Using blocking I/O here is acceptable as we're writing to local disk - with temp_path.open("wb") as f: - while chunk := await upload_file.read(UPLOAD_CHUNK_SIZE): - f.write(chunk) - except Exception: - # Cleanup on error - if temp_path.exists(): - temp_path.unlink() - raise - else: - return temp_path - # ==================== # Endpoints # ==================== - -@router.post( - "/ingest_vcf", - summary="Upload and ingest VCF file", - description=( - "Upload a compressed VCF file (.vcf.gz) to register variants and store allele frequency data. " - "**Requirements:** File must be gzip-compressed (.vcf.gz), contain required INFO fields " - "(AC, AN, AC_Het, AC_Hom, AC_Hemi), and be under 5GB. " - "Processing is synchronous with a 30-minute timeout." - ), - tags=[EndpointTag.SEARCH], -) -async def ingest_vcf_endpoint( - request: Request, - file: UploadFile, - assembly: Annotated[ - ReferenceAssembly, - Query(..., description="Reference genome assembly (GRCh37 or GRCh38)"), - ], -) -> VcfIngestionResponse: - """Upload and ingest a VCF file with allele frequency data. - - Requirements: .vcf.gz format, <5GB, INFO fields (AC, AN, AC_Het, AC_Hom, AC_Hemi). - Synchronous processing with 30-minute timeout. Variants batched in groups of 1000. - - :param request: FastAPI request object - :param file: uploaded VCF file - :param assembly: reference assembly used in VCF - :return: ingestion status response - """ - temp_path: Path | None = None - - try: - # Validate filename extension - if not file.filename: - raise HTTPException(400, "Filename is required") # noqa: TRY301 - - try: - validate_filename_extension(file.filename) - except ValueError as e: - raise HTTPException(400, str(e)) from e - - # Validate content type (if provided) - if file.content_type and file.content_type not in { - "application/gzip", - "application/x-gzip", - "application/octet-stream", - }: - raise HTTPException( # noqa: TRY301 - 400, - f"Invalid content type: {file.content_type}", - ) - - # Validate gzip magic bytes - try: - validate_gzip_magic_bytes(file.file) - except ValueError as e: - raise HTTPException(400, str(e)) from e - - # Check file size - file.file.seek(0, 2) # Seek to end - file_size = file.file.tell() - file.file.seek(0) # Reset - - try: - validate_file_size(file_size) - except ValueError as e: - raise HTTPException(400, str(e)) from e - - # Save to temporary file - _logger.info("Saving uploaded file %s (%d bytes)", file.filename, file_size) - temp_path = await save_upload_file_temp(file) - - # Validate VCF format and required fields - try: - validate_vcf_header(temp_path) - except ValueError as e: - raise HTTPException( - 422, - f"VCF validation failed: {e!s}", - ) from e - - # Process VCF - anyvar_client = request.app.state.anyvar_client - anyvlm_storage = request.app.state.anyvlm_storage - _logger.info("Starting VCF ingestion for %s", file.filename) - - try: - ingest_vcf_function(temp_path, anyvar_client, anyvlm_storage, assembly) - except VcfAfColumnsError as e: - _logger.exception("VCF missing required INFO columns") - raise HTTPException(422, f"VCF validation failed: {e}") from e - except Exception as e: - _logger.exception("VCF ingestion failed") - raise HTTPException(500, f"Ingestion failed: {e}") from e - - _logger.info("Successfully ingested VCF: %s", file.filename) - return VcfIngestionResponse( - status="success", - message=f"Successfully ingested {file.filename}", - ) - - except HTTPException: - # Re-raise HTTP exceptions - raise - except Exception as e: - _logger.exception("Unexpected error during VCF upload") - raise HTTPException(500, f"Upload failed: {e}") from e - finally: - # Always cleanup temporary file - if temp_path and temp_path.exists(): - _logger.debug("Cleaning up temporary file: %s", temp_path) - temp_path.unlink() - - _allele_counts_description = """Search for a SNP and receive allele counts by zygosity, in accordance with the Variant-Level Matching protocol. * Unrecognized variants will return a `200 OK` response with a `resultsCount` of 0 diff --git a/src/anyvlm/utils/exceptions.py b/src/anyvlm/utils/exceptions.py index 5ee08ae..9d0152a 100644 --- a/src/anyvlm/utils/exceptions.py +++ b/src/anyvlm/utils/exceptions.py @@ -13,5 +13,9 @@ class UnexpectedVariantTypeError(Exception): """Raised when .type is not of the type expected by AnyVLM""" +class VcfIngestionError(Exception): + """Raised VCF ingestion is unsuccessful""" + + class VariantLookupError(Exception): """Raised when a variant cannot be retrieved from AnyVar""" diff --git a/src/anyvlm/utils/functions.py b/src/anyvlm/utils/functions.py index a2614f8..71e753f 100644 --- a/src/anyvlm/utils/functions.py +++ b/src/anyvlm/utils/functions.py @@ -2,9 +2,9 @@ from typing import cast -from anyvar.core.objects import SupportedVrsObject from ga4gh.vrs.models import Allele +from anyvlm.anyvar.types import SupportedVrsObject from anyvlm.utils.exceptions import ( IncompleteVariantError, UnexpectedVariantTypeError, diff --git a/tests/unit/functions/test_ingest_vcf.py b/tests/unit/functions/test_ingest_vcf.py index dae30f2..66b4d9d 100644 --- a/tests/unit/functions/test_ingest_vcf.py +++ b/tests/unit/functions/test_ingest_vcf.py @@ -1,12 +1,13 @@ from collections.abc import Iterable, Sequence from pathlib import Path +from unittest.mock import MagicMock import pytest -from anyvar.core.objects import SupportedVrsVariation from anyvar.mapping.liftover import ReferenceAssembly from ga4gh.vrs.models import Allele from anyvlm.anyvar.base_client import BaseAnyVarClient +from anyvlm.anyvar.types import SupportedVrsVariation from anyvlm.functions.ingest_vcf import VcfAfColumnsError, ingest_vcf from anyvlm.storage.base_storage import Storage @@ -140,3 +141,27 @@ def test_ingest_vcf_an_zero( stub_anyvar_client, postgres_storage, ) + + +def test_ingest_vcf_malformed_header_gz( + stub_anyvar_client: BaseAnyVarClient, test_data_dir: Path +): + """Test malformed gzipped VCF header is rejected during open.""" + with pytest.raises(ValueError, match="Not a valid VCF file"): + ingest_vcf( + test_data_dir / "vcf" / "malformed_header.vcf.gz", + stub_anyvar_client, + MagicMock(spec=Storage), + ) + + +def test_ingest_vcf_not_a_vcf_gz( + stub_anyvar_client: BaseAnyVarClient, test_data_dir: Path +): + """Test gzipped non-VCF content is rejected during open.""" + with pytest.raises(ValueError, match="Not a valid VCF file"): + ingest_vcf( + test_data_dir / "vcf" / "not_a_vcf.txt.gz", + stub_anyvar_client, + MagicMock(spec=Storage), + ) diff --git a/tests/unit/test_vcf_upload_cli_wrapper.py b/tests/unit/test_vcf_upload_cli_wrapper.py new file mode 100644 index 0000000..a092cf4 --- /dev/null +++ b/tests/unit/test_vcf_upload_cli_wrapper.py @@ -0,0 +1,169 @@ +"""Test VCF upload CLI wrapper functionality.""" + +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import click +import pytest +from anyvar.mapping.liftover import ReferenceAssembly +from click.testing import CliRunner + +# Constants for testing +MAX_FILE_SIZE = 5 * 1024 * 1024 * 1024 # 5GB +UPLOAD_CHUNK_SIZE = 1024 * 1024 # 1MB + + +@pytest.fixture +def runner() -> CliRunner: + """Create Click CLI runner.""" + return CliRunner() + + +@pytest.fixture(scope="module") +def test_vcf_dir(test_data_dir: Path) -> Path: + """Path to VCF test data directory.""" + return test_data_dir / "vcf" + + +@pytest.fixture(scope="module") +def valid_vcf_gz(test_vcf_dir: Path) -> Path: + """Path to valid compressed VCF.""" + return test_vcf_dir / "valid_small.vcf.gz" + + +@pytest.fixture(scope="module") +def missing_fields_vcf_gz(test_vcf_dir: Path) -> Path: + """Path to VCF missing required INFO fields.""" + return test_vcf_dir / "missing_info_fields.vcf.gz" + + +@pytest.fixture(scope="module") +def malformed_vcf_gz(test_vcf_dir: Path) -> Path: + """Path to VCF with malformed header.""" + return test_vcf_dir / "malformed_header.vcf.gz" + + +@pytest.fixture(scope="module") +def not_vcf_gz(test_vcf_dir: Path) -> Path: + """Path to gzipped text file (not a VCF).""" + return test_vcf_dir / "not_a_vcf.txt.gz" + + +# ==================== +# CLI Wrapper Integration Tests +# ==================== + + +class TestIngestVcfCliWrapper: + """Test the `ingest_vcf_cli_wrapper` function""" + + @staticmethod + def _invoke_ingest_vcf( + runner: CliRunner, args: list[str], **patches: MagicMock + ) -> click.testing.Result: + from anyvlm.cli import _cli # pyright: ignore[reportPrivateUsage] + + default_config = SimpleNamespace( + anyvar_uri="http://example-anyvar", + storage_uri="postgresql://example-storage", + ) + with ( + patch( + "anyvlm.cli.get_config", + return_value=patches.get("config", default_config), + ), + patch( + "anyvlm.cli.create_anyvar_client", + return_value=patches.get("anyvar_client", MagicMock()), + ), + patch( + "anyvlm.cli.create_anyvlm_storage", + return_value=patches.get("anyvlm_storage", MagicMock()), + ), + ): + return runner.invoke(_cli, ["ingest-vcf", *args]) + + def test_missing_file_parameter(self, runner: CliRunner): + """Test CLI invocation without file parameter.""" + result = self._invoke_ingest_vcf(runner, ["--assembly", "GRCh38"]) + + assert result.exit_code == 2 + assert "Missing option '--file'" in result.output + + def test_missing_assembly_parameter(self, runner: CliRunner, valid_vcf_gz: Path): + """Test CLI invocation without assembly parameter.""" + result = self._invoke_ingest_vcf(runner, ["--file", str(valid_vcf_gz)]) + + assert result.exit_code == 2 + assert "Missing option '--assembly'" in result.output + + def test_invalid_assembly_value(self, runner: CliRunner, valid_vcf_gz: Path): + """Test CLI invocation with invalid assembly value.""" + result = self._invoke_ingest_vcf( + runner, ["--file", str(valid_vcf_gz), "--assembly", "GRCh99"] + ) + + assert result.exit_code == 2 + assert "Invalid value for '--assembly'" in result.output + + def test_not_a_vcf_file(self, runner: CliRunner, not_vcf_gz: Path): + """Test ingestion fails for gzipped content that is not a VCF.""" + with runner.isolated_filesystem(): + renamed_not_vcf = Path("not_a_vcf.vcf.gz") + renamed_not_vcf.write_bytes(not_vcf_gz.read_bytes()) + + result = self._invoke_ingest_vcf( + runner, ["--file", str(renamed_not_vcf), "--assembly", "GRCh38"] + ) + + assert result.exit_code == 1 + # assert isinstance(result.exception, VcfIngestionError) + # assert "Not a valid VCF file" in str(result.exception) + + def test_vcf_missing_required_fields( + self, runner: CliRunner, missing_fields_vcf_gz: Path + ): + """Test ingestion fails for VCF missing required INFO fields.""" + result = self._invoke_ingest_vcf( + runner, ["--file", str(missing_fields_vcf_gz), "--assembly", "GRCh38"] + ) + + assert result.exit_code == 1 + # assert isinstance(result.exception, VcfIngestionError) + assert "required INFO fields" in str(result.exception) + + @patch("anyvlm.cli.ingest_vcf_function") + def test_successful_upload_and_ingestion( + self, mock_ingest: MagicMock, runner: CliRunner, valid_vcf_gz: Path + ): + """Test successful VCF ingestion via the CLI wrapper.""" + mock_ingest.return_value = None + + result = self._invoke_ingest_vcf( + runner, ["--file", str(valid_vcf_gz), "--assembly", "GRCh38"] + ) + + assert mock_ingest.called + call_args = mock_ingest.call_args + + assert isinstance(call_args.kwargs["vcf_path"], Path) + assert call_args.kwargs["av"] is not None + assert call_args.kwargs["storage"] is not None + assert call_args.kwargs["assembly"] == ReferenceAssembly.GRCH38 + + assert result.exit_code == 0 + assert "Ingestion complete" in result.output + + def test_assembly_grch37_parameter(self, runner: CliRunner, valid_vcf_gz: Path): + """Test that GRCh37 assembly parameter is accepted and used.""" + with patch("anyvlm.cli.ingest_vcf_function") as mock_ingest: + mock_ingest.return_value = None + + result = self._invoke_ingest_vcf( + runner, ["--file", str(valid_vcf_gz), "--assembly", "GRCh37"] + ) + + assert result.exit_code == 0 + call_args = mock_ingest.call_args + assert call_args.kwargs["assembly"] == ReferenceAssembly.GRCH37 diff --git a/tests/unit/test_vcf_upload_endpoint.py b/tests/unit/test_vcf_upload_endpoint.py deleted file mode 100644 index 03ca9f0..0000000 --- a/tests/unit/test_vcf_upload_endpoint.py +++ /dev/null @@ -1,404 +0,0 @@ -"""Test VCF upload endpoint functionality.""" - -import io -from pathlib import Path -from unittest.mock import MagicMock, patch - -import pytest -from anyvar.mapping.liftover import ReferenceAssembly -from fastapi.testclient import TestClient - -from anyvlm.functions.ingest_vcf import VcfAfColumnsError -from anyvlm.main import app - -# Constants for testing -MAX_FILE_SIZE = 5 * 1024 * 1024 * 1024 # 5GB -UPLOAD_CHUNK_SIZE = 1024 * 1024 # 1MB -ENDPOINT = "/anyvlm/ingest_vcf" - - -@pytest.fixture(scope="module") -def client(): - """Create FastAPI test client with mock anyvar_client.""" - # Set up mock anyvar client on app state - mock_anyvar_client = MagicMock() - mock_anyvlm_storage = MagicMock() - app.state.anyvar_client = mock_anyvar_client - app.state.anyvlm_storage = mock_anyvlm_storage - return TestClient(app=app) - - -@pytest.fixture(scope="module") -def test_vcf_dir(test_data_dir: Path) -> Path: - """Path to VCF test data directory.""" - return test_data_dir / "vcf" - - -@pytest.fixture(scope="module") -def valid_vcf_gz(test_vcf_dir: Path) -> Path: - """Path to valid compressed VCF.""" - return test_vcf_dir / "valid_small.vcf.gz" - - -@pytest.fixture(scope="module") -def missing_fields_vcf_gz(test_vcf_dir: Path) -> Path: - """Path to VCF missing required INFO fields.""" - return test_vcf_dir / "missing_info_fields.vcf.gz" - - -@pytest.fixture(scope="module") -def malformed_vcf_gz(test_vcf_dir: Path) -> Path: - """Path to VCF with malformed header.""" - return test_vcf_dir / "malformed_header.vcf.gz" - - -@pytest.fixture(scope="module") -def not_vcf_gz(test_vcf_dir: Path) -> Path: - """Path to gzipped text file (not a VCF).""" - return test_vcf_dir / "not_a_vcf.txt.gz" - - -# ==================== -# Validation Helper Tests -# ==================== - - -class TestFileValidation: - """Test file validation functions.""" - - def test_validate_filename_extension_valid(self): - """Test that .vcf.gz extension passes validation.""" - from anyvlm.restapi.vlm import validate_filename_extension - - # Should not raise - validate_filename_extension("test.vcf.gz") - validate_filename_extension("path/to/file.vcf.gz") - - def test_validate_filename_extension_invalid(self): - """Test that non-.vcf.gz extensions fail validation.""" - from anyvlm.restapi.vlm import validate_filename_extension - - with pytest.raises(ValueError, match="Only .vcf.gz files"): - validate_filename_extension("test.vcf") - - with pytest.raises(ValueError, match="Only .vcf.gz files"): - validate_filename_extension("test.gz") - - with pytest.raises(ValueError, match="Only .vcf.gz files"): - validate_filename_extension("test.txt.gz") - - def test_validate_gzip_magic_bytes_valid(self, valid_vcf_gz: Path): - """Test gzip magic bytes validation with valid file.""" - from anyvlm.restapi.vlm import validate_gzip_magic_bytes - - with valid_vcf_gz.open("rb") as f: - content = f.read() - file_obj = io.BytesIO(content) - validate_gzip_magic_bytes(file_obj) - # Verify file pointer was reset - assert file_obj.tell() == 0 - - def test_validate_gzip_magic_bytes_invalid(self): - """Test gzip magic bytes validation with invalid file.""" - from anyvlm.restapi.vlm import validate_gzip_magic_bytes - - # Non-gzip content - file_obj = io.BytesIO(b"Not a gzip file") - with pytest.raises(ValueError, match="not a valid gzip file"): - validate_gzip_magic_bytes(file_obj) - - def test_validate_file_size_within_limit(self, valid_vcf_gz: Path): - """Test file size validation for file within limit.""" - from anyvlm.restapi.vlm import validate_file_size - - file_size = valid_vcf_gz.stat().st_size - assert file_size < MAX_FILE_SIZE # Sanity check - - # Should not raise - validate_file_size(file_size) - - def test_validate_file_size_exceeds_limit(self): - """Test file size validation for file exceeding limit.""" - from anyvlm.restapi.vlm import validate_file_size - - too_large = MAX_FILE_SIZE + 1 - with pytest.raises(ValueError, match="File too large"): - validate_file_size(too_large) - - def test_validate_vcf_header_valid(self, valid_vcf_gz: Path): - """Test VCF header validation with valid file.""" - from anyvlm.restapi.vlm import validate_vcf_header - - # Should not raise - validate_vcf_header(valid_vcf_gz) - - def test_validate_vcf_header_missing_format_declaration( - self, malformed_vcf_gz: Path - ): - """Test VCF header validation fails on missing fileformat.""" - from anyvlm.restapi.vlm import validate_vcf_header - - with pytest.raises(ValueError, match="Not a valid VCF"): - validate_vcf_header(malformed_vcf_gz) - - def test_validate_vcf_header_missing_required_fields( - self, missing_fields_vcf_gz: Path - ): - """Test VCF header validation fails on missing INFO fields.""" - from anyvlm.restapi.vlm import validate_vcf_header - - with pytest.raises(ValueError, match="VCF missing required INFO fields.*AN"): - validate_vcf_header(missing_fields_vcf_gz) - - -# ==================== -# Endpoint Integration Tests -# ==================== - - -class TestIngestVcfEndpoint: - """Test the /ingest_vcf HTTP endpoint.""" - - def test_endpoint_exists(self, client: TestClient): - """Test that the endpoint exists and accepts POST.""" - response = client.post(ENDPOINT) - # Should not be 404 - assert response.status_code != 404 - - def test_missing_file_parameter(self, client: TestClient): - """Test request without file parameter.""" - response = client.post( - ENDPOINT, - params={"assembly": "GRCh38"}, - ) - assert response.status_code == 422 # Unprocessable Entity - assert "file" in response.text.lower() or "required" in response.text.lower() - - def test_missing_assembly_parameter(self, client: TestClient, valid_vcf_gz: Path): - """Test request without assembly parameter.""" - with valid_vcf_gz.open("rb") as f: - files = {"file": ("test.vcf.gz", f, "application/gzip")} - response = client.post(ENDPOINT, files=files) - - assert response.status_code == 422 - assert ( - "assembly" in response.text.lower() or "required" in response.text.lower() - ) - - def test_invalid_assembly_value(self, client: TestClient, valid_vcf_gz: Path): - """Test request with invalid assembly value.""" - with valid_vcf_gz.open("rb") as f: - files = {"file": ("test.vcf.gz", f, "application/gzip")} - response = client.post( - ENDPOINT, - params={"assembly": "GRCh99"}, # Invalid - files=files, - ) - - assert response.status_code == 422 - - def test_invalid_file_extension(self, client: TestClient, valid_vcf_gz: Path): - """Test upload with wrong file extension.""" - with valid_vcf_gz.open("rb") as f: - # Use .vcf extension (should be .vcf.gz) - files = {"file": ("test.vcf", f, "application/gzip")} - response = client.post( - ENDPOINT, - params={"assembly": "GRCh38"}, - files=files, - ) - - assert response.status_code == 400 - json_response = response.json() - assert "detail" in json_response - assert ".vcf.gz" in json_response["detail"] - - def test_not_gzipped_file(self, client: TestClient): - """Test upload of non-gzipped content.""" - # Plain text, not gzipped - content = b"This is not gzipped" - files = {"file": ("test.vcf.gz", io.BytesIO(content), "application/gzip")} - - response = client.post( - ENDPOINT, - params={"assembly": "GRCh38"}, - files=files, - ) - - assert response.status_code == 400 - json_response = response.json() - assert "detail" in json_response - assert "gzip" in json_response["detail"].lower() - - def test_not_a_vcf_file(self, client: TestClient, not_vcf_gz: Path): - """Test upload of gzipped file that's not a VCF.""" - with not_vcf_gz.open("rb") as f: - files = {"file": ("test.vcf.gz", f, "application/gzip")} - response = client.post( - ENDPOINT, - params={"assembly": "GRCh38"}, - files=files, - ) - - assert response.status_code == 422 - json_response = response.json() - assert "detail" in json_response - assert "vcf" in json_response["detail"].lower() - - def test_vcf_missing_required_fields( - self, client: TestClient, missing_fields_vcf_gz: Path - ): - """Test upload of VCF missing required INFO fields.""" - with missing_fields_vcf_gz.open("rb") as f: - files = {"file": ("test.vcf.gz", f, "application/gzip")} - response = client.post( - ENDPOINT, - params={"assembly": "GRCh38"}, - files=files, - ) - - assert response.status_code == 422 - json_response = response.json() - assert "detail" in json_response - assert ( - "info" in json_response["detail"].lower() - or "field" in json_response["detail"].lower() - ) - - @patch("anyvlm.restapi.vlm.ingest_vcf_function") - def test_successful_upload_and_ingestion( - self, mock_ingest: MagicMock, client: TestClient, valid_vcf_gz: Path - ): - """Test successful VCF upload and ingestion.""" - # Mock the ingest_vcf function to avoid needing real AnyVar - mock_ingest.return_value = None - - with valid_vcf_gz.open("rb") as f: - files = {"file": ("test.vcf.gz", f, "application/gzip")} - response = client.post( - ENDPOINT, - params={"assembly": "GRCh38"}, - files=files, - ) - - assert response.status_code == 200 - json_response = response.json() - assert json_response["status"] == "success" - assert "message" in json_response - - # Verify ingest_vcf was called - assert mock_ingest.called - call_args = mock_ingest.call_args - - # Check Path argument - assert isinstance(call_args[0][0], Path) - - # Check AnyVar client was passed - assert call_args[0][1] is not None - - # Check AnyVLM storage was passed - assert call_args[0][2] is not None - - # Check assembly (4th positional argument) - assert call_args[0][3] == ReferenceAssembly.GRCH38 - - @patch("anyvlm.restapi.vlm.ingest_vcf_function") - def test_ingestion_failure_propagates( - self, mock_ingest: MagicMock, client: TestClient, valid_vcf_gz: Path - ): - """Test that ingestion errors are properly handled and reported.""" - # Mock ingest_vcf to raise an error - mock_ingest.side_effect = VcfAfColumnsError("Missing AC_Het field") - - with valid_vcf_gz.open("rb") as f: - files = {"file": ("test.vcf.gz", f, "application/gzip")} - response = client.post( - ENDPOINT, - params={"assembly": "GRCh38"}, - files=files, - ) - - assert response.status_code == 422 - json_response = response.json() - assert "detail" in json_response - assert "AC_Het" in json_response["detail"] - - def test_temp_file_cleanup_on_success(self, client: TestClient, valid_vcf_gz: Path): - """Test that temporary files are cleaned up after successful ingestion.""" - with patch("anyvlm.restapi.vlm.ingest_vcf_function") as mock_ingest: - mock_ingest.return_value = None - - with valid_vcf_gz.open("rb") as f: - files = {"file": ("test.vcf.gz", f, "application/gzip")} - response = client.post( - ENDPOINT, - params={"assembly": "GRCh38"}, - files=files, - ) - - assert response.status_code == 200 - - # Verify the temp file path that was passed to ingest_vcf no longer exists - if mock_ingest.called: - temp_path = mock_ingest.call_args[0][0] - assert not temp_path.exists(), "Temporary file should be cleaned up" - - def test_temp_file_cleanup_on_error(self, client: TestClient, valid_vcf_gz: Path): - """Test that temporary files are cleaned up even when ingestion fails.""" - with patch("anyvlm.restapi.vlm.ingest_vcf_function") as mock_ingest: - mock_ingest.side_effect = Exception("Ingestion failed") - - with valid_vcf_gz.open("rb") as f: - files = {"file": ("test.vcf.gz", f, "application/gzip")} - response = client.post( - ENDPOINT, - params={"assembly": "GRCh38"}, - files=files, - ) - - assert response.status_code == 500 - - # Verify cleanup happened - if mock_ingest.called: - temp_path = mock_ingest.call_args[0][0] - assert not temp_path.exists(), ( - "Temporary file should be cleaned up even on error" - ) - - def test_assembly_grch37_parameter(self, client: TestClient, valid_vcf_gz: Path): - """Test that GRCh37 assembly parameter is accepted and used.""" - with patch("anyvlm.restapi.vlm.ingest_vcf_function") as mock_ingest: - mock_ingest.return_value = None - - with valid_vcf_gz.open("rb") as f: - files = {"file": ("test.vcf.gz", f, "application/gzip")} - response = client.post( - ENDPOINT, - params={"assembly": "GRCh37"}, - files=files, - ) - - assert response.status_code == 200 - - # Verify GRCh37 was passed (4th positional argument) - call_args = mock_ingest.call_args - assert call_args[0][3] == ReferenceAssembly.GRCH37 - - -# ==================== -# File Size Limit Tests -# ==================== - - -class TestFileSizeLimits: - """Test file size limit enforcement.""" - - def test_file_size_check_with_mock_large_file(self): - """Test that files exceeding size limit are rejected.""" - - # We'll need to test this at the validation function level - # since mocking the actual upload size is complex - from anyvlm.restapi.vlm import validate_file_size - - with pytest.raises(ValueError, match="File too large"): - validate_file_size(MAX_FILE_SIZE + 1)