diff --git a/tests/conftest.py b/tests/conftest.py index 470cb447ad..12b16a06cb 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 @@ -16,29 +16,21 @@ 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 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 @@ -50,11 +42,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 @@ -76,51 +63,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 @@ -384,6 +334,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 real_crunchy_api(crunchy_config) -> CrunchyAPI: return CrunchyAPI(crunchy_config) @@ -423,9 +426,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.""" @@ -514,1325 +554,272 @@ def sv_research_vcf_file() -> str: # Common file fixtures @pytest.fixture(scope="session") -def fixtures_dir() -> Path: - """Return the path to the fixtures dir.""" - return Path("tests", "fixtures") - - -@pytest.fixture(scope="session") -def analysis_dir(fixtures_dir: Path) -> Path: - """Return the path to the analysis dir.""" - return Path(fixtures_dir, "analysis") - - -@pytest.fixture(scope="session") -def microsalt_analysis_dir(analysis_dir: Path) -> Path: - """Return the path to the analysis dir.""" - return Path(analysis_dir, "microsalt") - - -@pytest.fixture(scope="session") -def apps_dir(fixtures_dir: Path) -> Path: - """Return the path to the apps dir.""" - return Path(fixtures_dir, "apps") - - -@pytest.fixture(scope="session") -def cgweb_orders_dir(fixtures_dir: Path) -> Path: - """Return the path to the cgweb_orders dir.""" - return Path(fixtures_dir, "cgweb_orders") - - -@pytest.fixture(scope="session") -def data_dir(fixtures_dir: Path) -> Path: - """Return the path to the data dir.""" - return Path(fixtures_dir, "data") - - -@pytest.fixture -def fastq_dir(demultiplex_fixtures: Path) -> Path: - """Return the path to the fastq files dir.""" - return Path(demultiplex_fixtures, "fastq") - - -@pytest.fixture -def spring_dir(demultiplex_fixtures: Path) -> Path: - """Return the path to the fastq files dir.""" - return Path(demultiplex_fixtures, "spring") - - -@pytest.fixture -def project_dir(tmpdir_factory) -> Generator[Path, None, None]: - """Path to a temporary directory where intermediate files can be stored.""" - yield Path(tmpdir_factory.mktemp("data")) - - -@pytest.fixture -def tmp_file(project_dir) -> Path: - """Return a temp file path.""" - return Path(project_dir, "test") - - -@pytest.fixture -def non_existing_file_path(project_dir: Path) -> Path: - """Return the path to a non-existing file.""" - return Path(project_dir, "a_file.txt") - - -@pytest.fixture(scope="session") -def content() -> str: - """Return some content for a file.""" - return ( - "Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt" - " ut labore et dolore magna aliqua. Ut enim ad minim veniam, quis nostrud exercitation ull" - "amco laboris nisi ut aliquip ex ea commodo consequat. Duis aute irure dolor in reprehende" - "rit in voluptate velit esse cillum dolore eu fugiat nulla pariatur. Excepteur sint occaec" - "at cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id est laborum." - ) - - -@pytest.fixture -def filled_file(non_existing_file_path: Path, content: str) -> Path: - """Return the path to a existing file with some content.""" - with open(non_existing_file_path, "w") as outfile: - outfile.write(content) - return non_existing_file_path - - -@pytest.fixture(scope="session") -def orderforms(fixtures_dir: Path) -> Path: - """Return the path to the directory with order forms.""" - return Path(fixtures_dir, "orderforms") - - -@pytest.fixture -def hk_file(filled_file: Path, case_id: str) -> File: - """Return a housekeeper File object.""" - return File(id=case_id, path=filled_file.as_posix()) - - -@pytest.fixture -def mip_dna_store_files(apps_dir: Path) -> Path: - """Return the path to the directory with mip dna store files.""" - return Path(apps_dir, "mip", "dna", "store") - - -@pytest.fixture -def case_qc_sample_info_path(mip_dna_store_files: Path) -> Path: - """Return path to case_qc_sample_info.yaml.""" - return Path(mip_dna_store_files, "case_qc_sample_info.yaml") - - -@pytest.fixture -def delivery_report_html(mip_dna_store_files: Path) -> Path: - """Return the path to a qc metrics deliverables file with case data.""" - return Path(mip_dna_store_files, "empty_delivery_report.html") - - -@pytest.fixture -def mip_deliverables_file(mip_dna_store_files: Path) -> Path: - """Fixture for general deliverables file in mip.""" - return Path(mip_dna_store_files, "case_id_deliverables.yaml") - - -@pytest.fixture -def case_qc_metrics_deliverables(apps_dir: Path) -> Path: - """Return the path to a qc metrics deliverables file with case data.""" - return Path(apps_dir, "mip", "case_metrics_deliverables.yaml") - - -@pytest.fixture -def mip_analysis_dir(analysis_dir: Path) -> Path: - """Return the path to the directory with mip analysis files.""" - return Path(analysis_dir, "mip") - - -@pytest.fixture -def balsamic_analysis_dir(analysis_dir: Path) -> Path: - """Return the path to the directory with balsamic analysis files.""" - return Path(analysis_dir, "balsamic") - - -@pytest.fixture -def balsamic_wgs_analysis_dir(balsamic_analysis_dir: Path) -> Path: - """Return the path to the directory with balsamic analysis files.""" - return Path(balsamic_analysis_dir, "tn_wgs") - - -@pytest.fixture -def mip_dna_analysis_dir(mip_analysis_dir: Path) -> Path: - """Return the path to the directory with mip dna analysis files.""" - return Path(mip_analysis_dir, "dna") - - -@pytest.fixture -def rnafusion_analysis_dir(analysis_dir: Path) -> Path: - """Return the path to the directory with rnafusion analysis files.""" - return Path(analysis_dir, "rnafusion") - - -@pytest.fixture -def sample_cram(mip_dna_analysis_dir: Path) -> Path: - """Return the path to the cram file for a sample.""" - return Path(mip_dna_analysis_dir, "adm1.cram") - - -@pytest.fixture(name="father_sample_cram") -def father_sample_cram( - mip_dna_analysis_dir: Path, - father_sample_id: str, -) -> Path: - """Return the path to the cram file for the father sample.""" - return Path(mip_dna_analysis_dir, father_sample_id + FileExtensions.CRAM) - - -@pytest.fixture(name="mother_sample_cram") -def mother_sample_cram(mip_dna_analysis_dir: Path, mother_sample_id: str) -> Path: - """Return the path to the cram file for the mother sample.""" - return Path(mip_dna_analysis_dir, mother_sample_id + FileExtensions.CRAM) - - -@pytest.fixture(name="sample_cram_files") -def sample_crams( - sample_cram: Path, father_sample_cram: Path, mother_sample_cram: Path -) -> list[Path]: - """Return a list of cram paths for three samples.""" - return [sample_cram, father_sample_cram, mother_sample_cram] - - -@pytest.fixture(name="vcf_file") -def vcf_file(mip_dna_store_files: Path) -> Path: - """Return the path to a VCF file.""" - return Path(mip_dna_store_files, "yellowhog_clinical_selected.vcf") - - -@pytest.fixture(name="fastq_file") -def fastq_file(fastq_dir: Path) -> Path: - """Return the path to a FASTQ file.""" - return Path(fastq_dir, "dummy_run_R1_001.fastq.gz") - - -@pytest.fixture(name="fastq_file_father") -def fastq_file_father(fastq_dir: Path) -> Path: - """Return the path to a FASTQ file.""" - return Path(fastq_dir, "fastq_run_R1_001.fastq.gz") - - -@pytest.fixture(name="spring_file") -def spring_file(spring_dir: Path) -> Path: - """Return the path to an existing spring file.""" - return Path(spring_dir, "dummy_run_001.spring") - - -@pytest.fixture(name="spring_meta_data_file") -def spring_meta_data_file(spring_dir: Path) -> Path: - """Return the path to an existing spring file.""" - return Path(spring_dir, "dummy_spring_meta_data.json") - - -@pytest.fixture(name="spring_file_father") -def spring_file_father(spring_dir: Path) -> Path: - """Return the path to a second existing spring file.""" - return Path(spring_dir, "dummy_run_002.spring") - - -@pytest.fixture(name="madeline_output") -def madeline_output(apps_dir: Path) -> Path: - """Return str of path for file with Madeline output.""" - return Path(apps_dir, "madeline", "madeline.xml") - - -@pytest.fixture(name="file_does_not_exist") -def file_does_not_exist() -> Path: - """Return a file path that does not exist.""" - return Path("file", "does", "not", "exist") - - -# Compression fixtures - - -@pytest.fixture(name="run_name") -def run_name() -> str: - """Return the name of a fastq run.""" - return "fastq_run" - - -@pytest.fixture(name="original_fastq_data") -def original_fastq_data(fastq_dir: Path, run_name) -> CompressionData: - """Return a compression object with a path to the original fastq files.""" - return CompressionData(Path(fastq_dir, run_name)) - - -@pytest.fixture(name="fastq_stub") -def fastq_stub(project_dir: Path, run_name: str) -> Path: - """Creates a path to the base format of a fastq run.""" - return Path(project_dir, run_name) - - -@pytest.fixture(name="compression_object") -def compression_object(fastq_stub: Path, original_fastq_data: CompressionData) -> CompressionData: - """Creates compression data object with information about files used in fastq compression.""" - working_files: CompressionData = CompressionData(fastq_stub) - working_file_map: dict[str, str] = { - original_fastq_data.fastq_first.as_posix(): working_files.fastq_first.as_posix(), - original_fastq_data.fastq_second.as_posix(): working_files.fastq_second.as_posix(), - } - for original_file, working_file in working_file_map.items(): - shutil.copy(original_file, working_file) - return working_files - - -# Demultiplex fixtures - - -@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(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_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 -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_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") +def fixtures_dir() -> Path: + """Return the path to the fixtures dir.""" + return Path("tests", "fixtures") @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 - ) +def analysis_dir(fixtures_dir: Path) -> Path: + """Return the path to the analysis dir.""" + return Path(fixtures_dir, "analysis") @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 - ) +def microsalt_analysis_dir(analysis_dir: Path) -> Path: + """Return the path to the analysis dir.""" + return Path(analysis_dir, "microsalt") @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 - ) +def apps_dir(fixtures_dir: Path) -> Path: + """Return the path to the apps dir.""" + return Path(fixtures_dir, "apps") @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 - ) +def cgweb_orders_dir(fixtures_dir: Path) -> Path: + """Return the path to the cgweb_orders dir.""" + return Path(fixtures_dir, "cgweb_orders") @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) +def data_dir(fixtures_dir: Path) -> Path: + """Return the path to the data dir.""" + return Path(fixtures_dir, "data") @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, - ) +def fastq_dir(demultiplex_fixtures: Path) -> Path: + """Return the path to the fastq files dir.""" + return Path(demultiplex_fixtures, "fastq") @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, - ) +def spring_dir(demultiplex_fixtures: Path) -> Path: + """Return the path to the fastq files dir.""" + return Path(demultiplex_fixtures, "spring") -@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 +def project_dir(tmpdir_factory) -> Generator[Path, None, None]: + """Path to a temporary directory where intermediate files can be stored.""" + yield Path(tmpdir_factory.mktemp("data")) -@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 +def tmp_file(project_dir) -> Path: + """Return a temp file path.""" + return Path(project_dir, "test") -@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 +def non_existing_file_path(project_dir: Path) -> Path: + """Return the path to a non-existing file.""" + return Path(project_dir, "a_file.txt") -@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 content() -> str: + """Return some content for a file.""" + return ( + "Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt" + " ut labore et dolore magna aliqua. Ut enim ad minim veniam, quis nostrud exercitation ull" + "amco laboris nisi ut aliquip ex ea commodo consequat. Duis aute irure dolor in reprehende" + "rit in voluptate velit esse cillum dolore eu fugiat nulla pariatur. Excepteur sint occaec" + "at cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id est laborum." + ) -@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 +def filled_file(non_existing_file_path: Path, content: str) -> Path: + """Return the path to a existing file with some content.""" + with open(non_existing_file_path, "w") as outfile: + outfile.write(content) + return non_existing_file_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) +def orderforms(fixtures_dir: Path) -> Path: + """Return the path to the directory with order forms.""" + return Path(fixtures_dir, "orderforms") -@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 +def hk_file(filled_file: Path, case_id: str) -> File: + """Return a housekeeper File object.""" + return File(id=case_id, path=filled_file.as_posix()) -@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 +def mip_dna_store_files(apps_dir: Path) -> Path: + """Return the path to the directory with mip dna store files.""" + return Path(apps_dir, "mip", "dna", "store") -@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 +def case_qc_sample_info_path(mip_dna_store_files: Path) -> Path: + """Return path to case_qc_sample_info.yaml.""" + return Path(mip_dna_store_files, "case_qc_sample_info.yaml") -@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 delivery_report_html(mip_dna_store_files: Path) -> Path: + """Return the path to a qc metrics deliverables file with case data.""" + return Path(mip_dna_store_files, "empty_delivery_report.html") @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 - ) +def mip_deliverables_file(mip_dna_store_files: Path) -> Path: + """Fixture for general deliverables file in mip.""" + return Path(mip_dna_store_files, "case_id_deliverables.yaml") @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 - ) +def case_qc_metrics_deliverables(apps_dir: Path) -> Path: + """Return the path to a qc metrics deliverables file with case data.""" + return Path(apps_dir, "mip", "case_metrics_deliverables.yaml") -@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 +def mip_analysis_dir(analysis_dir: Path) -> Path: + """Return the path to the directory with mip analysis files.""" + return Path(analysis_dir, "mip") -@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 +def balsamic_analysis_dir(analysis_dir: Path) -> Path: + """Return the path to the directory with balsamic analysis files.""" + return Path(analysis_dir, "balsamic") -@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 +def balsamic_wgs_analysis_dir(balsamic_analysis_dir: Path) -> Path: + """Return the path to the directory with balsamic analysis files.""" + return Path(balsamic_analysis_dir, "tn_wgs") -@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 +def mip_dna_analysis_dir(mip_analysis_dir: Path) -> Path: + """Return the path to the directory with mip dna analysis files.""" + return Path(mip_analysis_dir, "dna") -@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 rnafusion_analysis_dir(analysis_dir: Path) -> Path: + """Return the path to the directory with rnafusion analysis files.""" + return Path(analysis_dir, "rnafusion") -@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 +def sample_cram(mip_dna_analysis_dir: Path) -> Path: + """Return the path to the cram file for a sample.""" + return Path(mip_dna_analysis_dir, "adm1.cram") -@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(name="father_sample_cram") +def father_sample_cram( + mip_dna_analysis_dir: Path, + father_sample_id: str, +) -> Path: + """Return the path to the cram file for the father sample.""" + return Path(mip_dna_analysis_dir, father_sample_id + FileExtensions.CRAM) -@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(name="mother_sample_cram") +def mother_sample_cram(mip_dna_analysis_dir: Path, mother_sample_id: str) -> Path: + """Return the path to the cram file for the mother sample.""" + return Path(mip_dna_analysis_dir, mother_sample_id + FileExtensions.CRAM) -@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(name="sample_cram_files") +def sample_crams( + sample_cram: Path, father_sample_cram: Path, mother_sample_cram: Path +) -> list[Path]: + """Return a list of cram paths for three samples.""" + return [sample_cram, father_sample_cram, mother_sample_cram] -@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(name="vcf_file") +def vcf_file(mip_dna_store_files: Path) -> Path: + """Return the path to a VCF file.""" + return Path(mip_dna_store_files, "yellowhog_clinical_selected.vcf") -@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(name="fastq_file") +def fastq_file(fastq_dir: Path) -> Path: + """Return the path to a FASTQ file.""" + return Path(fastq_dir, "dummy_run_R1_001.fastq.gz") -@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="fastq_file_father") +def fastq_file_father(fastq_dir: Path) -> Path: + """Return the path to a FASTQ file.""" + return Path(fastq_dir, "fastq_run_R1_001.fastq.gz") -@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="spring_file") +def spring_file(spring_dir: Path) -> Path: + """Return the path to an existing spring file.""" + return Path(spring_dir, "dummy_run_001.spring") -@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="spring_meta_data_file") +def spring_meta_data_file(spring_dir: Path) -> Path: + """Return the path to an existing spring file.""" + return Path(spring_dir, "dummy_spring_meta_data.json") -@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(name="spring_file_father") +def spring_file_father(spring_dir: Path) -> Path: + """Return the path to a second existing spring file.""" + return Path(spring_dir, "dummy_run_002.spring") -@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="madeline_output") +def madeline_output(apps_dir: Path) -> Path: + """Return str of path for file with Madeline output.""" + return Path(apps_dir, "madeline", "madeline.xml") -@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(name="file_does_not_exist") +def file_does_not_exist() -> Path: + """Return a file path that does not exist.""" + return Path("file", "does", "not", "exist") -@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 - ) +# Compression fixtures -@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="run_name") +def run_name() -> str: + """Return the name of a fastq run.""" + return "fastq_run" -@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="original_fastq_data") +def original_fastq_data(fastq_dir: Path, run_name) -> CompressionData: + """Return a compression object with a path to the original fastq files.""" + return CompressionData(Path(fastq_dir, run_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(name="fastq_stub") +def fastq_stub(project_dir: Path, run_name: str) -> Path: + """Creates a path to the base format of a fastq run.""" + return Path(project_dir, run_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 +@pytest.fixture(name="compression_object") +def compression_object(fastq_stub: Path, original_fastq_data: CompressionData) -> CompressionData: + """Creates compression data object with information about files used in fastq compression.""" + working_files: CompressionData = CompressionData(fastq_stub) + working_file_map: dict[str, str] = { + original_fastq_data.fastq_first.as_posix(): working_files.fastq_first.as_posix(), + original_fastq_data.fastq_second.as_posix(): working_files.fastq_second.as_posix(), + } + for original_file, working_file in working_file_map.items(): + shutil.copy(original_file, working_file) + return working_files # Genotype file fixture @@ -2143,6 +1130,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.""" @@ -3260,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 @@ -3484,6 +2504,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..807c34ee41 --- /dev/null +++ b/tests/fixture_plugins/demultiplex_fixtures/path_fixtures.py @@ -0,0 +1,458 @@ +"""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 + +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: + """ + 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, CORRECT_SAMPLE_SHEET) + + +@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, CORRECT_SAMPLE_SHEET) + + +@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, CORRECT_SAMPLE_SHEET) + + +@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)