From 164d7a91bd9164e000490a6db4d94815ffc2db3d Mon Sep 17 00:00:00 2001 From: Sebastian Diaz Date: Tue, 9 Jan 2024 17:20:54 +0100 Subject: [PATCH 1/4] Move timestamp and emultiplex fixtures to plugin modules to clean conftest --- tests/conftest.py | 1492 +++-------------- tests/fixture_plugins/__init__.py | 0 .../demultiplex_fixtures/__init__.py | 0 .../flow_cell_fixtures.py | 172 ++ .../demultiplex_fixtures/name_fixtures.py | 116 ++ .../demultiplex_fixtures/path_fixtures.py | 456 +++++ .../run_parameters_fixtures.py | 98 ++ .../demultiplex_fixtures/sample_fixtures.py | 135 ++ tests/fixture_plugins/timestamp_fixtures.py | 46 + 9 files changed, 1281 insertions(+), 1234 deletions(-) create mode 100644 tests/fixture_plugins/__init__.py create mode 100644 tests/fixture_plugins/demultiplex_fixtures/__init__.py create mode 100644 tests/fixture_plugins/demultiplex_fixtures/flow_cell_fixtures.py create mode 100644 tests/fixture_plugins/demultiplex_fixtures/name_fixtures.py create mode 100644 tests/fixture_plugins/demultiplex_fixtures/path_fixtures.py create mode 100644 tests/fixture_plugins/demultiplex_fixtures/run_parameters_fixtures.py create mode 100644 tests/fixture_plugins/demultiplex_fixtures/sample_fixtures.py create mode 100644 tests/fixture_plugins/timestamp_fixtures.py diff --git a/tests/conftest.py b/tests/conftest.py index 7031d7c179..6ab4439090 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -5,7 +5,7 @@ import os import shutil from copy import deepcopy -from datetime import MAXYEAR, datetime, timedelta +from datetime import datetime from pathlib import Path from subprocess import CompletedProcess from typing import Any, Generator @@ -15,27 +15,21 @@ from requests import Response from cg.apps.demultiplex.demultiplex_api import DemultiplexingAPI -from cg.apps.demultiplex.sample_sheet.sample_models import ( - FlowCellSampleBcl2Fastq, - FlowCellSampleBCLConvert, -) -from cg.apps.demultiplex.sample_sheet.sample_sheet_creator import SampleSheetCreatorBCLConvert from cg.apps.downsample.downsample import DownsampleAPI from cg.apps.gens import GensAPI from cg.apps.gt import GenotypeAPI from cg.apps.hermes.hermes_api import HermesApi from cg.apps.housekeeper.hk import HousekeeperAPI -from cg.apps.lims.api import LimsAPI +from cg.apps.lims import LimsAPI from cg.apps.slurm.slurm_api import SlurmAPI from cg.constants import FileExtensions, Pipeline, SequencingFileTag from cg.constants.constants import CaseActions, FileFormat, Strandedness -from cg.constants.demultiplexing import BclConverter, DemultiplexingDirsAndFiles +from cg.constants.demultiplexing import DemultiplexingDirsAndFiles from cg.constants.housekeeper_tags import HK_DELIVERY_REPORT_TAG -from cg.constants.nanopore_files import NanoporeDirsAndFiles from cg.constants.priority import SlurmQos from cg.constants.sequencing import SequencingPlatform from cg.constants.subject import Sex -from cg.io.controller import ReadFile, WriteFile +from cg.io.controller import WriteFile from cg.io.json import read_json, write_json from cg.io.yaml import write_yaml from cg.meta.encryption.encryption import FlowCellEncryptionAPI @@ -47,11 +41,6 @@ from cg.meta.workflow.taxprofiler import TaxprofilerAnalysisAPI from cg.models import CompressionData from cg.models.cg_config import CGConfig, PDCArchivingDirectory -from cg.models.demultiplex.run_parameters import ( - RunParametersHiSeq, - RunParametersNovaSeq6000, - RunParametersNovaSeqX, -) from cg.models.downsample.downsample_data import DownsampleData from cg.models.flow_cell.flow_cell import FlowCellDirectoryData from cg.models.rnafusion.rnafusion import RnafusionParameters @@ -73,51 +62,14 @@ LOG = logging.getLogger(__name__) - -# Timestamp fixture - - -@pytest.fixture(scope="session") -def old_timestamp() -> datetime: - """Return a time stamp in date time format.""" - return datetime(1900, 1, 1) - - -@pytest.fixture(scope="session") -def timestamp() -> datetime: - """Return a time stamp in date time format.""" - return datetime(2020, 5, 1) - - -@pytest.fixture(scope="session") -def later_timestamp() -> datetime: - """Return a time stamp in date time format.""" - return datetime(2020, 6, 1) - - -@pytest.fixture(scope="session") -def future_date() -> datetime: - """Return a distant date in the future for which no events happen later.""" - return datetime(MAXYEAR, 1, 1, 1, 1, 1) - - -@pytest.fixture(scope="session") -def timestamp_now() -> datetime: - """Return a time stamp of today's date in date time format.""" - return datetime.now() - - -@pytest.fixture(scope="session") -def timestamp_yesterday(timestamp_now: datetime) -> datetime: - """Return a time stamp of yesterday's date in date time format.""" - return timestamp_now - timedelta(days=1) - - -@pytest.fixture(scope="session") -def timestamp_in_2_weeks(timestamp_now: datetime) -> datetime: - """Return a time stamp 14 days ahead in time.""" - return timestamp_now + timedelta(days=14) - +pytest_plugins = [ + "tests.fixture_plugins.timestamp_fixtures", + "tests.fixture_plugins.demultiplex_fixtures.flow_cell_fixtures", + "tests.fixture_plugins.demultiplex_fixtures.name_fixtures", + "tests.fixture_plugins.demultiplex_fixtures.path_fixtures", + "tests.fixture_plugins.demultiplex_fixtures.run_parameters_fixtures", + "tests.fixture_plugins.demultiplex_fixtures.sample_fixtures", +] # Case fixtures @@ -381,6 +333,59 @@ def crunchy_config() -> dict[str, dict[str, Any]]: } +@pytest.fixture +def demultiplexing_context_for_demux( + demultiplexing_api_for_demux: DemultiplexingAPI, + cg_context: CGConfig, + store_with_demultiplexed_samples: Store, +) -> CGConfig: + """Return cg context with a demultiplex context.""" + cg_context.demultiplex_api_ = demultiplexing_api_for_demux + cg_context.housekeeper_api_ = demultiplexing_api_for_demux.hk_api + cg_context.status_db_ = store_with_demultiplexed_samples + return cg_context + + +@pytest.fixture(name="demultiplex_context") +def demultiplex_context( + demultiplexing_api: DemultiplexingAPI, + real_housekeeper_api: HousekeeperAPI, + cg_context: CGConfig, + store_with_demultiplexed_samples: Store, +) -> CGConfig: + """Return cg context with a demultiplex context.""" + cg_context.demultiplex_api_ = demultiplexing_api + cg_context.housekeeper_api_ = real_housekeeper_api + cg_context.status_db_ = store_with_demultiplexed_samples + return cg_context + + +@pytest.fixture(name="demultiplex_configs_for_demux") +def demultiplex_configs_for_demux( + tmp_flow_cells_demux_all_directory: Path, + tmp_empty_demultiplexed_runs_directory: Path, +) -> dict: + """Return demultiplex configs.""" + return { + "flow_cells_dir": tmp_flow_cells_demux_all_directory.as_posix(), + "demultiplexed_flow_cells_dir": tmp_empty_demultiplexed_runs_directory.as_posix(), + "demultiplex": {"slurm": {"account": "test", "mail_user": "testuser@github.se"}}, + } + + +@pytest.fixture(name="demultiplex_configs") +def demultiplex_configs( + tmp_flow_cells_directory: Path, + tmp_demultiplexed_runs_directory: Path, +) -> dict: + """Return demultiplex configs.""" + return { + "flow_cells_dir": tmp_flow_cells_directory.as_posix(), + "demultiplexed_flow_cells_dir": tmp_demultiplexed_runs_directory.as_posix(), + "demultiplex": {"slurm": {"account": "test", "mail_user": "testuser@github.se"}}, + } + + @pytest.fixture def hk_config_dict(root_path: Path): """Housekeeper configs.""" @@ -415,9 +420,46 @@ def gens_config() -> dict[str, dict[str, str]]: } +@pytest.fixture(name="sample_sheet_context") +def sample_sheet_context( + cg_context: CGConfig, lims_api: LimsAPI, populated_housekeeper_api: HousekeeperAPI +) -> CGConfig: + """Return cg context with added Lims and Housekeeper API.""" + cg_context.lims_api_ = lims_api + cg_context.housekeeper_api_ = populated_housekeeper_api + return cg_context + + # Api fixtures +@pytest.fixture(name="demultiplexing_api_for_demux") +def demultiplexing_api_for_demux( + demultiplex_configs_for_demux: dict, + sbatch_process: Process, + populated_housekeeper_api: HousekeeperAPI, +) -> DemultiplexingAPI: + """Return demultiplex API.""" + demux_api = DemultiplexingAPI( + config=demultiplex_configs_for_demux, + housekeeper_api=populated_housekeeper_api, + ) + demux_api.slurm_api.process = sbatch_process + return demux_api + + +@pytest.fixture +def demultiplexing_api( + demultiplex_configs: dict, sbatch_process: Process, populated_housekeeper_api: HousekeeperAPI +) -> DemultiplexingAPI: + """Return demultiplex API.""" + demux_api = DemultiplexingAPI( + config=demultiplex_configs, housekeeper_api=populated_housekeeper_api + ) + demux_api.slurm_api.process = sbatch_process + return demux_api + + @pytest.fixture def rsync_api(cg_context: CGConfig) -> RsyncAPI: """RsyncAPI fixture.""" @@ -774,1212 +816,159 @@ def compression_object(fastq_stub: Path, original_fastq_data: CompressionData) - return working_files -# Demultiplex fixtures +# Genotype file fixture -@pytest.fixture -def lims_novaseq_bcl_convert_samples( - lims_novaseq_samples_raw: list[dict], -) -> list[FlowCellSampleBCLConvert]: - """Return a list of parsed flow cell samples demultiplexed with BCL convert.""" - return [FlowCellSampleBCLConvert.model_validate(sample) for sample in lims_novaseq_samples_raw] +@pytest.fixture(name="bcf_file") +def bcf_file(apps_dir: Path) -> Path: + """Return the path to a BCF file.""" + return Path(apps_dir, "gt", "yellowhog.bcf") -@pytest.fixture -def lims_novaseq_bcl2fastq_samples( - lims_novaseq_samples_raw: list[dict], -) -> list[FlowCellSampleBcl2Fastq]: - """Return a list of parsed Bcl2fastq flow cell samples""" - return [FlowCellSampleBcl2Fastq.model_validate(sample) for sample in lims_novaseq_samples_raw] +# Gens file fixtures -@pytest.fixture -def lims_novaseq_6000_bcl2fastq_samples( - lims_novaseq_6000_sample_raw: list[dict], -) -> list[FlowCellSampleBcl2Fastq]: - """Return a list of parsed Bcl2fastq flow cell samples""" - return [ - FlowCellSampleBcl2Fastq.model_validate(sample) for sample in lims_novaseq_6000_sample_raw - ] +@pytest.fixture(name="gens_fracsnp_path") +def gens_fracsnp_path(mip_dna_analysis_dir: Path, sample_id: str) -> Path: + """Path to Gens fracsnp/baf bed file.""" + return Path(mip_dna_analysis_dir, f"{sample_id}.baf.bed.gz") -@pytest.fixture(name="tmp_flow_cells_directory") -def tmp_flow_cells_directory(tmp_path: Path, flow_cells_dir: Path) -> Path: - """ - Return the path to a temporary flow cells directory with flow cells ready for demultiplexing. - Generates a copy of the original flow cells directory - """ - original_dir = flow_cells_dir - tmp_dir = Path(tmp_path, "flow_cells") +@pytest.fixture(name="gens_coverage_path") +def gens_coverage_path(mip_dna_analysis_dir: Path, sample_id: str) -> Path: + """Path to Gens coverage bed file.""" + return Path(mip_dna_analysis_dir, f"{sample_id}.cov.bed.gz") - return Path(shutil.copytree(original_dir, tmp_dir)) +# Housekeeper, Chanjo file fixtures -@pytest.fixture(name="tmp_flow_cells_demux_all_directory") -def tmp_flow_cells_demux_all_directory(tmp_path: Path, flow_cells_demux_all_dir: Path) -> Path: - """ - Return the path to a temporary flow cells directory with flow cells ready for demultiplexing. - Generates a copy of the original flow cells directory. - This fixture is used for testing of the cg demutliplex all cmd. - """ - original_dir = flow_cells_demux_all_dir - tmp_dir = Path(tmp_path, "flow_cells_demux_all") - return Path(shutil.copytree(original_dir, tmp_dir)) +@pytest.fixture(name="bed_file") +def bed_file(analysis_dir) -> Path: + """Return the path to a bed file.""" + return Path(analysis_dir, "sample_coverage.bed") -@pytest.fixture(name="tmp_flow_cell_directory_bcl2fastq") -def flow_cell_working_directory_bcl2fastq( - bcl2fastq_flow_cell_dir: Path, tmp_flow_cells_directory: Path -) -> Path: - """Return the path to a working directory that will be deleted after test is run. +# Helper fixtures - This is a path to a flow cell directory with the run parameters present. - """ - return Path(tmp_flow_cells_directory, bcl2fastq_flow_cell_dir.name) +@pytest.fixture(scope="session") +def helpers() -> StoreHelpers: + """Return a class with helper functions for the stores.""" + return StoreHelpers() -@pytest.fixture(name="tmp_flow_cell_directory_bclconvert") -def flow_cell_working_directory_bclconvert( - bcl_convert_flow_cell_dir: Path, tmp_flow_cells_directory: Path -) -> Path: - """Return the path to a working directory that will be deleted after test is run. - This is a path to a flow cell directory with the run parameters present. - """ - return Path(tmp_flow_cells_directory, bcl_convert_flow_cell_dir.name) +@pytest.fixture(name="small_helpers") +def small_helpers() -> SmallHelpers: + """Return a class with small helper functions.""" + return SmallHelpers() -@pytest.fixture -def tmp_flow_cell_name_no_run_parameters() -> str: - """This is the name of a flow cell directory with the run parameters missing.""" - return "180522_A00689_0200_BHLCKNCCXY" +# HK fixtures -@pytest.fixture -def tmp_flow_cell_name_malformed_sample_sheet() -> str: - """ "Returns the name of a flow cell directory ready for demultiplexing with BCL convert. - Contains a sample sheet with malformed headers. - """ - return "201203_A00689_0200_AHVKJCDRXY" +@pytest.fixture(name="root_path") +def root_path(project_dir: Path) -> Path: + """Return the path to a hk bundles dir.""" + _root_path = Path(project_dir, "bundles") + _root_path.mkdir(parents=True, exist_ok=True) + return _root_path -@pytest.fixture -def tmp_flow_cell_name_no_sample_sheet() -> str: - """Return the name of a flow cell directory with the run parameters and sample sheet missing.""" - return "170407_A00689_0209_BHHKVCALXX" +@pytest.fixture(name="hk_bundle_sample_path") +def hk_bundle_sample_path(sample_id: str, timestamp: datetime) -> Path: + """Return the relative path to a Housekeeper bundle mock sample.""" + return Path(sample_id, timestamp.strftime("%Y-%m-%d")) -@pytest.fixture(name="tmp_flow_cell_name_ready_for_demultiplexing_bcl2fastq") -def tmp_flow_cell_name_ready_for_demultiplexing_bcl2fastq() -> str: - """Returns the name of a flow cell directory ready for demultiplexing with bcl2fastq.""" - return "211101_D00483_0615_AHLG5GDRXY" +@pytest.fixture(name="hk_bundle_data") +def hk_bundle_data( + case_id: str, + bed_file: Path, + delivery_report_html: Path, + timestamp_yesterday: datetime, + sample_id: str, + father_sample_id: str, + mother_sample_id: str, +) -> dict[str, Any]: + """Return some bundle data for Housekeeper.""" + return { + "name": case_id, + "created": timestamp_yesterday, + "expires": timestamp_yesterday, + "files": [ + { + "path": bed_file.as_posix(), + "archive": False, + "tags": ["bed", sample_id, father_sample_id, mother_sample_id, "coverage"], + }, + { + "path": delivery_report_html.as_posix(), + "archive": False, + "tags": [HK_DELIVERY_REPORT_TAG], + }, + ], + } -@pytest.fixture -def tmp_flow_cells_directory_no_run_parameters( - tmp_flow_cell_name_no_run_parameters: str, tmp_flow_cells_directory: Path -) -> Path: - """This is a path to a flow cell directory with the run parameters missing.""" - return Path(tmp_flow_cells_directory, tmp_flow_cell_name_no_run_parameters) +@pytest.fixture(name="hk_sample_bundle") +def hk_sample_bundle( + fastq_file: Path, + sample_hk_bundle_no_files: dict, + sample_id: str, + spring_file: Path, +) -> dict: + """Returns a dict for building a housekeeper bundle for a sample.""" + sample_hk_bundle_no_files["files"] = [ + { + "path": spring_file.as_posix(), + "archive": False, + "tags": [SequencingFileTag.SPRING, sample_id], + }, + { + "path": fastq_file.as_posix(), + "archive": False, + "tags": [SequencingFileTag.FASTQ, sample_id], + }, + ] + return sample_hk_bundle_no_files -@pytest.fixture(name="tmp_flow_cells_directory_no_sample_sheet") -def tmp_flow_cells_directory_no_sample_sheet( - tmp_flow_cell_name_no_sample_sheet: str, tmp_flow_cells_directory: Path -) -> Path: - """This is a path to a flow cell directory with the sample sheet and run parameters missing.""" - return Path(tmp_flow_cells_directory, tmp_flow_cell_name_no_sample_sheet) +@pytest.fixture(name="hk_father_sample_bundle") +def hk_father_sample_bundle( + fastq_file_father: Path, + helpers, + sample_hk_bundle_no_files: dict, + father_sample_id: str, + spring_file_father: Path, +) -> dict: + """Returns a dict for building a housekeeper bundle for a second sample.""" + father_sample_bundle = deepcopy(sample_hk_bundle_no_files) + father_sample_bundle["name"] = father_sample_id + father_sample_bundle["files"] = [ + { + "path": spring_file_father.as_posix(), + "archive": False, + "tags": [SequencingFileTag.SPRING, father_sample_id], + }, + { + "path": fastq_file_father.as_posix(), + "archive": False, + "tags": [SequencingFileTag.FASTQ, father_sample_id], + }, + ] + return father_sample_bundle -@pytest.fixture -def tmp_flow_cells_directory_malformed_sample_sheet( - tmp_flow_cell_name_malformed_sample_sheet: str, tmp_flow_cells_directory: Path -) -> Path: - """This is a path to a flow cell directory with a sample sheet with malformed headers.""" - return Path(tmp_flow_cells_directory, tmp_flow_cell_name_malformed_sample_sheet) - -@pytest.fixture -def tmp_flow_cells_directory_ready_for_demultiplexing_bcl_convert( - bcl_convert_flow_cell_full_name: str, tmp_flow_cells_directory: Path -) -> Path: - """This is a path to a flow cell directory with the run parameters missing.""" - return Path(tmp_flow_cells_directory, bcl_convert_flow_cell_full_name) - - -@pytest.fixture -def tmp_flow_cells_directory_ready_for_demultiplexing_bcl2fastq( - tmp_flow_cell_name_ready_for_demultiplexing_bcl2fastq: str, tmp_flow_cells_directory: Path -) -> Path: - """This is a path to a flow cell directory with the run parameters missing.""" - return Path(tmp_flow_cells_directory, tmp_flow_cell_name_ready_for_demultiplexing_bcl2fastq) - - -# Temporary demultiplexed runs fixtures -@pytest.fixture(name="tmp_demultiplexed_runs_directory") -def tmp_demultiplexed_flow_cells_directory(tmp_path: Path, demultiplexed_runs: Path) -> Path: - """Return the path to a temporary demultiplex-runs directory. - Generates a copy of the original demultiplexed-runs - """ - original_dir = demultiplexed_runs - tmp_dir = Path(tmp_path, "demultiplexed-runs") - return Path(shutil.copytree(original_dir, tmp_dir)) - - -@pytest.fixture(name="tmp_demultiplexed_runs_bcl2fastq_directory") -def tmp_demultiplexed_runs_bcl2fastq_directory( - tmp_demultiplexed_runs_directory: Path, bcl2fastq_flow_cell_dir: Path -) -> Path: - """Return the path to a temporary demultiplex-runs bcl2fastq flow cell directory.""" - return Path(tmp_demultiplexed_runs_directory, bcl2fastq_flow_cell_dir.name) - - -@pytest.fixture(name="tmp_bcl2fastq_flow_cell") -def tmp_bcl2fastq_flow_cell( - tmp_demultiplexed_runs_bcl2fastq_directory: Path, -) -> FlowCellDirectoryData: - """Create a flow cell object with flow cell that is demultiplexed.""" - return FlowCellDirectoryData( - flow_cell_path=tmp_demultiplexed_runs_bcl2fastq_directory, - bcl_converter=BclConverter.BCL2FASTQ, - ) - - -@pytest.fixture -def novaseq6000_flow_cell( - tmp_flow_cells_directory_malformed_sample_sheet: Path, -) -> FlowCellDirectoryData: - """Return a NovaSeq6000 flow cell.""" - return FlowCellDirectoryData( - flow_cell_path=tmp_flow_cells_directory_malformed_sample_sheet, - bcl_converter=BclConverter.BCLCONVERT, - ) - - -@pytest.fixture(name="tmp_bcl_convert_flow_cell") -def tmp_bcl_convert_flow_cell( - tmp_flow_cell_directory_bclconvert: Path, -) -> FlowCellDirectoryData: - """Create a flow cell object with flow cell that is demultiplexed.""" - return FlowCellDirectoryData( - flow_cell_path=tmp_flow_cell_directory_bclconvert, - bcl_converter=BclConverter.DRAGEN, - ) - - -@pytest.fixture(name="tmp_demultiplexed_runs_not_finished_directory") -def tmp_demultiplexed_runs_not_finished_flow_cells_directory( - tmp_path: Path, demux_results_not_finished_dir: Path -) -> Path: - """ - Return a temporary demultiplex-runs-unfinished path with an unfinished flow cell directory. - Generates a copy of the original demultiplexed-runs-unfinished directory. - """ - original_dir = demux_results_not_finished_dir - tmp_dir = Path(tmp_path, "demultiplexed-runs-unfinished") - return Path(shutil.copytree(original_dir, tmp_dir)) - - -@pytest.fixture(name="demultiplexed_runs_unfinished_bcl2fastq_flow_cell_directory") -def demultiplexed_runs_bcl2fastq_flow_cell_directory( - tmp_demultiplexed_runs_not_finished_directory: Path, - bcl2fastq_flow_cell_full_name: str, -) -> Path: - """Copy the content of a demultiplexed but not finished directory to a temporary location.""" - return Path(tmp_demultiplexed_runs_not_finished_directory, bcl2fastq_flow_cell_full_name) - - -@pytest.fixture(name="tmp_unfinished_bcl2fastq_flow_cell") -def unfinished_bcl2fastq_flow_cell( - demultiplexed_runs_unfinished_bcl2fastq_flow_cell_directory: Path, - bcl2fastq_flow_cell_full_name: str, -) -> FlowCellDirectoryData: - """Copy the content of a demultiplexed but not finished directory to a temporary location.""" - return FlowCellDirectoryData( - flow_cell_path=demultiplexed_runs_unfinished_bcl2fastq_flow_cell_directory, - bcl_converter=BclConverter.BCL2FASTQ, - ) - - -@pytest.fixture(name="sample_sheet_context") -def sample_sheet_context( - cg_context: CGConfig, lims_api: LimsAPI, populated_housekeeper_api: HousekeeperAPI -) -> CGConfig: - """Return cg context with added Lims and Housekeeper API.""" - cg_context.lims_api_ = lims_api - cg_context.housekeeper_api_ = populated_housekeeper_api - return cg_context - - -@pytest.fixture -def bcl_convert_sample_sheet_creator( - bcl_convert_flow_cell: FlowCellDirectoryData, - lims_novaseq_bcl_convert_samples: list[FlowCellSampleBCLConvert], -) -> SampleSheetCreatorBCLConvert: - """Returns a sample sheet creator for version 2 sample sheets with dragen format.""" - return SampleSheetCreatorBCLConvert( - flow_cell=bcl_convert_flow_cell, - lims_samples=lims_novaseq_bcl_convert_samples, - ) - - -@pytest.fixture(scope="session") -def bcl_convert_demultiplexed_flow_cell_sample_internal_ids() -> list[str]: - """ - Sample id:s present in sample sheet for dummy flow cell demultiplexed with BCL Convert in - cg/tests/fixtures/apps/demultiplexing/demultiplexed-runs/230504_A00689_0804_BHY7FFDRX2. - """ - return ["ACC11927A2", "ACC11927A5"] - - -@pytest.fixture(scope="session") -def bcl2fastq_demultiplexed_flow_cell_sample_internal_ids() -> list[str]: - """ - Sample id:s present in sample sheet for dummy flow cell demultiplexed with BCL Convert in - cg/tests/fixtures/apps/demultiplexing/demultiplexed-runs/170407_A00689_0209_BHHKVCALXX. - """ - return ["SVE2528A1"] - - -@pytest.fixture(scope="session") -def flow_cell_name_demultiplexed_with_bcl2fastq() -> str: - """Return the name of a flow cell that has been demultiplexed with BCL2Fastq.""" - return "HHKVCALXX" - - -@pytest.fixture(scope="session") -def flow_cell_directory_name_demultiplexed_with_bcl2fastq( - flow_cell_name_demultiplexed_with_bcl2fastq: str, -): - """Return the name of a flow cell directory that has been demultiplexed with BCL2Fastq.""" - return f"170407_ST-E00198_0209_B{flow_cell_name_demultiplexed_with_bcl2fastq}" - - -@pytest.fixture(scope="session") -def flow_cell_name_demultiplexed_with_bcl_convert() -> str: - return "HY7FFDRX2" - - -@pytest.fixture(scope="session") -def flow_cell_directory_name_demultiplexed_with_bcl_convert( - flow_cell_name_demultiplexed_with_bcl_convert: str, -): - return f"230504_A00689_0804_B{flow_cell_name_demultiplexed_with_bcl_convert}" - - -# Fixtures for test demultiplex flow cell -@pytest.fixture -def tmp_empty_demultiplexed_runs_directory(tmp_demultiplexed_runs_directory) -> Path: - return Path(tmp_demultiplexed_runs_directory, "empty") - - -@pytest.fixture -def store_with_demultiplexed_samples( - store: Store, - helpers: StoreHelpers, - bcl_convert_demultiplexed_flow_cell_sample_internal_ids: list[str], - bcl2fastq_demultiplexed_flow_cell_sample_internal_ids: list[str], - flow_cell_name_demultiplexed_with_bcl2fastq: str, - flow_cell_name_demultiplexed_with_bcl_convert: str, -) -> Store: - """Return a store with samples that have been demultiplexed with BCL Convert and BCL2Fastq.""" - helpers.add_flow_cell( - store, flow_cell_name_demultiplexed_with_bcl_convert, sequencer_type="novaseq" - ) - helpers.add_flow_cell( - store, flow_cell_name_demultiplexed_with_bcl2fastq, sequencer_type="hiseqx" - ) - for i, sample_internal_id in enumerate(bcl_convert_demultiplexed_flow_cell_sample_internal_ids): - helpers.add_sample(store, internal_id=sample_internal_id, name=f"sample_bcl_convert_{i}") - helpers.add_sample_lane_sequencing_metrics( - store, - sample_internal_id=sample_internal_id, - flow_cell_name=flow_cell_name_demultiplexed_with_bcl_convert, - ) - - for i, sample_internal_id in enumerate(bcl2fastq_demultiplexed_flow_cell_sample_internal_ids): - helpers.add_sample(store, internal_id=sample_internal_id, name=f"sample_bcl2fastq_{i}") - helpers.add_sample_lane_sequencing_metrics( - store, - sample_internal_id=sample_internal_id, - flow_cell_name=flow_cell_name_demultiplexed_with_bcl2fastq, - ) - return store - - -@pytest.fixture -def demultiplexing_context_for_demux( - demultiplexing_api_for_demux: DemultiplexingAPI, - cg_context: CGConfig, - store_with_demultiplexed_samples: Store, -) -> CGConfig: - """Return cg context with a demultiplex context.""" - cg_context.demultiplex_api_ = demultiplexing_api_for_demux - cg_context.housekeeper_api_ = demultiplexing_api_for_demux.hk_api - cg_context.status_db_ = store_with_demultiplexed_samples - return cg_context - - -@pytest.fixture(name="demultiplex_context") -def demultiplex_context( - demultiplexing_api: DemultiplexingAPI, - real_housekeeper_api: HousekeeperAPI, - cg_context: CGConfig, - store_with_demultiplexed_samples: Store, -) -> CGConfig: - """Return cg context with a demultiplex context.""" - cg_context.demultiplex_api_ = demultiplexing_api - cg_context.housekeeper_api_ = real_housekeeper_api - cg_context.status_db_ = store_with_demultiplexed_samples - return cg_context - - -@pytest.fixture(name="demultiplex_configs_for_demux") -def demultiplex_configs_for_demux( - tmp_flow_cells_demux_all_directory: Path, - tmp_empty_demultiplexed_runs_directory: Path, -) -> dict: - """Return demultiplex configs.""" - return { - "flow_cells_dir": tmp_flow_cells_demux_all_directory.as_posix(), - "demultiplexed_flow_cells_dir": tmp_empty_demultiplexed_runs_directory.as_posix(), - "demultiplex": {"slurm": {"account": "test", "mail_user": "testuser@github.se"}}, - } - - -@pytest.fixture(name="demultiplex_configs") -def demultiplex_configs( - tmp_flow_cells_directory: Path, - tmp_demultiplexed_runs_directory: Path, -) -> dict: - """Return demultiplex configs.""" - return { - "flow_cells_dir": tmp_flow_cells_directory.as_posix(), - "demultiplexed_flow_cells_dir": tmp_demultiplexed_runs_directory.as_posix(), - "demultiplex": {"slurm": {"account": "test", "mail_user": "testuser@github.se"}}, - } - - -@pytest.fixture(name="demultiplexing_api_for_demux") -def demultiplexing_api_for_demux( - demultiplex_configs_for_demux: dict, - sbatch_process: Process, - populated_housekeeper_api: HousekeeperAPI, -) -> DemultiplexingAPI: - """Return demultiplex API.""" - demux_api = DemultiplexingAPI( - config=demultiplex_configs_for_demux, - housekeeper_api=populated_housekeeper_api, - ) - demux_api.slurm_api.process = sbatch_process - return demux_api - - -@pytest.fixture -def demultiplexing_api( - demultiplex_configs: dict, sbatch_process: Process, populated_housekeeper_api: HousekeeperAPI -) -> DemultiplexingAPI: - """Return demultiplex API.""" - demux_api = DemultiplexingAPI( - config=demultiplex_configs, housekeeper_api=populated_housekeeper_api - ) - demux_api.slurm_api.process = sbatch_process - return demux_api - - -@pytest.fixture(name="novaseq6000_bcl_convert_sample_sheet_path") -def novaseq6000_sample_sheet_path() -> Path: - """Return the path to a NovaSeq 6000 BCL convert sample sheet.""" - return Path( - "tests", - "fixtures", - "apps", - "sequencing_metrics_parser", - "230622_A00621_0864_AHY7FFDRX2", - "Unaligned", - "Reports", - "SampleSheet.csv", - ) - - -@pytest.fixture(scope="session") -def demultiplex_fixtures(apps_dir: Path) -> Path: - """Return the path to the demultiplex fixture directory.""" - return Path(apps_dir, "demultiplexing") - - -@pytest.fixture(scope="session") -def raw_lims_sample_dir(demultiplex_fixtures: Path) -> Path: - """Return the path to the raw samples fixture directory.""" - return Path(demultiplex_fixtures, "raw_lims_samples") - - -@pytest.fixture(scope="session") -def run_parameters_dir(demultiplex_fixtures: Path) -> Path: - """Return the path to the run parameters fixture directory.""" - return Path(demultiplex_fixtures, "run_parameters") - - -@pytest.fixture(scope="session") -def demultiplexed_runs(demultiplex_fixtures: Path) -> Path: - """Return the path to the demultiplexed flow cells fixture directory.""" - return Path(demultiplex_fixtures, "demultiplexed-runs") - - -@pytest.fixture(scope="session") -def flow_cells_dir(demultiplex_fixtures: Path) -> Path: - """Return the path to the sequenced flow cells fixture directory.""" - return Path(demultiplex_fixtures, DemultiplexingDirsAndFiles.FLOW_CELLS_DIRECTORY_NAME) - - -@pytest.fixture(scope="session") -def nanopore_flow_cells_dir(demultiplex_fixtures: Path) -> Path: - """Return the path to the sequenced flow cells fixture directory.""" - return Path(demultiplex_fixtures, NanoporeDirsAndFiles.DATA_DIRECTORY) - - -@pytest.fixture(scope="session") -def flow_cells_demux_all_dir(demultiplex_fixtures: Path) -> Path: - """Return the path to the sequenced flow cells fixture directory.""" - return Path(demultiplex_fixtures, "flow_cells_demux_all") - - -@pytest.fixture(scope="session") -def demux_results_not_finished_dir(demultiplex_fixtures: Path) -> Path: - """Return the path to a dir with demultiplexing results where nothing has been cleaned.""" - return Path(demultiplex_fixtures, "demultiplexed-runs-unfinished") - - -@pytest.fixture -def novaseq_6000_post_1_5_kits_flow_cell(tmp_flow_cells_directory: Path) -> Path: - return Path(tmp_flow_cells_directory, "230912_A00187_1009_AHK33MDRX3") - - -@pytest.fixture() -def novaseq_6000_post_1_5_kits_flow_cell_data(flow_cells_dir: Path) -> FlowCellDirectoryData: - return FlowCellDirectoryData(Path(flow_cells_dir, "230912_A00187_1009_AHK33MDRX3")) - - -@pytest.fixture -def novaseq_6000_post_1_5_kits_correct_sample_sheet( - novaseq_6000_post_1_5_kits_flow_cell: Path, -) -> Path: - return Path(novaseq_6000_post_1_5_kits_flow_cell, "CorrectSampleSheet.csv") - - -@pytest.fixture -def novaseq_6000_post_1_5_kits_raw_lims_samples( - novaseq_6000_post_1_5_kits_flow_cell: Path, -) -> Path: - return Path(novaseq_6000_post_1_5_kits_flow_cell, "HK33MDRX3_raw.json") - - -@pytest.fixture -def novaseq_6000_post_1_5_kits_lims_samples( - novaseq_6000_post_1_5_kits_raw_lims_samples: Path, -) -> list[FlowCellSampleBCLConvert]: - return [ - FlowCellSampleBCLConvert.model_validate(sample) - for sample in read_json(novaseq_6000_post_1_5_kits_raw_lims_samples) - ] - - -@pytest.fixture() -def novaseq_6000_pre_1_5_kits_flow_cell_data(flow_cells_dir: Path) -> FlowCellDirectoryData: - return FlowCellDirectoryData(Path(flow_cells_dir, "190927_A00689_0069_BHLYWYDSXX")) - - -@pytest.fixture -def novaseq_6000_pre_1_5_kits_flow_cell(tmp_flow_cells_directory: Path) -> Path: - return Path(tmp_flow_cells_directory, "190927_A00689_0069_BHLYWYDSXX") - - -@pytest.fixture -def novaseq_6000_pre_1_5_kits_correct_sample_sheet( - novaseq_6000_pre_1_5_kits_flow_cell: Path, -) -> Path: - return Path(novaseq_6000_pre_1_5_kits_flow_cell, "CorrectSampleSheet.csv") - - -@pytest.fixture -def novaseq_6000_pre_1_5_kits_raw_lims_samples(novaseq_6000_pre_1_5_kits_flow_cell: Path) -> Path: - return Path(novaseq_6000_pre_1_5_kits_flow_cell, "HLYWYDSXX_raw.json") - - -@pytest.fixture -def novaseq_6000_pre_1_5_kits_lims_samples( - novaseq_6000_pre_1_5_kits_raw_lims_samples: Path, -) -> list[FlowCellSampleBCLConvert]: - return [ - FlowCellSampleBCLConvert.model_validate(sample) - for sample in read_json(novaseq_6000_pre_1_5_kits_raw_lims_samples) - ] - - -@pytest.fixture -def novaseq_x_flow_cell_directory(tmp_flow_cells_directory: Path) -> Path: - return Path(tmp_flow_cells_directory, "20231108_LH00188_0028_B22F52TLT3") - - -@pytest.fixture() -def novaseq_x_flow_cell_data(flow_cells_dir: Path) -> FlowCellDirectoryData: - return FlowCellDirectoryData(Path(flow_cells_dir, "20231108_LH00188_0028_B22F52TLT3")) - - -@pytest.fixture -def novaseq_x_correct_sample_sheet(novaseq_x_flow_cell_directory: Path) -> Path: - return Path(novaseq_x_flow_cell_directory, "CorrectSampleSheet.csv") - - -@pytest.fixture -def novaseq_x_raw_lims_samples(novaseq_x_flow_cell_directory: Path) -> Path: - return Path(novaseq_x_flow_cell_directory, "22F52TLT3_raw.json") - - -@pytest.fixture -def novaseq_x_lims_samples(novaseq_x_raw_lims_samples: Path) -> list[FlowCellSampleBCLConvert]: - return [ - FlowCellSampleBCLConvert.model_validate(sample) - for sample in read_json(novaseq_x_raw_lims_samples) - ] - - -@pytest.fixture(scope="session") -def hiseq_x_single_index_flow_cell_name() -> str: - """Return the full name of a HiSeqX flow cell with only one index.""" - return "170517_ST-E00266_0210_BHJCFFALXX" - - -@pytest.fixture(scope="session") -def hiseq_x_dual_index_flow_cell_name() -> str: - """Return the full name of a HiSeqX flow cell with two indexes.""" - return "180508_ST-E00269_0269_AHL32LCCXY" - - -@pytest.fixture(scope="session") -def hiseq_2500_dual_index_flow_cell_name() -> str: - """Return the full name of a HiSeq2500 flow cell with double indexes.""" - return "181005_D00410_0735_BHM2LNBCX2" - - -@pytest.fixture(scope="session") -def hiseq_2500_custom_index_flow_cell_name() -> str: - """Return the full name of a HiSeq2500 flow cell with double indexes.""" - return "180509_D00450_0598_BHGYFNBCX2" - - -@pytest.fixture(scope="session") -def bcl2fastq_flow_cell_full_name() -> str: - """Return full flow cell name.""" - return "201203_D00483_0200_AHVKJCDRXX" - - -@pytest.fixture(scope="session") -def bcl_convert_flow_cell_full_name() -> str: - """Return the full name of a bcl_convert flow cell.""" - return "211101_A00187_0615_AHLG5GDRZZ" - - -@pytest.fixture(scope="session") -def novaseq_x_flow_cell_full_name() -> str: - """Return the full name of a NovaSeqX flow cell.""" - return "20230508_LH00188_0003_A22522YLT3" - - -@pytest.fixture(scope="session") -def novaseq_x_manifest_file(novaseq_x_flow_cell_dir: Path) -> Path: - """Return the path to a NovaSeqX manifest file.""" - return Path(novaseq_x_flow_cell_dir, "Manifest.tsv") - - -@pytest.fixture(scope="session") -def hiseq_x_single_index_flow_cell_dir( - flow_cells_dir: Path, hiseq_x_single_index_flow_cell_name: str -) -> Path: - """Return the path to a HiSeqX flow cell.""" - return Path(flow_cells_dir, hiseq_x_single_index_flow_cell_name) - - -@pytest.fixture(scope="session") -def hiseq_x_dual_index_flow_cell_dir( - flow_cells_dir: Path, hiseq_x_dual_index_flow_cell_name: str -) -> Path: - """Return the path to a HiSeqX flow cell.""" - return Path(flow_cells_dir, hiseq_x_dual_index_flow_cell_name) - - -@pytest.fixture(scope="session") -def hiseq_2500_dual_index_flow_cell_dir( - flow_cells_dir: Path, hiseq_2500_dual_index_flow_cell_name: str -) -> Path: - """Return the path to a HiSeq2500 flow cell.""" - return Path(flow_cells_dir, hiseq_2500_dual_index_flow_cell_name) - - -@pytest.fixture(scope="session") -def hiseq_2500_custom_index_flow_cell_dir( - flow_cells_dir: Path, hiseq_2500_custom_index_flow_cell_name: str -) -> Path: - """Return the path to a HiSeq2500 flow cell.""" - return Path(flow_cells_dir, hiseq_2500_custom_index_flow_cell_name) - - -@pytest.fixture(scope="session") -def bcl2fastq_flow_cell_dir(flow_cells_dir: Path, bcl2fastq_flow_cell_full_name: str) -> Path: - """Return the path to the bcl2fastq flow cell demultiplex fixture directory.""" - return Path(flow_cells_dir, bcl2fastq_flow_cell_full_name) - - -@pytest.fixture(scope="session") -def bcl_convert_flow_cell_dir(flow_cells_dir: Path, bcl_convert_flow_cell_full_name: str) -> Path: - """Return the path to the bcl_convert flow cell demultiplex fixture directory.""" - return Path(flow_cells_dir, bcl_convert_flow_cell_full_name) - - -@pytest.fixture(scope="session") -def novaseq_x_flow_cell_dir(flow_cells_dir: Path, novaseq_x_flow_cell_full_name: str) -> Path: - """Return the path to the NovaSeqX flow cell demultiplex fixture directory.""" - return Path(flow_cells_dir, novaseq_x_flow_cell_full_name) - - -@pytest.fixture -def hiseq_x_single_index_bcl_convert_lims_samples( - hiseq_x_single_index_flow_cell_dir: Path, -) -> list[FlowCellSampleBCLConvert]: - """Return a list of BCLConvert samples from a HiSeqX single index flow cell.""" - path = Path( - hiseq_x_single_index_flow_cell_dir, f"HJCFFALXX_bcl_convert_raw{FileExtensions.JSON}" - ) - return [FlowCellSampleBCLConvert.model_validate(sample) for sample in read_json(path)] - - -@pytest.fixture -def hiseq_x_dual_index_bcl_convert_lims_samples( - hiseq_x_dual_index_flow_cell_dir: Path, -) -> list[FlowCellSampleBCLConvert]: - """Return a list of BCLConvert samples from a HiSeqX dual index flow cell.""" - path = Path(hiseq_x_dual_index_flow_cell_dir, f"HL32LCCXY_bcl_convert_raw{FileExtensions.JSON}") - return [FlowCellSampleBCLConvert.model_validate(sample) for sample in read_json(path)] - - -@pytest.fixture -def hiseq_2500_dual_index_bcl_convert_lims_samples( - hiseq_2500_dual_index_flow_cell_dir: Path, -) -> list[FlowCellSampleBCLConvert]: - """Return a list of BCLConvert samples from a HiSeq2500 dual index flow cell.""" - path = Path(hiseq_2500_dual_index_flow_cell_dir, "HM2LNBCX2_bcl_convert_raw.json") - return [FlowCellSampleBCLConvert.model_validate(sample) for sample in read_json(path)] - - -@pytest.fixture -def hiseq_2500_custom_index_bcl_convert_lims_samples( - hiseq_2500_custom_index_flow_cell_dir: Path, -) -> list[FlowCellSampleBCLConvert]: - """Return a list of BCLConvert samples from a HiSeq2500 custom index flow cell.""" - path = Path(hiseq_2500_custom_index_flow_cell_dir, "HGYFNBCX2_bcl_convert_raw.json") - return [FlowCellSampleBCLConvert.model_validate(sample) for sample in read_json(path)] - - -@pytest.fixture(scope="session") -def novaseq_bcl2fastq_sample_sheet_path(bcl2fastq_flow_cell_dir: Path) -> Path: - """Return the path to a NovaSeq6000 Bcl2fastq sample sheet.""" - return Path(bcl2fastq_flow_cell_dir, DemultiplexingDirsAndFiles.SAMPLE_SHEET_FILE_NAME) - - -@pytest.fixture(scope="session") -def novaseq_bcl_convert_sample_sheet_path(bcl_convert_flow_cell_dir: Path) -> Path: - """Return the path to a NovaSeq6000 bcl_convert sample sheet.""" - return Path(bcl_convert_flow_cell_dir, DemultiplexingDirsAndFiles.SAMPLE_SHEET_FILE_NAME) - - -@pytest.fixture(scope="session") -def run_parameters_wrong_instrument(run_parameters_dir: Path) -> Path: - """Return a NovaSeqX run parameters file path with a wrong instrument value.""" - return Path(run_parameters_dir, "RunParameters_novaseq_X_wrong_instrument.xml") - - -@pytest.fixture(scope="session") -def hiseq_x_single_index_run_parameters_path( - hiseq_x_single_index_flow_cell_dir: Path, -) -> Path: - """Return the path to a HiSeqX run parameters file with single index.""" - return Path( - hiseq_x_single_index_flow_cell_dir, DemultiplexingDirsAndFiles.RUN_PARAMETERS_CAMEL_CASE - ) - - -@pytest.fixture(scope="session") -def hiseq_x_dual_index_run_parameters_path( - hiseq_x_dual_index_flow_cell_dir: Path, -) -> Path: - """Return the path to a HiSeqX run parameters file with dual index.""" - return Path( - hiseq_x_dual_index_flow_cell_dir, DemultiplexingDirsAndFiles.RUN_PARAMETERS_CAMEL_CASE - ) - - -@pytest.fixture(scope="session") -def hiseq_2500_dual_index_run_parameters_path( - hiseq_2500_dual_index_flow_cell_dir: Path, -) -> Path: - """Return the path to a HiSeq2500 run parameters file with dual index.""" - return Path( - hiseq_2500_dual_index_flow_cell_dir, DemultiplexingDirsAndFiles.RUN_PARAMETERS_CAMEL_CASE - ) - - -@pytest.fixture(scope="session") -def hiseq_2500_custom_index_run_parameters_path( - hiseq_2500_custom_index_flow_cell_dir: Path, -) -> Path: - """Return the path to a HiSeq2500 run parameters file with custom index.""" - return Path( - hiseq_2500_custom_index_flow_cell_dir, DemultiplexingDirsAndFiles.RUN_PARAMETERS_CAMEL_CASE - ) - - -@pytest.fixture(scope="session") -def novaseq_6000_run_parameters_path(bcl2fastq_flow_cell_dir: Path) -> Path: - """Return the path to a NovaSeq6000 run parameters file.""" - return Path(bcl2fastq_flow_cell_dir, DemultiplexingDirsAndFiles.RUN_PARAMETERS_PASCAL_CASE) - - -@pytest.fixture -def novaseq_6000_run_parameters_pre_1_5_kits_path( - novaseq_6000_pre_1_5_kits_flow_cell: Path, -) -> Path: - """Return the path to a NovaSeq6000 pre 1.5 kit run parameters file.""" - return Path( - novaseq_6000_pre_1_5_kits_flow_cell, - DemultiplexingDirsAndFiles.RUN_PARAMETERS_PASCAL_CASE, - ) - - -@pytest.fixture -def novaseq_6000_run_parameters_post_1_5_kits_path( - novaseq_6000_post_1_5_kits_flow_cell: Path, -) -> Path: - """Return the path to a NovaSeq6000 post 1.5 kit run parameters file.""" - return Path( - novaseq_6000_post_1_5_kits_flow_cell, - DemultiplexingDirsAndFiles.RUN_PARAMETERS_PASCAL_CASE, - ) - - -@pytest.fixture(scope="session") -def novaseq_x_run_parameters_path(novaseq_x_flow_cell_dir: Path) -> Path: - """Return the path to a NovaSeqX run parameters file.""" - return Path(novaseq_x_flow_cell_dir, DemultiplexingDirsAndFiles.RUN_PARAMETERS_PASCAL_CASE) - - -@pytest.fixture(scope="function") -def run_parameters_hiseq_different_index(run_parameters_dir: Path) -> RunParametersHiSeq: - """Return a HiSeq RunParameters object with different index cycles.""" - path = Path(run_parameters_dir, "RunParameters_hiseq_2500_different_index_cycles.xml") - return RunParametersHiSeq(run_parameters_path=path) - - -@pytest.fixture(scope="function") -def run_parameters_novaseq_6000_different_index( - run_parameters_dir: Path, -) -> RunParametersNovaSeq6000: - """Return a NovaSeq6000 RunParameters object with different index cycles.""" - path = Path(run_parameters_dir, "RunParameters_novaseq_6000_different_index_cycles.xml") - return RunParametersNovaSeq6000(run_parameters_path=path) - - -@pytest.fixture(scope="function") -def run_parameters_novaseq_x_different_index(run_parameters_dir: Path) -> RunParametersNovaSeqX: - """Return a NovaSeqX RunParameters object with different index cycles.""" - path = Path(run_parameters_dir, "RunParameters_novaseq_X_different_index_cycles.xml") - return RunParametersNovaSeqX(run_parameters_path=path) - - -@pytest.fixture(scope="module") -def run_parameters_missing_versions_path( - run_parameters_dir: Path, -) -> Path: - """Return a NovaSeq6000 run parameters path without software and reagent kit versions.""" - return Path(run_parameters_dir, "RunParameters_novaseq_no_software_nor_reagent_version.xml") - - -@pytest.fixture(scope="session") -def hiseq_x_single_index_run_parameters( - hiseq_x_single_index_run_parameters_path: Path, -) -> RunParametersHiSeq: - """Return a HiSeqX run parameters object with single index.""" - return RunParametersHiSeq(run_parameters_path=hiseq_x_single_index_run_parameters_path) - - -@pytest.fixture(scope="session") -def hiseq_x_dual_index_run_parameters( - hiseq_x_dual_index_run_parameters_path: Path, -) -> RunParametersHiSeq: - """Return a HiSeqX run parameters object with dual index.""" - return RunParametersHiSeq(run_parameters_path=hiseq_x_dual_index_run_parameters_path) - - -@pytest.fixture(scope="session") -def hiseq_2500_dual_index_run_parameters( - hiseq_2500_dual_index_run_parameters_path: Path, -) -> RunParametersHiSeq: - """Return a HiSeq2500 run parameters object with dual index.""" - return RunParametersHiSeq(run_parameters_path=hiseq_2500_dual_index_run_parameters_path) - - -@pytest.fixture(scope="session") -def hiseq_2500_custom_index_run_parameters( - hiseq_2500_custom_index_run_parameters_path: Path, -) -> RunParametersHiSeq: - """Return a HiSeq2500 run parameters object with custom index.""" - return RunParametersHiSeq(run_parameters_path=hiseq_2500_custom_index_run_parameters_path) - - -@pytest.fixture(scope="session") -def novaseq_6000_run_parameters( - novaseq_6000_run_parameters_path: Path, -) -> RunParametersNovaSeq6000: - """Return a NovaSeq6000 run parameters object.""" - return RunParametersNovaSeq6000(run_parameters_path=novaseq_6000_run_parameters_path) - - -@pytest.fixture -def novaseq_6000_run_parameters_pre_1_5_kits( - novaseq_6000_run_parameters_pre_1_5_kits_path: Path, -) -> RunParametersNovaSeq6000: - """Return a NovaSeq6000 run parameters pre 1.5 kit object.""" - return RunParametersNovaSeq6000( - run_parameters_path=novaseq_6000_run_parameters_pre_1_5_kits_path - ) - - -@pytest.fixture -def novaseq_6000_run_parameters_post_1_5_kits(novaseq_6000_run_parameters_post_1_5_kits_path: Path): - """Return a NovaSeq6000 run parameters post 1.5 kit object.""" - return RunParametersNovaSeq6000( - run_parameters_path=novaseq_6000_run_parameters_post_1_5_kits_path - ) - - -@pytest.fixture(scope="session") -def novaseq_x_run_parameters( - novaseq_x_run_parameters_path: Path, -) -> RunParametersNovaSeqX: - """Return a NovaSeqX run parameters object.""" - return RunParametersNovaSeqX(run_parameters_path=novaseq_x_run_parameters_path) - - -@pytest.fixture(scope="module") -def hiseq_x_single_index_flow_cell( - hiseq_x_single_index_flow_cell_dir: Path, -) -> FlowCellDirectoryData: - """Return a single-index HiSeqX flow cell.""" - return FlowCellDirectoryData(flow_cell_path=hiseq_x_single_index_flow_cell_dir) - - -@pytest.fixture(scope="module") -def hiseq_x_dual_index_flow_cell( - hiseq_x_dual_index_flow_cell_dir: Path, -) -> FlowCellDirectoryData: - """Return a dual-index HiSeqX flow cell.""" - return FlowCellDirectoryData(flow_cell_path=hiseq_x_dual_index_flow_cell_dir) - - -@pytest.fixture(scope="module") -def hiseq_2500_dual_index_flow_cell( - hiseq_2500_dual_index_flow_cell_dir: Path, -) -> FlowCellDirectoryData: - """Return a dual-index HiSeq2500 flow cell.""" - return FlowCellDirectoryData(flow_cell_path=hiseq_2500_dual_index_flow_cell_dir) - - -@pytest.fixture(scope="module") -def hiseq_2500_custom_index_flow_cell( - hiseq_2500_custom_index_flow_cell_dir: Path, -) -> FlowCellDirectoryData: - """Return a custom-index HiSeq2500 flow cell.""" - return FlowCellDirectoryData(flow_cell_path=hiseq_2500_custom_index_flow_cell_dir) - - -@pytest.fixture(scope="session") -def bcl2fastq_flow_cell(bcl2fastq_flow_cell_dir: Path) -> FlowCellDirectoryData: - """Create a flow cell object with flow cell that is demultiplexed.""" - return FlowCellDirectoryData( - flow_cell_path=bcl2fastq_flow_cell_dir, bcl_converter=BclConverter.BCL2FASTQ - ) - - -@pytest.fixture(scope="session") -def novaseq_flow_cell_demultiplexed_with_bcl2fastq( - bcl_convert_flow_cell_dir: Path, -) -> FlowCellDirectoryData: - """Return a Novaseq6000 flow cell object demultiplexed using Bcl2fastq.""" - return FlowCellDirectoryData( - flow_cell_path=bcl_convert_flow_cell_dir, bcl_converter=BclConverter.BCL2FASTQ - ) - - -@pytest.fixture(scope="module") -def bcl_convert_flow_cell(bcl_convert_flow_cell_dir: Path) -> FlowCellDirectoryData: - """Create a bcl_convert flow cell object with flow cell that is demultiplexed.""" - return FlowCellDirectoryData( - flow_cell_path=bcl_convert_flow_cell_dir, bcl_converter=BclConverter.DRAGEN - ) - - -@pytest.fixture(scope="function") -def novaseq_6000_flow_cell(bcl_convert_flow_cell: FlowCellDirectoryData) -> FlowCellDirectoryData: - """Return a NovaSeq6000 flow cell object.""" - return bcl_convert_flow_cell - - -@pytest.fixture(scope="function") -def novaseq_x_flow_cell(novaseq_x_flow_cell_dir: Path) -> FlowCellDirectoryData: - """Create a NovaSeqX flow cell object with flow cell that is demultiplexed.""" - return FlowCellDirectoryData( - flow_cell_path=novaseq_x_flow_cell_dir, bcl_converter=BclConverter.DRAGEN - ) - - -@pytest.fixture(scope="session") -def bcl2fastq_flow_cell_id(bcl2fastq_flow_cell: FlowCellDirectoryData) -> str: - """Return flow cell id from bcl2fastq flow cell object.""" - return bcl2fastq_flow_cell.id - - -@pytest.fixture(scope="module") -def bcl_convert_flow_cell_id(bcl_convert_flow_cell: FlowCellDirectoryData) -> str: - """Return flow cell id from bcl_convert flow cell object.""" - return bcl_convert_flow_cell.id - - -@pytest.fixture(name="demultiplexing_delivery_file") -def demultiplexing_delivery_file(bcl2fastq_flow_cell: FlowCellDirectoryData) -> Path: - """Return demultiplexing delivery started file.""" - return Path(bcl2fastq_flow_cell.path, DemultiplexingDirsAndFiles.DELIVERY) - - -@pytest.fixture(name="hiseq_x_tile_dir") -def hiseq_x_tile_dir(bcl2fastq_flow_cell: FlowCellDirectoryData) -> Path: - """Return HiSeqX tile dir.""" - return Path(bcl2fastq_flow_cell.path, DemultiplexingDirsAndFiles.HISEQ_X_TILE_DIR) - - -@pytest.fixture(name="lims_novaseq_samples_file") -def lims_novaseq_samples_file(raw_lims_sample_dir: Path) -> Path: - """Return the path to a file with sample info in lims format.""" - return Path(raw_lims_sample_dir, "raw_samplesheet_novaseq.json") - - -@pytest.fixture -def lims_novaseq_6000_samples_file(bcl2fastq_flow_cell_dir: Path) -> Path: - """Return the path to the file with the raw samples of HVKJCDRXX flow cell in lims format.""" - return Path(bcl2fastq_flow_cell_dir, "HVKJCDRXX_raw.json") - - -@pytest.fixture -def lims_novaseq_samples_raw(lims_novaseq_samples_file: Path) -> list[dict]: - """Return a list of raw flow cell samples.""" - return ReadFile.get_content_from_file( - file_format=FileFormat.JSON, file_path=lims_novaseq_samples_file - ) - - -@pytest.fixture -def lims_novaseq_6000_sample_raw(lims_novaseq_6000_samples_file: Path) -> list[dict]: - """Return the list of raw samples from flow cell HVKJCDRXX.""" - return ReadFile.get_content_from_file( - file_format=FileFormat.JSON, file_path=lims_novaseq_6000_samples_file - ) - - -@pytest.fixture(name="demultiplexed_flow_cell") -def demultiplexed_flow_cell(demultiplexed_runs: Path, bcl2fastq_flow_cell_full_name: str) -> Path: - """Return the path to a demultiplexed flow cell with bcl2fastq.""" - return Path(demultiplexed_runs, bcl2fastq_flow_cell_full_name) - - -@pytest.fixture(name="bcl_convert_demultiplexed_flow_cell") -def bcl_convert_demultiplexed_flow_cell( - demultiplexed_runs: Path, bcl_convert_flow_cell_full_name: str -) -> Path: - """Return the path to a demultiplexed flow cell with BCLConvert.""" - return Path(demultiplexed_runs, bcl_convert_flow_cell_full_name) - - -@pytest.fixture(name="novaseqx_demultiplexed_flow_cell") -def novaseqx_demultiplexed_flow_cell(demultiplexed_runs: Path, novaseq_x_flow_cell_full_name: str): - """Return the path to a demultiplexed NovaSeqX flow cell.""" - return Path(demultiplexed_runs, novaseq_x_flow_cell_full_name) - - -@pytest.fixture() -def novaseqx_flow_cell_with_sample_sheet_no_fastq( - novaseqx_flow_cell_directory: Path, novaseqx_demultiplexed_flow_cell: Path -) -> FlowCellDirectoryData: - """Return a flow cell from a tmp dir with a sample sheet and no sample fastq files.""" - novaseqx_flow_cell_directory.mkdir(parents=True, exist_ok=True) - flow_cell = FlowCellDirectoryData(novaseqx_flow_cell_directory) - sample_sheet_path = Path( - novaseqx_demultiplexed_flow_cell, DemultiplexingDirsAndFiles.SAMPLE_SHEET_FILE_NAME - ) - flow_cell._sample_sheet_path_hk = sample_sheet_path - return flow_cell - - -# Genotype file fixture - - -@pytest.fixture(name="bcf_file") -def bcf_file(apps_dir: Path) -> Path: - """Return the path to a BCF file.""" - return Path(apps_dir, "gt", "yellowhog.bcf") - - -# Gens file fixtures - - -@pytest.fixture(name="gens_fracsnp_path") -def gens_fracsnp_path(mip_dna_analysis_dir: Path, sample_id: str) -> Path: - """Path to Gens fracsnp/baf bed file.""" - return Path(mip_dna_analysis_dir, f"{sample_id}.baf.bed.gz") - - -@pytest.fixture(name="gens_coverage_path") -def gens_coverage_path(mip_dna_analysis_dir: Path, sample_id: str) -> Path: - """Path to Gens coverage bed file.""" - return Path(mip_dna_analysis_dir, f"{sample_id}.cov.bed.gz") - - -# Housekeeper, Chanjo file fixtures - - -@pytest.fixture(name="bed_file") -def bed_file(analysis_dir) -> Path: - """Return the path to a bed file.""" - return Path(analysis_dir, "sample_coverage.bed") - - -# Helper fixtures - - -@pytest.fixture(scope="session") -def helpers() -> StoreHelpers: - """Return a class with helper functions for the stores.""" - return StoreHelpers() - - -@pytest.fixture(name="small_helpers") -def small_helpers() -> SmallHelpers: - """Return a class with small helper functions.""" - return SmallHelpers() - - -# HK fixtures - - -@pytest.fixture(name="root_path") -def root_path(project_dir: Path) -> Path: - """Return the path to a hk bundles dir.""" - _root_path = Path(project_dir, "bundles") - _root_path.mkdir(parents=True, exist_ok=True) - return _root_path - - -@pytest.fixture(name="hk_bundle_sample_path") -def hk_bundle_sample_path(sample_id: str, timestamp: datetime) -> Path: - """Return the relative path to a Housekeeper bundle mock sample.""" - return Path(sample_id, timestamp.strftime("%Y-%m-%d")) - - -@pytest.fixture(name="hk_bundle_data") -def hk_bundle_data( - case_id: str, - bed_file: Path, - delivery_report_html: Path, - timestamp_yesterday: datetime, - sample_id: str, - father_sample_id: str, - mother_sample_id: str, -) -> dict[str, Any]: - """Return some bundle data for Housekeeper.""" - return { - "name": case_id, - "created": timestamp_yesterday, - "expires": timestamp_yesterday, - "files": [ - { - "path": bed_file.as_posix(), - "archive": False, - "tags": ["bed", sample_id, father_sample_id, mother_sample_id, "coverage"], - }, - { - "path": delivery_report_html.as_posix(), - "archive": False, - "tags": [HK_DELIVERY_REPORT_TAG], - }, - ], - } - - -@pytest.fixture(name="hk_sample_bundle") -def hk_sample_bundle( - fastq_file: Path, - sample_hk_bundle_no_files: dict, - sample_id: str, - spring_file: Path, -) -> dict: - """Returns a dict for building a housekeeper bundle for a sample.""" - sample_hk_bundle_no_files["files"] = [ - { - "path": spring_file.as_posix(), - "archive": False, - "tags": [SequencingFileTag.SPRING, sample_id], - }, - { - "path": fastq_file.as_posix(), - "archive": False, - "tags": [SequencingFileTag.FASTQ, sample_id], - }, - ] - return sample_hk_bundle_no_files - - -@pytest.fixture(name="hk_father_sample_bundle") -def hk_father_sample_bundle( - fastq_file_father: Path, - helpers, - sample_hk_bundle_no_files: dict, - father_sample_id: str, - spring_file_father: Path, -) -> dict: - """Returns a dict for building a housekeeper bundle for a second sample.""" - father_sample_bundle = deepcopy(sample_hk_bundle_no_files) - father_sample_bundle["name"] = father_sample_id - father_sample_bundle["files"] = [ - { - "path": spring_file_father.as_posix(), - "archive": False, - "tags": [SequencingFileTag.SPRING, father_sample_id], - }, - { - "path": fastq_file_father.as_posix(), - "archive": False, - "tags": [SequencingFileTag.FASTQ, father_sample_id], - }, - ] - return father_sample_bundle - - -@pytest.fixture(name="sample_hk_bundle_no_files") -def sample_hk_bundle_no_files(sample_id: str, timestamp: datetime) -> dict: - """Create a complete bundle mock for testing compression.""" - return { - "name": sample_id, - "created": timestamp, - "expires": timestamp, - "files": [], - } +@pytest.fixture(name="sample_hk_bundle_no_files") +def sample_hk_bundle_no_files(sample_id: str, timestamp: datetime) -> dict: + """Create a complete bundle mock for testing compression.""" + return { + "name": sample_id, + "created": timestamp, + "expires": timestamp, + "files": [], + } @pytest.fixture(name="case_hk_bundle_no_files") @@ -2135,6 +1124,40 @@ def analysis_store_single_case( yield base_store +@pytest.fixture +def store_with_demultiplexed_samples( + store: Store, + helpers: StoreHelpers, + bcl_convert_demultiplexed_flow_cell_sample_internal_ids: list[str], + bcl2fastq_demultiplexed_flow_cell_sample_internal_ids: list[str], + flow_cell_name_demultiplexed_with_bcl2fastq: str, + flow_cell_name_demultiplexed_with_bcl_convert: str, +) -> Store: + """Return a store with samples that have been demultiplexed with BCL Convert and BCL2Fastq.""" + helpers.add_flow_cell( + store, flow_cell_name_demultiplexed_with_bcl_convert, sequencer_type="novaseq" + ) + helpers.add_flow_cell( + store, flow_cell_name_demultiplexed_with_bcl2fastq, sequencer_type="hiseqx" + ) + for i, sample_internal_id in enumerate(bcl_convert_demultiplexed_flow_cell_sample_internal_ids): + helpers.add_sample(store, internal_id=sample_internal_id, name=f"sample_bcl_convert_{i}") + helpers.add_sample_lane_sequencing_metrics( + store, + sample_internal_id=sample_internal_id, + flow_cell_name=flow_cell_name_demultiplexed_with_bcl_convert, + ) + + for i, sample_internal_id in enumerate(bcl2fastq_demultiplexed_flow_cell_sample_internal_ids): + helpers.add_sample(store, internal_id=sample_internal_id, name=f"sample_bcl2fastq_{i}") + helpers.add_sample_lane_sequencing_metrics( + store, + sample_internal_id=sample_internal_id, + flow_cell_name=flow_cell_name_demultiplexed_with_bcl2fastq, + ) + return store + + @pytest.fixture(name="collaboration_id") def collaboration_id() -> str: """Return a default customer group.""" @@ -3476,6 +2499,7 @@ def nf_analysis_housekeeper( helpers: StoreHelpers, mock_fastq_files: list[Path], sample_id: str, + timestamp_now: datetime, ): """Create populated Housekeeper sample bundle mock.""" diff --git a/tests/fixture_plugins/__init__.py b/tests/fixture_plugins/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/fixture_plugins/demultiplex_fixtures/__init__.py b/tests/fixture_plugins/demultiplex_fixtures/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/fixture_plugins/demultiplex_fixtures/flow_cell_fixtures.py b/tests/fixture_plugins/demultiplex_fixtures/flow_cell_fixtures.py new file mode 100644 index 0000000000..09213170bd --- /dev/null +++ b/tests/fixture_plugins/demultiplex_fixtures/flow_cell_fixtures.py @@ -0,0 +1,172 @@ +"""Fixtures for flow cell objects.""" +from pathlib import Path + +import pytest + +from cg.constants.demultiplexing import BclConverter, DemultiplexingDirsAndFiles +from cg.models.flow_cell.flow_cell import FlowCellDirectoryData + +# Functional flow cells + + +@pytest.fixture(scope="module") +def hiseq_x_single_index_flow_cell( + hiseq_x_single_index_flow_cell_dir: Path, +) -> FlowCellDirectoryData: + """Return a single-index HiSeqX flow cell.""" + return FlowCellDirectoryData(flow_cell_path=hiseq_x_single_index_flow_cell_dir) + + +@pytest.fixture(scope="module") +def hiseq_x_dual_index_flow_cell( + hiseq_x_dual_index_flow_cell_dir: Path, +) -> FlowCellDirectoryData: + """Return a dual-index HiSeqX flow cell.""" + return FlowCellDirectoryData(flow_cell_path=hiseq_x_dual_index_flow_cell_dir) + + +@pytest.fixture(scope="module") +def hiseq_2500_dual_index_flow_cell( + hiseq_2500_dual_index_flow_cell_dir: Path, +) -> FlowCellDirectoryData: + """Return a dual-index HiSeq2500 flow cell.""" + return FlowCellDirectoryData(flow_cell_path=hiseq_2500_dual_index_flow_cell_dir) + + +@pytest.fixture(scope="module") +def hiseq_2500_custom_index_flow_cell( + hiseq_2500_custom_index_flow_cell_dir: Path, +) -> FlowCellDirectoryData: + """Return a custom-index HiSeq2500 flow cell.""" + return FlowCellDirectoryData(flow_cell_path=hiseq_2500_custom_index_flow_cell_dir) + + +@pytest.fixture() +def novaseq_6000_post_1_5_kits_flow_cell_data(flow_cells_dir: Path) -> FlowCellDirectoryData: + return FlowCellDirectoryData(Path(flow_cells_dir, "230912_A00187_1009_AHK33MDRX3")) + + +@pytest.fixture() +def novaseq_6000_pre_1_5_kits_flow_cell_data(flow_cells_dir: Path) -> FlowCellDirectoryData: + return FlowCellDirectoryData(Path(flow_cells_dir, "190927_A00689_0069_BHLYWYDSXX")) + + +@pytest.fixture() +def novaseq_x_flow_cell_data(flow_cells_dir: Path) -> FlowCellDirectoryData: + return FlowCellDirectoryData(Path(flow_cells_dir, "20231108_LH00188_0028_B22F52TLT3")) + + +# Broken flow cells + + +@pytest.fixture(scope="session") +def bcl2fastq_flow_cell(bcl2fastq_flow_cell_dir: Path) -> FlowCellDirectoryData: + """Create a flow cell object with flow cell that is demultiplexed.""" + return FlowCellDirectoryData( + flow_cell_path=bcl2fastq_flow_cell_dir, bcl_converter=BclConverter.BCL2FASTQ + ) + + +@pytest.fixture(scope="session") +def novaseq_flow_cell_demultiplexed_with_bcl2fastq( + bcl_convert_flow_cell_dir: Path, +) -> FlowCellDirectoryData: + """Return a Novaseq6000 flow cell object demultiplexed using Bcl2fastq.""" + return FlowCellDirectoryData( + flow_cell_path=bcl_convert_flow_cell_dir, bcl_converter=BclConverter.BCL2FASTQ + ) + + +@pytest.fixture(scope="module") +def bcl_convert_flow_cell(bcl_convert_flow_cell_dir: Path) -> FlowCellDirectoryData: + """Create a bcl_convert flow cell object with flow cell that is demultiplexed.""" + return FlowCellDirectoryData( + flow_cell_path=bcl_convert_flow_cell_dir, bcl_converter=BclConverter.DRAGEN + ) + + +@pytest.fixture(scope="function") +def novaseq_6000_flow_cell(bcl_convert_flow_cell: FlowCellDirectoryData) -> FlowCellDirectoryData: + """Return a NovaSeq6000 flow cell object.""" + return bcl_convert_flow_cell + + +@pytest.fixture(scope="function") +def novaseq_x_flow_cell(novaseq_x_flow_cell_dir: Path) -> FlowCellDirectoryData: + """Create a NovaSeqX flow cell object with flow cell that is demultiplexed.""" + return FlowCellDirectoryData( + flow_cell_path=novaseq_x_flow_cell_dir, bcl_converter=BclConverter.DRAGEN + ) + + +@pytest.fixture() +def novaseqx_flow_cell_with_sample_sheet_no_fastq( + novaseqx_flow_cell_directory: Path, novaseqx_demultiplexed_flow_cell: Path +) -> FlowCellDirectoryData: + """Return a flow cell from a tmp dir with a sample sheet and no sample fastq files.""" + novaseqx_flow_cell_directory.mkdir(parents=True, exist_ok=True) + flow_cell = FlowCellDirectoryData(novaseqx_flow_cell_directory) + sample_sheet_path = Path( + novaseqx_demultiplexed_flow_cell, DemultiplexingDirsAndFiles.SAMPLE_SHEET_FILE_NAME + ) + flow_cell._sample_sheet_path_hk = sample_sheet_path + return flow_cell + + +@pytest.fixture(name="tmp_bcl2fastq_flow_cell") +def tmp_bcl2fastq_flow_cell( + tmp_demultiplexed_runs_bcl2fastq_directory: Path, +) -> FlowCellDirectoryData: + """Create a flow cell object with flow cell that is demultiplexed.""" + return FlowCellDirectoryData( + flow_cell_path=tmp_demultiplexed_runs_bcl2fastq_directory, + bcl_converter=BclConverter.BCL2FASTQ, + ) + + +@pytest.fixture +def novaseq6000_flow_cell( + tmp_flow_cells_directory_malformed_sample_sheet: Path, +) -> FlowCellDirectoryData: + """Return a NovaSeq6000 flow cell.""" + return FlowCellDirectoryData( + flow_cell_path=tmp_flow_cells_directory_malformed_sample_sheet, + bcl_converter=BclConverter.BCLCONVERT, + ) + + +@pytest.fixture(name="tmp_bcl_convert_flow_cell") +def tmp_bcl_convert_flow_cell( + tmp_flow_cell_directory_bclconvert: Path, +) -> FlowCellDirectoryData: + """Create a flow cell object with flow cell that is demultiplexed.""" + return FlowCellDirectoryData( + flow_cell_path=tmp_flow_cell_directory_bclconvert, + bcl_converter=BclConverter.DRAGEN, + ) + + +@pytest.fixture(name="tmp_unfinished_bcl2fastq_flow_cell") +def unfinished_bcl2fastq_flow_cell( + demultiplexed_runs_unfinished_bcl2fastq_flow_cell_directory: Path, +) -> FlowCellDirectoryData: + """Copy the content of a demultiplexed but not finished directory to a temporary location.""" + return FlowCellDirectoryData( + flow_cell_path=demultiplexed_runs_unfinished_bcl2fastq_flow_cell_directory, + bcl_converter=BclConverter.BCL2FASTQ, + ) + + +# Flow cell attributes + + +@pytest.fixture(scope="session") +def bcl2fastq_flow_cell_id(bcl2fastq_flow_cell: FlowCellDirectoryData) -> str: + """Return flow cell id from bcl2fastq flow cell object.""" + return bcl2fastq_flow_cell.id + + +@pytest.fixture(scope="module") +def bcl_convert_flow_cell_id(bcl_convert_flow_cell: FlowCellDirectoryData) -> str: + """Return flow cell id from bcl_convert flow cell object.""" + return bcl_convert_flow_cell.id diff --git a/tests/fixture_plugins/demultiplex_fixtures/name_fixtures.py b/tests/fixture_plugins/demultiplex_fixtures/name_fixtures.py new file mode 100644 index 0000000000..c4bbb1d3b4 --- /dev/null +++ b/tests/fixture_plugins/demultiplex_fixtures/name_fixtures.py @@ -0,0 +1,116 @@ +import pytest + + +@pytest.fixture +def tmp_flow_cell_name_no_run_parameters() -> str: + """This is the name of a flow cell directory with the run parameters missing.""" + return "180522_A00689_0200_BHLCKNCCXY" + + +@pytest.fixture +def tmp_flow_cell_name_malformed_sample_sheet() -> str: + """ "Returns the name of a flow cell directory ready for demultiplexing with BCL convert. + Contains a sample sheet with malformed headers. + """ + return "201203_A00689_0200_AHVKJCDRXY" + + +@pytest.fixture +def tmp_flow_cell_name_no_sample_sheet() -> str: + """Return the name of a flow cell directory with the run parameters and sample sheet missing.""" + return "170407_A00689_0209_BHHKVCALXX" + + +@pytest.fixture(name="tmp_flow_cell_name_ready_for_demultiplexing_bcl2fastq") +def tmp_flow_cell_name_ready_for_demultiplexing_bcl2fastq() -> str: + """Returns the name of a flow cell directory ready for demultiplexing with bcl2fastq.""" + return "211101_D00483_0615_AHLG5GDRXY" + + +@pytest.fixture(scope="session") +def flow_cell_name_demultiplexed_with_bcl2fastq() -> str: + """Return the name of a flow cell that has been demultiplexed with BCL2Fastq.""" + return "HHKVCALXX" + + +@pytest.fixture(scope="session") +def flow_cell_directory_name_demultiplexed_with_bcl2fastq( + flow_cell_name_demultiplexed_with_bcl2fastq: str, +) -> str: + """Return the name of a flow cell directory that has been demultiplexed with BCL2Fastq.""" + return f"170407_ST-E00198_0209_B{flow_cell_name_demultiplexed_with_bcl2fastq}" + + +@pytest.fixture(scope="session") +def flow_cell_name_demultiplexed_with_bcl_convert() -> str: + return "HY7FFDRX2" + + +@pytest.fixture(scope="session") +def flow_cell_directory_name_demultiplexed_with_bcl_convert( + flow_cell_name_demultiplexed_with_bcl_convert: str, +) -> str: + return f"230504_A00689_0804_B{flow_cell_name_demultiplexed_with_bcl_convert}" + + +@pytest.fixture(scope="session") +def hiseq_x_single_index_flow_cell_name() -> str: + """Return the full name of a HiSeqX flow cell with only one index.""" + return "170517_ST-E00266_0210_BHJCFFALXX" + + +@pytest.fixture(scope="session") +def hiseq_x_dual_index_flow_cell_name() -> str: + """Return the full name of a HiSeqX flow cell with two indexes.""" + return "180508_ST-E00269_0269_AHL32LCCXY" + + +@pytest.fixture(scope="session") +def hiseq_2500_dual_index_flow_cell_name() -> str: + """Return the full name of a HiSeq2500 flow cell with double indexes.""" + return "181005_D00410_0735_BHM2LNBCX2" + + +@pytest.fixture(scope="session") +def hiseq_2500_custom_index_flow_cell_name() -> str: + """Return the full name of a HiSeq2500 flow cell with double indexes.""" + return "180509_D00450_0598_BHGYFNBCX2" + + +@pytest.fixture(scope="session") +def bcl2fastq_flow_cell_full_name() -> str: + """Return full flow cell name.""" + return "201203_D00483_0200_AHVKJCDRXX" + + +@pytest.fixture(scope="session") +def bcl_convert_flow_cell_full_name() -> str: + """Return the full name of a bcl_convert flow cell.""" + return "211101_A00187_0615_AHLG5GDRZZ" + + +@pytest.fixture(scope="session") +def novaseq_x_flow_cell_full_name() -> str: + """Return the full name of a NovaSeqX flow cell.""" + return "20230508_LH00188_0003_A22522YLT3" + + +# Lists + + +@pytest.fixture(scope="session") +def bcl_convert_demultiplexed_flow_cell_sample_internal_ids() -> list[str]: + """ + Sample id:s present in sample sheet for dummy flow cell demultiplexed with BCL Convert in + cg/tests/fixtures/apps/demultiplexing/demultiplexed-runs/230504_A00689_0804_BHY7FFDRX2. + """ + return ["ACC11927A2", "ACC11927A5"] + + +@pytest.fixture(scope="session") +def bcl2fastq_demultiplexed_flow_cell_sample_internal_ids() -> list[str]: + """ + Sample id:s present in sample sheet for dummy flow cell demultiplexed with BCL Convert in + cg/tests/fixtures/apps/demultiplexing/demultiplexed-runs/170407_A00689_0209_BHHKVCALXX. + """ + return ["SVE2528A1"] diff --git a/tests/fixture_plugins/demultiplex_fixtures/path_fixtures.py b/tests/fixture_plugins/demultiplex_fixtures/path_fixtures.py new file mode 100644 index 0000000000..7629a2a27f --- /dev/null +++ b/tests/fixture_plugins/demultiplex_fixtures/path_fixtures.py @@ -0,0 +1,456 @@ +"""Path fixtures for demultiplex tests.""" +import shutil +from pathlib import Path + +import pytest + +from cg.constants.demultiplexing import DemultiplexingDirsAndFiles +from cg.constants.nanopore_files import NanoporeDirsAndFiles +from cg.models.flow_cell.flow_cell import FlowCellDirectoryData + + +@pytest.fixture(name="tmp_flow_cells_directory") +def tmp_flow_cells_directory(tmp_path: Path, flow_cells_dir: Path) -> Path: + """ + Return the path to a temporary flow cells directory with flow cells ready for demultiplexing. + Generates a copy of the original flow cells directory + """ + original_dir = flow_cells_dir + tmp_dir = Path(tmp_path, "flow_cells") + + return Path(shutil.copytree(original_dir, tmp_dir)) + + +@pytest.fixture(name="tmp_flow_cells_demux_all_directory") +def tmp_flow_cells_demux_all_directory(tmp_path: Path, flow_cells_demux_all_dir: Path) -> Path: + """ + Return the path to a temporary flow cells directory with flow cells ready for demultiplexing. + Generates a copy of the original flow cells directory. + This fixture is used for testing of the cg demutliplex all cmd. + """ + original_dir = flow_cells_demux_all_dir + tmp_dir = Path(tmp_path, "flow_cells_demux_all") + + return Path(shutil.copytree(original_dir, tmp_dir)) + + +@pytest.fixture(name="tmp_flow_cell_directory_bcl2fastq") +def flow_cell_working_directory_bcl2fastq( + bcl2fastq_flow_cell_dir: Path, tmp_flow_cells_directory: Path +) -> Path: + """Return the path to a working directory that will be deleted after test is run. + + This is a path to a flow cell directory with the run parameters present. + """ + return Path(tmp_flow_cells_directory, bcl2fastq_flow_cell_dir.name) + + +@pytest.fixture(name="tmp_flow_cell_directory_bclconvert") +def flow_cell_working_directory_bclconvert( + bcl_convert_flow_cell_dir: Path, tmp_flow_cells_directory: Path +) -> Path: + """Return the path to a working directory that will be deleted after test is run. + This is a path to a flow cell directory with the run parameters present. + """ + return Path(tmp_flow_cells_directory, bcl_convert_flow_cell_dir.name) + + +@pytest.fixture +def tmp_flow_cells_directory_no_run_parameters( + tmp_flow_cell_name_no_run_parameters: str, tmp_flow_cells_directory: Path +) -> Path: + """This is a path to a flow cell directory with the run parameters missing.""" + return Path(tmp_flow_cells_directory, tmp_flow_cell_name_no_run_parameters) + + +@pytest.fixture(name="tmp_flow_cells_directory_no_sample_sheet") +def tmp_flow_cells_directory_no_sample_sheet( + tmp_flow_cell_name_no_sample_sheet: str, tmp_flow_cells_directory: Path +) -> Path: + """This is a path to a flow cell directory with the sample sheet and run parameters missing.""" + return Path(tmp_flow_cells_directory, tmp_flow_cell_name_no_sample_sheet) + + +@pytest.fixture +def tmp_flow_cells_directory_malformed_sample_sheet( + tmp_flow_cell_name_malformed_sample_sheet: str, tmp_flow_cells_directory: Path +) -> Path: + """This is a path to a flow cell directory with a sample sheet with malformed headers.""" + return Path(tmp_flow_cells_directory, tmp_flow_cell_name_malformed_sample_sheet) + + +@pytest.fixture +def tmp_flow_cells_directory_ready_for_demultiplexing_bcl_convert( + bcl_convert_flow_cell_full_name: str, tmp_flow_cells_directory: Path +) -> Path: + """This is a path to a flow cell directory with the run parameters missing.""" + return Path(tmp_flow_cells_directory, bcl_convert_flow_cell_full_name) + + +@pytest.fixture +def tmp_flow_cells_directory_ready_for_demultiplexing_bcl2fastq( + tmp_flow_cell_name_ready_for_demultiplexing_bcl2fastq: str, tmp_flow_cells_directory: Path +) -> Path: + """This is a path to a flow cell directory with the run parameters missing.""" + return Path(tmp_flow_cells_directory, tmp_flow_cell_name_ready_for_demultiplexing_bcl2fastq) + + +# Temporary demultiplexed runs fixtures +@pytest.fixture(name="tmp_demultiplexed_runs_directory") +def tmp_demultiplexed_flow_cells_directory(tmp_path: Path, demultiplexed_runs: Path) -> Path: + """Return the path to a temporary demultiplex-runs directory. + Generates a copy of the original demultiplexed-runs + """ + original_dir = demultiplexed_runs + tmp_dir = Path(tmp_path, "demultiplexed-runs") + return Path(shutil.copytree(original_dir, tmp_dir)) + + +@pytest.fixture(name="tmp_demultiplexed_runs_bcl2fastq_directory") +def tmp_demultiplexed_runs_bcl2fastq_directory( + tmp_demultiplexed_runs_directory: Path, bcl2fastq_flow_cell_dir: Path +) -> Path: + """Return the path to a temporary demultiplex-runs bcl2fastq flow cell directory.""" + return Path(tmp_demultiplexed_runs_directory, bcl2fastq_flow_cell_dir.name) + + +@pytest.fixture(name="tmp_demultiplexed_runs_not_finished_directory") +def tmp_demultiplexed_runs_not_finished_flow_cells_directory( + tmp_path: Path, demux_results_not_finished_dir: Path +) -> Path: + """ + Return a temporary demultiplex-runs-unfinished path with an unfinished flow cell directory. + Generates a copy of the original demultiplexed-runs-unfinished directory. + """ + original_dir = demux_results_not_finished_dir + tmp_dir = Path(tmp_path, "demultiplexed-runs-unfinished") + return Path(shutil.copytree(original_dir, tmp_dir)) + + +@pytest.fixture(name="demultiplexed_runs_unfinished_bcl2fastq_flow_cell_directory") +def demultiplexed_runs_bcl2fastq_flow_cell_directory( + tmp_demultiplexed_runs_not_finished_directory: Path, + bcl2fastq_flow_cell_full_name: str, +) -> Path: + """Copy the content of a demultiplexed but not finished directory to a temporary location.""" + return Path(tmp_demultiplexed_runs_not_finished_directory, bcl2fastq_flow_cell_full_name) + + +@pytest.fixture(name="novaseq6000_bcl_convert_sample_sheet_path") +def novaseq6000_sample_sheet_path() -> Path: + """Return the path to a NovaSeq 6000 BCL convert sample sheet.""" + return Path( + "tests", + "fixtures", + "apps", + "sequencing_metrics_parser", + "230622_A00621_0864_AHY7FFDRX2", + "Unaligned", + "Reports", + "SampleSheet.csv", + ) + + +@pytest.fixture(scope="session") +def demultiplex_fixtures(apps_dir: Path) -> Path: + """Return the path to the demultiplex fixture directory.""" + return Path(apps_dir, "demultiplexing") + + +@pytest.fixture(scope="session") +def raw_lims_sample_dir(demultiplex_fixtures: Path) -> Path: + """Return the path to the raw samples fixture directory.""" + return Path(demultiplex_fixtures, "raw_lims_samples") + + +@pytest.fixture(scope="session") +def run_parameters_dir(demultiplex_fixtures: Path) -> Path: + """Return the path to the run parameters fixture directory.""" + return Path(demultiplex_fixtures, "run_parameters") + + +@pytest.fixture(scope="session") +def demultiplexed_runs(demultiplex_fixtures: Path) -> Path: + """Return the path to the demultiplexed flow cells fixture directory.""" + return Path(demultiplex_fixtures, "demultiplexed-runs") + + +@pytest.fixture(scope="session") +def flow_cells_dir(demultiplex_fixtures: Path) -> Path: + """Return the path to the sequenced flow cells fixture directory.""" + return Path(demultiplex_fixtures, DemultiplexingDirsAndFiles.FLOW_CELLS_DIRECTORY_NAME) + + +@pytest.fixture(scope="session") +def nanopore_flow_cells_dir(demultiplex_fixtures: Path) -> Path: + """Return the path to the sequenced flow cells fixture directory.""" + return Path(demultiplex_fixtures, NanoporeDirsAndFiles.DATA_DIRECTORY) + + +@pytest.fixture(scope="session") +def flow_cells_demux_all_dir(demultiplex_fixtures: Path) -> Path: + """Return the path to the sequenced flow cells fixture directory.""" + return Path(demultiplex_fixtures, "flow_cells_demux_all") + + +@pytest.fixture(scope="session") +def demux_results_not_finished_dir(demultiplex_fixtures: Path) -> Path: + """Return the path to a dir with demultiplexing results where nothing has been cleaned.""" + return Path(demultiplex_fixtures, "demultiplexed-runs-unfinished") + + +@pytest.fixture +def novaseq_6000_post_1_5_kits_flow_cell(tmp_flow_cells_directory: Path) -> Path: + return Path(tmp_flow_cells_directory, "230912_A00187_1009_AHK33MDRX3") + + +@pytest.fixture +def novaseq_6000_post_1_5_kits_correct_sample_sheet( + novaseq_6000_post_1_5_kits_flow_cell: Path, +) -> Path: + return Path(novaseq_6000_post_1_5_kits_flow_cell, "CorrectSampleSheet.csv") + + +@pytest.fixture +def novaseq_6000_post_1_5_kits_raw_lims_samples( + novaseq_6000_post_1_5_kits_flow_cell: Path, +) -> Path: + return Path(novaseq_6000_post_1_5_kits_flow_cell, "HK33MDRX3_raw.json") + + +@pytest.fixture +def novaseq_6000_pre_1_5_kits_flow_cell(tmp_flow_cells_directory: Path) -> Path: + return Path(tmp_flow_cells_directory, "190927_A00689_0069_BHLYWYDSXX") + + +@pytest.fixture +def novaseq_6000_pre_1_5_kits_correct_sample_sheet( + novaseq_6000_pre_1_5_kits_flow_cell: Path, +) -> Path: + return Path(novaseq_6000_pre_1_5_kits_flow_cell, "CorrectSampleSheet.csv") + + +@pytest.fixture +def novaseq_6000_pre_1_5_kits_raw_lims_samples(novaseq_6000_pre_1_5_kits_flow_cell: Path) -> Path: + return Path(novaseq_6000_pre_1_5_kits_flow_cell, "HLYWYDSXX_raw.json") + + +@pytest.fixture +def novaseq_x_flow_cell_directory(tmp_flow_cells_directory: Path) -> Path: + return Path(tmp_flow_cells_directory, "20231108_LH00188_0028_B22F52TLT3") + + +@pytest.fixture +def novaseq_x_correct_sample_sheet(novaseq_x_flow_cell_directory: Path) -> Path: + return Path(novaseq_x_flow_cell_directory, "CorrectSampleSheet.csv") + + +@pytest.fixture +def novaseq_x_raw_lims_samples(novaseq_x_flow_cell_directory: Path) -> Path: + return Path(novaseq_x_flow_cell_directory, "22F52TLT3_raw.json") + + +@pytest.fixture(scope="session") +def novaseq_x_manifest_file(novaseq_x_flow_cell_dir: Path) -> Path: + """Return the path to a NovaSeqX manifest file.""" + return Path(novaseq_x_flow_cell_dir, "Manifest.tsv") + + +@pytest.fixture(scope="session") +def hiseq_x_single_index_flow_cell_dir( + flow_cells_dir: Path, hiseq_x_single_index_flow_cell_name: str +) -> Path: + """Return the path to a HiSeqX flow cell.""" + return Path(flow_cells_dir, hiseq_x_single_index_flow_cell_name) + + +@pytest.fixture(scope="session") +def hiseq_x_dual_index_flow_cell_dir( + flow_cells_dir: Path, hiseq_x_dual_index_flow_cell_name: str +) -> Path: + """Return the path to a HiSeqX flow cell.""" + return Path(flow_cells_dir, hiseq_x_dual_index_flow_cell_name) + + +@pytest.fixture(scope="session") +def hiseq_2500_dual_index_flow_cell_dir( + flow_cells_dir: Path, hiseq_2500_dual_index_flow_cell_name: str +) -> Path: + """Return the path to a HiSeq2500 flow cell.""" + return Path(flow_cells_dir, hiseq_2500_dual_index_flow_cell_name) + + +@pytest.fixture(scope="session") +def hiseq_2500_custom_index_flow_cell_dir( + flow_cells_dir: Path, hiseq_2500_custom_index_flow_cell_name: str +) -> Path: + """Return the path to a HiSeq2500 flow cell.""" + return Path(flow_cells_dir, hiseq_2500_custom_index_flow_cell_name) + + +@pytest.fixture(scope="session") +def bcl2fastq_flow_cell_dir(flow_cells_dir: Path, bcl2fastq_flow_cell_full_name: str) -> Path: + """Return the path to the bcl2fastq flow cell demultiplex fixture directory.""" + return Path(flow_cells_dir, bcl2fastq_flow_cell_full_name) + + +@pytest.fixture(scope="session") +def bcl_convert_flow_cell_dir(flow_cells_dir: Path, bcl_convert_flow_cell_full_name: str) -> Path: + """Return the path to the bcl_convert flow cell demultiplex fixture directory.""" + return Path(flow_cells_dir, bcl_convert_flow_cell_full_name) + + +@pytest.fixture(scope="session") +def novaseq_x_flow_cell_dir(flow_cells_dir: Path, novaseq_x_flow_cell_full_name: str) -> Path: + """Return the path to the NovaSeqX flow cell demultiplex fixture directory.""" + return Path(flow_cells_dir, novaseq_x_flow_cell_full_name) + + +@pytest.fixture(scope="session") +def novaseq_bcl2fastq_sample_sheet_path(bcl2fastq_flow_cell_dir: Path) -> Path: + """Return the path to a NovaSeq6000 Bcl2fastq sample sheet.""" + return Path(bcl2fastq_flow_cell_dir, DemultiplexingDirsAndFiles.SAMPLE_SHEET_FILE_NAME) + + +@pytest.fixture(scope="session") +def novaseq_bcl_convert_sample_sheet_path(bcl_convert_flow_cell_dir: Path) -> Path: + """Return the path to a NovaSeq6000 bcl_convert sample sheet.""" + return Path(bcl_convert_flow_cell_dir, DemultiplexingDirsAndFiles.SAMPLE_SHEET_FILE_NAME) + + +@pytest.fixture(scope="session") +def run_parameters_wrong_instrument(run_parameters_dir: Path) -> Path: + """Return a NovaSeqX run parameters file path with a wrong instrument value.""" + return Path(run_parameters_dir, "RunParameters_novaseq_X_wrong_instrument.xml") + + +@pytest.fixture(scope="session") +def hiseq_x_single_index_run_parameters_path( + hiseq_x_single_index_flow_cell_dir: Path, +) -> Path: + """Return the path to a HiSeqX run parameters file with single index.""" + return Path( + hiseq_x_single_index_flow_cell_dir, DemultiplexingDirsAndFiles.RUN_PARAMETERS_CAMEL_CASE + ) + + +@pytest.fixture(scope="session") +def hiseq_x_dual_index_run_parameters_path( + hiseq_x_dual_index_flow_cell_dir: Path, +) -> Path: + """Return the path to a HiSeqX run parameters file with dual index.""" + return Path( + hiseq_x_dual_index_flow_cell_dir, DemultiplexingDirsAndFiles.RUN_PARAMETERS_CAMEL_CASE + ) + + +@pytest.fixture(scope="session") +def hiseq_2500_dual_index_run_parameters_path( + hiseq_2500_dual_index_flow_cell_dir: Path, +) -> Path: + """Return the path to a HiSeq2500 run parameters file with dual index.""" + return Path( + hiseq_2500_dual_index_flow_cell_dir, DemultiplexingDirsAndFiles.RUN_PARAMETERS_CAMEL_CASE + ) + + +@pytest.fixture(scope="session") +def hiseq_2500_custom_index_run_parameters_path( + hiseq_2500_custom_index_flow_cell_dir: Path, +) -> Path: + """Return the path to a HiSeq2500 run parameters file with custom index.""" + return Path( + hiseq_2500_custom_index_flow_cell_dir, DemultiplexingDirsAndFiles.RUN_PARAMETERS_CAMEL_CASE + ) + + +@pytest.fixture(scope="session") +def novaseq_6000_run_parameters_path(bcl2fastq_flow_cell_dir: Path) -> Path: + """Return the path to a NovaSeq6000 run parameters file.""" + return Path(bcl2fastq_flow_cell_dir, DemultiplexingDirsAndFiles.RUN_PARAMETERS_PASCAL_CASE) + + +@pytest.fixture +def novaseq_6000_run_parameters_pre_1_5_kits_path( + novaseq_6000_pre_1_5_kits_flow_cell: Path, +) -> Path: + """Return the path to a NovaSeq6000 pre 1.5 kit run parameters file.""" + return Path( + novaseq_6000_pre_1_5_kits_flow_cell, + DemultiplexingDirsAndFiles.RUN_PARAMETERS_PASCAL_CASE, + ) + + +@pytest.fixture +def novaseq_6000_run_parameters_post_1_5_kits_path( + novaseq_6000_post_1_5_kits_flow_cell: Path, +) -> Path: + """Return the path to a NovaSeq6000 post 1.5 kit run parameters file.""" + return Path( + novaseq_6000_post_1_5_kits_flow_cell, + DemultiplexingDirsAndFiles.RUN_PARAMETERS_PASCAL_CASE, + ) + + +@pytest.fixture(scope="session") +def novaseq_x_run_parameters_path(novaseq_x_flow_cell_dir: Path) -> Path: + """Return the path to a NovaSeqX run parameters file.""" + return Path(novaseq_x_flow_cell_dir, DemultiplexingDirsAndFiles.RUN_PARAMETERS_PASCAL_CASE) + + +@pytest.fixture(scope="module") +def run_parameters_missing_versions_path( + run_parameters_dir: Path, +) -> Path: + """Return a NovaSeq6000 run parameters path without software and reagent kit versions.""" + return Path(run_parameters_dir, "RunParameters_novaseq_no_software_nor_reagent_version.xml") + + +@pytest.fixture(name="demultiplexing_delivery_file") +def demultiplexing_delivery_file(bcl2fastq_flow_cell: FlowCellDirectoryData) -> Path: + """Return demultiplexing delivery started file.""" + return Path(bcl2fastq_flow_cell.path, DemultiplexingDirsAndFiles.DELIVERY) + + +@pytest.fixture(name="hiseq_x_tile_dir") +def hiseq_x_tile_dir(bcl2fastq_flow_cell: FlowCellDirectoryData) -> Path: + """Return HiSeqX tile dir.""" + return Path(bcl2fastq_flow_cell.path, DemultiplexingDirsAndFiles.HISEQ_X_TILE_DIR) + + +@pytest.fixture(name="lims_novaseq_samples_file") +def lims_novaseq_samples_file(raw_lims_sample_dir: Path) -> Path: + """Return the path to a file with sample info in lims format.""" + return Path(raw_lims_sample_dir, "raw_samplesheet_novaseq.json") + + +@pytest.fixture +def lims_novaseq_6000_samples_file(bcl2fastq_flow_cell_dir: Path) -> Path: + """Return the path to the file with the raw samples of HVKJCDRXX flow cell in lims format.""" + return Path(bcl2fastq_flow_cell_dir, "HVKJCDRXX_raw.json") + + +@pytest.fixture(name="demultiplexed_flow_cell") +def demultiplexed_flow_cell(demultiplexed_runs: Path, bcl2fastq_flow_cell_full_name: str) -> Path: + """Return the path to a demultiplexed flow cell with bcl2fastq.""" + return Path(demultiplexed_runs, bcl2fastq_flow_cell_full_name) + + +@pytest.fixture(name="bcl_convert_demultiplexed_flow_cell") +def bcl_convert_demultiplexed_flow_cell( + demultiplexed_runs: Path, bcl_convert_flow_cell_full_name: str +) -> Path: + """Return the path to a demultiplexed flow cell with BCLConvert.""" + return Path(demultiplexed_runs, bcl_convert_flow_cell_full_name) + + +# Fixtures for test demultiplex flow cell +@pytest.fixture +def tmp_empty_demultiplexed_runs_directory(tmp_demultiplexed_runs_directory) -> Path: + return Path(tmp_demultiplexed_runs_directory, "empty") + + +@pytest.fixture(name="novaseqx_demultiplexed_flow_cell") +def novaseqx_demultiplexed_flow_cell(demultiplexed_runs: Path, novaseq_x_flow_cell_full_name: str): + """Return the path to a demultiplexed NovaSeqX flow cell.""" + return Path(demultiplexed_runs, novaseq_x_flow_cell_full_name) diff --git a/tests/fixture_plugins/demultiplex_fixtures/run_parameters_fixtures.py b/tests/fixture_plugins/demultiplex_fixtures/run_parameters_fixtures.py new file mode 100644 index 0000000000..e8c7459cec --- /dev/null +++ b/tests/fixture_plugins/demultiplex_fixtures/run_parameters_fixtures.py @@ -0,0 +1,98 @@ +from pathlib import Path + +import pytest + +from cg.models.demultiplex.run_parameters import ( + RunParametersHiSeq, + RunParametersNovaSeq6000, + RunParametersNovaSeqX, +) + + +@pytest.fixture(scope="function") +def run_parameters_hiseq_different_index(run_parameters_dir: Path) -> RunParametersHiSeq: + """Return a HiSeq RunParameters object with different index cycles.""" + path = Path(run_parameters_dir, "RunParameters_hiseq_2500_different_index_cycles.xml") + return RunParametersHiSeq(run_parameters_path=path) + + +@pytest.fixture(scope="function") +def run_parameters_novaseq_6000_different_index( + run_parameters_dir: Path, +) -> RunParametersNovaSeq6000: + """Return a NovaSeq6000 RunParameters object with different index cycles.""" + path = Path(run_parameters_dir, "RunParameters_novaseq_6000_different_index_cycles.xml") + return RunParametersNovaSeq6000(run_parameters_path=path) + + +@pytest.fixture(scope="function") +def run_parameters_novaseq_x_different_index(run_parameters_dir: Path) -> RunParametersNovaSeqX: + """Return a NovaSeqX RunParameters object with different index cycles.""" + path = Path(run_parameters_dir, "RunParameters_novaseq_X_different_index_cycles.xml") + return RunParametersNovaSeqX(run_parameters_path=path) + + +@pytest.fixture(scope="session") +def hiseq_x_single_index_run_parameters( + hiseq_x_single_index_run_parameters_path: Path, +) -> RunParametersHiSeq: + """Return a HiSeqX run parameters object with single index.""" + return RunParametersHiSeq(run_parameters_path=hiseq_x_single_index_run_parameters_path) + + +@pytest.fixture(scope="session") +def hiseq_x_dual_index_run_parameters( + hiseq_x_dual_index_run_parameters_path: Path, +) -> RunParametersHiSeq: + """Return a HiSeqX run parameters object with dual index.""" + return RunParametersHiSeq(run_parameters_path=hiseq_x_dual_index_run_parameters_path) + + +@pytest.fixture(scope="session") +def hiseq_2500_dual_index_run_parameters( + hiseq_2500_dual_index_run_parameters_path: Path, +) -> RunParametersHiSeq: + """Return a HiSeq2500 run parameters object with dual index.""" + return RunParametersHiSeq(run_parameters_path=hiseq_2500_dual_index_run_parameters_path) + + +@pytest.fixture(scope="session") +def hiseq_2500_custom_index_run_parameters( + hiseq_2500_custom_index_run_parameters_path: Path, +) -> RunParametersHiSeq: + """Return a HiSeq2500 run parameters object with custom index.""" + return RunParametersHiSeq(run_parameters_path=hiseq_2500_custom_index_run_parameters_path) + + +@pytest.fixture(scope="session") +def novaseq_6000_run_parameters( + novaseq_6000_run_parameters_path: Path, +) -> RunParametersNovaSeq6000: + """Return a NovaSeq6000 run parameters object.""" + return RunParametersNovaSeq6000(run_parameters_path=novaseq_6000_run_parameters_path) + + +@pytest.fixture +def novaseq_6000_run_parameters_pre_1_5_kits( + novaseq_6000_run_parameters_pre_1_5_kits_path: Path, +) -> RunParametersNovaSeq6000: + """Return a NovaSeq6000 run parameters pre 1.5 kit object.""" + return RunParametersNovaSeq6000( + run_parameters_path=novaseq_6000_run_parameters_pre_1_5_kits_path + ) + + +@pytest.fixture +def novaseq_6000_run_parameters_post_1_5_kits(novaseq_6000_run_parameters_post_1_5_kits_path: Path): + """Return a NovaSeq6000 run parameters post 1.5 kit object.""" + return RunParametersNovaSeq6000( + run_parameters_path=novaseq_6000_run_parameters_post_1_5_kits_path + ) + + +@pytest.fixture(scope="session") +def novaseq_x_run_parameters( + novaseq_x_run_parameters_path: Path, +) -> RunParametersNovaSeqX: + """Return a NovaSeqX run parameters object.""" + return RunParametersNovaSeqX(run_parameters_path=novaseq_x_run_parameters_path) diff --git a/tests/fixture_plugins/demultiplex_fixtures/sample_fixtures.py b/tests/fixture_plugins/demultiplex_fixtures/sample_fixtures.py new file mode 100644 index 0000000000..9973ddf58e --- /dev/null +++ b/tests/fixture_plugins/demultiplex_fixtures/sample_fixtures.py @@ -0,0 +1,135 @@ +"""Demultiplex sample fixtures.""" +from pathlib import Path + +import pytest + +from cg.apps.demultiplex.sample_sheet.sample_models import ( + FlowCellSampleBcl2Fastq, + FlowCellSampleBCLConvert, +) +from cg.apps.demultiplex.sample_sheet.sample_sheet_creator import SampleSheetCreatorBCLConvert +from cg.constants import FileExtensions +from cg.constants.constants import FileFormat +from cg.io.controller import ReadFile +from cg.io.json import read_json +from cg.models.flow_cell.flow_cell import FlowCellDirectoryData + + +@pytest.fixture +def lims_novaseq_bcl_convert_samples( + lims_novaseq_samples_raw: list[dict], +) -> list[FlowCellSampleBCLConvert]: + """Return a list of parsed flow cell samples demultiplexed with BCL convert.""" + return [FlowCellSampleBCLConvert.model_validate(sample) for sample in lims_novaseq_samples_raw] + + +@pytest.fixture +def lims_novaseq_bcl2fastq_samples( + lims_novaseq_samples_raw: list[dict], +) -> list[FlowCellSampleBcl2Fastq]: + """Return a list of parsed Bcl2fastq flow cell samples""" + return [FlowCellSampleBcl2Fastq.model_validate(sample) for sample in lims_novaseq_samples_raw] + + +@pytest.fixture +def lims_novaseq_6000_bcl2fastq_samples( + lims_novaseq_6000_sample_raw: list[dict], +) -> list[FlowCellSampleBcl2Fastq]: + """Return a list of parsed Bcl2fastq flow cell samples""" + return [ + FlowCellSampleBcl2Fastq.model_validate(sample) for sample in lims_novaseq_6000_sample_raw + ] + + +@pytest.fixture +def bcl_convert_sample_sheet_creator( + bcl_convert_flow_cell: FlowCellDirectoryData, + lims_novaseq_bcl_convert_samples: list[FlowCellSampleBCLConvert], +) -> SampleSheetCreatorBCLConvert: + """Returns a sample sheet creator for version 2 sample sheets with dragen format.""" + return SampleSheetCreatorBCLConvert( + flow_cell=bcl_convert_flow_cell, + lims_samples=lims_novaseq_bcl_convert_samples, + ) + + +@pytest.fixture +def novaseq_6000_post_1_5_kits_lims_samples( + novaseq_6000_post_1_5_kits_raw_lims_samples: Path, +) -> list[FlowCellSampleBCLConvert]: + return [ + FlowCellSampleBCLConvert.model_validate(sample) + for sample in read_json(novaseq_6000_post_1_5_kits_raw_lims_samples) + ] + + +@pytest.fixture +def novaseq_6000_pre_1_5_kits_lims_samples( + novaseq_6000_pre_1_5_kits_raw_lims_samples: Path, +) -> list[FlowCellSampleBCLConvert]: + return [ + FlowCellSampleBCLConvert.model_validate(sample) + for sample in read_json(novaseq_6000_pre_1_5_kits_raw_lims_samples) + ] + + +@pytest.fixture +def novaseq_x_lims_samples(novaseq_x_raw_lims_samples: Path) -> list[FlowCellSampleBCLConvert]: + return [ + FlowCellSampleBCLConvert.model_validate(sample) + for sample in read_json(novaseq_x_raw_lims_samples) + ] + + +@pytest.fixture +def hiseq_x_single_index_bcl_convert_lims_samples( + hiseq_x_single_index_flow_cell_dir: Path, +) -> list[FlowCellSampleBCLConvert]: + """Return a list of BCLConvert samples from a HiSeqX single index flow cell.""" + path = Path( + hiseq_x_single_index_flow_cell_dir, f"HJCFFALXX_bcl_convert_raw{FileExtensions.JSON}" + ) + return [FlowCellSampleBCLConvert.model_validate(sample) for sample in read_json(path)] + + +@pytest.fixture +def hiseq_x_dual_index_bcl_convert_lims_samples( + hiseq_x_dual_index_flow_cell_dir: Path, +) -> list[FlowCellSampleBCLConvert]: + """Return a list of BCLConvert samples from a HiSeqX dual index flow cell.""" + path = Path(hiseq_x_dual_index_flow_cell_dir, f"HL32LCCXY_bcl_convert_raw{FileExtensions.JSON}") + return [FlowCellSampleBCLConvert.model_validate(sample) for sample in read_json(path)] + + +@pytest.fixture +def hiseq_2500_dual_index_bcl_convert_lims_samples( + hiseq_2500_dual_index_flow_cell_dir: Path, +) -> list[FlowCellSampleBCLConvert]: + """Return a list of BCLConvert samples from a HiSeq2500 dual index flow cell.""" + path = Path(hiseq_2500_dual_index_flow_cell_dir, "HM2LNBCX2_bcl_convert_raw.json") + return [FlowCellSampleBCLConvert.model_validate(sample) for sample in read_json(path)] + + +@pytest.fixture +def hiseq_2500_custom_index_bcl_convert_lims_samples( + hiseq_2500_custom_index_flow_cell_dir: Path, +) -> list[FlowCellSampleBCLConvert]: + """Return a list of BCLConvert samples from a HiSeq2500 custom index flow cell.""" + path = Path(hiseq_2500_custom_index_flow_cell_dir, "HGYFNBCX2_bcl_convert_raw.json") + return [FlowCellSampleBCLConvert.model_validate(sample) for sample in read_json(path)] + + +@pytest.fixture +def lims_novaseq_samples_raw(lims_novaseq_samples_file: Path) -> list[dict]: + """Return a list of raw flow cell samples.""" + return ReadFile.get_content_from_file( + file_format=FileFormat.JSON, file_path=lims_novaseq_samples_file + ) + + +@pytest.fixture +def lims_novaseq_6000_sample_raw(lims_novaseq_6000_samples_file: Path) -> list[dict]: + """Return the list of raw samples from flow cell HVKJCDRXX.""" + return ReadFile.get_content_from_file( + file_format=FileFormat.JSON, file_path=lims_novaseq_6000_samples_file + ) diff --git a/tests/fixture_plugins/timestamp_fixtures.py b/tests/fixture_plugins/timestamp_fixtures.py new file mode 100644 index 0000000000..6e77e289bb --- /dev/null +++ b/tests/fixture_plugins/timestamp_fixtures.py @@ -0,0 +1,46 @@ +"""Timestamp fixtures.""" +from datetime import MAXYEAR, datetime, timedelta + +import pytest + + +@pytest.fixture(scope="session") +def old_timestamp() -> datetime: + """Return a time stamp in date time format.""" + return datetime(1900, 1, 1) + + +@pytest.fixture(scope="session") +def timestamp() -> datetime: + """Return a time stamp in date time format.""" + return datetime(2020, 5, 1) + + +@pytest.fixture(scope="session") +def later_timestamp() -> datetime: + """Return a time stamp in date time format.""" + return datetime(2020, 6, 1) + + +@pytest.fixture(scope="session") +def future_date() -> datetime: + """Return a distant date in the future for which no events happen later.""" + return datetime(MAXYEAR, 1, 1, 1, 1, 1) + + +@pytest.fixture(scope="session") +def timestamp_now() -> datetime: + """Return a time stamp of today's date in date time format.""" + return datetime.now() + + +@pytest.fixture(scope="session") +def timestamp_yesterday(timestamp_now: datetime) -> datetime: + """Return a time stamp of yesterday's date in date time format.""" + return timestamp_now - timedelta(days=1) + + +@pytest.fixture(scope="session") +def timestamp_in_2_weeks(timestamp_now: datetime) -> datetime: + """Return a time stamp 14 days ahead in time.""" + return timestamp_now + timedelta(days=14) From 828180ea577666c1c48fbe10fe38ad6ecee346c8 Mon Sep 17 00:00:00 2001 From: Sebastian Diaz Date: Wed, 10 Jan 2024 13:53:16 +0100 Subject: [PATCH 2/4] Made correct sample sheet name a constant --- cg/constants/demultiplexing.py | 1 + .../demultiplex_fixtures/path_fixtures.py | 10 +++++++--- 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/cg/constants/demultiplexing.py b/cg/constants/demultiplexing.py index 10cfe8d36e..2ff9ff2820 100644 --- a/cg/constants/demultiplexing.py +++ b/cg/constants/demultiplexing.py @@ -29,6 +29,7 @@ class DemultiplexingDirsAndFiles(StrEnum): RUN_PARAMETERS_PASCAL_CASE: str = "RunParameters.xml" RUN_PARAMETERS_CAMEL_CASE: str = "runParameters.xml" SAMPLE_SHEET_FILE_NAME: str = "SampleSheet.csv" + CORRECT_SAMPLE_SHEET: str = "CorrectSampleSheet.csv" UNALIGNED_DIR_NAME: str = "Unaligned" BCL2FASTQ_TILE_DIR_PATTERN: str = r"l\dt\d{2}" QUEUED_FOR_POST_PROCESSING: str = "post_processing_queued.txt" diff --git a/tests/fixture_plugins/demultiplex_fixtures/path_fixtures.py b/tests/fixture_plugins/demultiplex_fixtures/path_fixtures.py index 7629a2a27f..e6e3b8750f 100644 --- a/tests/fixture_plugins/demultiplex_fixtures/path_fixtures.py +++ b/tests/fixture_plugins/demultiplex_fixtures/path_fixtures.py @@ -208,7 +208,9 @@ def novaseq_6000_post_1_5_kits_flow_cell(tmp_flow_cells_directory: Path) -> Path def novaseq_6000_post_1_5_kits_correct_sample_sheet( novaseq_6000_post_1_5_kits_flow_cell: Path, ) -> Path: - return Path(novaseq_6000_post_1_5_kits_flow_cell, "CorrectSampleSheet.csv") + return Path( + novaseq_6000_post_1_5_kits_flow_cell, DemultiplexingDirsAndFiles.CORRECT_SAMPLE_SHEET + ) @pytest.fixture @@ -227,7 +229,9 @@ def novaseq_6000_pre_1_5_kits_flow_cell(tmp_flow_cells_directory: Path) -> Path: def novaseq_6000_pre_1_5_kits_correct_sample_sheet( novaseq_6000_pre_1_5_kits_flow_cell: Path, ) -> Path: - return Path(novaseq_6000_pre_1_5_kits_flow_cell, "CorrectSampleSheet.csv") + return Path( + novaseq_6000_pre_1_5_kits_flow_cell, DemultiplexingDirsAndFiles.CORRECT_SAMPLE_SHEET + ) @pytest.fixture @@ -242,7 +246,7 @@ def novaseq_x_flow_cell_directory(tmp_flow_cells_directory: Path) -> Path: @pytest.fixture def novaseq_x_correct_sample_sheet(novaseq_x_flow_cell_directory: Path) -> Path: - return Path(novaseq_x_flow_cell_directory, "CorrectSampleSheet.csv") + return Path(novaseq_x_flow_cell_directory, DemultiplexingDirsAndFiles.CORRECT_SAMPLE_SHEET) @pytest.fixture From 33720201a3148989c870f191a1b67c45d38d5e26 Mon Sep 17 00:00:00 2001 From: Sebastian Diaz Date: Wed, 10 Jan 2024 14:00:44 +0100 Subject: [PATCH 3/4] Fix merge conflicts --- tests/conftest.py | 11 ++--------- 1 file changed, 2 insertions(+), 9 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 682a0c3e12..12b16a06cb 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -16,13 +16,6 @@ from cg.apps.crunchy import CrunchyAPI from cg.apps.demultiplex.demultiplex_api import DemultiplexingAPI -from cg.apps.demultiplex.sample_sheet.sample_models import ( - FlowCellSampleBcl2Fastq, - FlowCellSampleBCLConvert, -) -from cg.apps.demultiplex.sample_sheet.sample_sheet_creator import ( - SampleSheetCreatorBCLConvert, -) from cg.apps.downsample.downsample import DownsampleAPI from cg.apps.gens import GensAPI from cg.apps.gt import GenotypeAPI @@ -393,7 +386,8 @@ def demultiplex_configs( "demultiplex": {"slurm": {"account": "test", "mail_user": "testuser@github.se"}}, } - + +@pytest.fixture def real_crunchy_api(crunchy_config) -> CrunchyAPI: return CrunchyAPI(crunchy_config) @@ -2287,7 +2281,6 @@ def rnafusion_context( case_id_not_enough_reads: str, sample_id_not_enough_reads: str, total_sequenced_reads_not_pass: int, - timestamp_yesterday: datetime, ) -> CGConfig: """context to use in cli""" cg_context.housekeeper_api_ = nf_analysis_housekeeper From 9ea970e932e6c5baf18ff6a892a18f787df219e2 Mon Sep 17 00:00:00 2001 From: Sebastian Diaz Date: Thu, 11 Jan 2024 09:44:41 +0100 Subject: [PATCH 4/4] Moved constant for Correct sample sheet to tests --- cg/constants/demultiplexing.py | 1 - .../demultiplex_fixtures/path_fixtures.py | 12 +++++------- 2 files changed, 5 insertions(+), 8 deletions(-) diff --git a/cg/constants/demultiplexing.py b/cg/constants/demultiplexing.py index 2ff9ff2820..10cfe8d36e 100644 --- a/cg/constants/demultiplexing.py +++ b/cg/constants/demultiplexing.py @@ -29,7 +29,6 @@ class DemultiplexingDirsAndFiles(StrEnum): RUN_PARAMETERS_PASCAL_CASE: str = "RunParameters.xml" RUN_PARAMETERS_CAMEL_CASE: str = "runParameters.xml" SAMPLE_SHEET_FILE_NAME: str = "SampleSheet.csv" - CORRECT_SAMPLE_SHEET: str = "CorrectSampleSheet.csv" UNALIGNED_DIR_NAME: str = "Unaligned" BCL2FASTQ_TILE_DIR_PATTERN: str = r"l\dt\d{2}" QUEUED_FOR_POST_PROCESSING: str = "post_processing_queued.txt" diff --git a/tests/fixture_plugins/demultiplex_fixtures/path_fixtures.py b/tests/fixture_plugins/demultiplex_fixtures/path_fixtures.py index e6e3b8750f..807c34ee41 100644 --- a/tests/fixture_plugins/demultiplex_fixtures/path_fixtures.py +++ b/tests/fixture_plugins/demultiplex_fixtures/path_fixtures.py @@ -8,6 +8,8 @@ from cg.constants.nanopore_files import NanoporeDirsAndFiles from cg.models.flow_cell.flow_cell import FlowCellDirectoryData +CORRECT_SAMPLE_SHEET: str = "CorrectSampleSheet.csv" + @pytest.fixture(name="tmp_flow_cells_directory") def tmp_flow_cells_directory(tmp_path: Path, flow_cells_dir: Path) -> Path: @@ -208,9 +210,7 @@ def novaseq_6000_post_1_5_kits_flow_cell(tmp_flow_cells_directory: Path) -> Path def novaseq_6000_post_1_5_kits_correct_sample_sheet( novaseq_6000_post_1_5_kits_flow_cell: Path, ) -> Path: - return Path( - novaseq_6000_post_1_5_kits_flow_cell, DemultiplexingDirsAndFiles.CORRECT_SAMPLE_SHEET - ) + return Path(novaseq_6000_post_1_5_kits_flow_cell, CORRECT_SAMPLE_SHEET) @pytest.fixture @@ -229,9 +229,7 @@ def novaseq_6000_pre_1_5_kits_flow_cell(tmp_flow_cells_directory: Path) -> Path: def novaseq_6000_pre_1_5_kits_correct_sample_sheet( novaseq_6000_pre_1_5_kits_flow_cell: Path, ) -> Path: - return Path( - novaseq_6000_pre_1_5_kits_flow_cell, DemultiplexingDirsAndFiles.CORRECT_SAMPLE_SHEET - ) + return Path(novaseq_6000_pre_1_5_kits_flow_cell, CORRECT_SAMPLE_SHEET) @pytest.fixture @@ -246,7 +244,7 @@ def novaseq_x_flow_cell_directory(tmp_flow_cells_directory: Path) -> Path: @pytest.fixture def novaseq_x_correct_sample_sheet(novaseq_x_flow_cell_directory: Path) -> Path: - return Path(novaseq_x_flow_cell_directory, DemultiplexingDirsAndFiles.CORRECT_SAMPLE_SHEET) + return Path(novaseq_x_flow_cell_directory, CORRECT_SAMPLE_SHEET) @pytest.fixture