Skip to content

Commit

Permalink
remove duplicated fastq and rml order fixtures
Browse files Browse the repository at this point in the history
  • Loading branch information
diitaz93 committed Jan 10, 2025
1 parent 6014245 commit ceb17a1
Show file tree
Hide file tree
Showing 2 changed files with 17 additions and 91 deletions.
74 changes: 1 addition & 73 deletions tests/services/order_validation_service/sample_rules/conftest.py
Original file line number Diff line number Diff line change
@@ -1,21 +1,15 @@
import pytest

from cg.models.orders.constants import OrderType
from cg.models.orders.sample_base import ContainerEnum, PriorityEnum, SexEnum
from cg.models.orders.sample_base import ContainerEnum, PriorityEnum
from cg.services.order_validation_service.constants import (
MINIMUM_VOLUME,
ElutionBuffer,
ExtractionMethod,
)
from cg.services.order_validation_service.workflows.fastq.constants import FastqDeliveryType
from cg.services.order_validation_service.workflows.fastq.models.order import FastqOrder
from cg.services.order_validation_service.workflows.fastq.models.sample import FastqSample
from cg.services.order_validation_service.workflows.microsalt.constants import MicrosaltDeliveryType
from cg.services.order_validation_service.workflows.microsalt.models.order import MicrosaltOrder
from cg.services.order_validation_service.workflows.microsalt.models.sample import MicrosaltSample
from cg.services.order_validation_service.workflows.rml.constants import RmlDeliveryType
from cg.services.order_validation_service.workflows.rml.models.order import RmlOrder
from cg.services.order_validation_service.workflows.rml.models.sample import RmlSample
from cg.store.models import Application
from cg.store.store import Store

Expand Down Expand Up @@ -82,69 +76,3 @@ def order_with_samples_in_same_well() -> MicrosaltOrder:
sample_1: MicrosaltSample = create_microsalt_sample(1)
sample_2: MicrosaltSample = create_microsalt_sample(1)
return create_microsalt_order([sample_1, sample_2])


def create_fastq_sample(id: int) -> FastqSample:
return FastqSample(
application="WGSPCFC030",
comment="",
container=ContainerEnum.tube,
container_name="Fastq tube",
name=f"fastq-sample-{id}",
volume=54,
concentration_ng_ul=30,
elution_buffer=ElutionBuffer.WATER,
priority=PriorityEnum.priority,
quantity=54,
require_qc_ok=True,
sex=SexEnum.male,
source="blood",
subject_id=f"fastq-subject-{id}",
)


@pytest.fixture
def valid_fastq_order() -> FastqOrder:
sample_1: FastqSample = create_fastq_sample(1)
sample_2: FastqSample = create_fastq_sample(2)
sample_3: FastqSample = create_fastq_sample(3)
samples = [sample_1, sample_2, sample_3]
return FastqOrder(
customer="cust000",
project_type=OrderType.FASTQ,
user_id=0,
delivery_type=FastqDeliveryType.FASTQ,
samples=samples,
name="FastqOrder",
)


def create_rml_sample(id: int) -> RmlSample:
return RmlSample(
application="RMLCUSR800",
index_number=4,
index="Kapa UDI NIPT",
rml_plate_name="Rml-plate",
name=f"rml-sample-{id}",
pool="pool-2",
pool_concentration=30,
volume=54,
priority=PriorityEnum.priority,
well_position_rml=f"A:{id}",
)


@pytest.fixture
def valid_rml_order() -> RmlOrder:
sample_1: RmlSample = create_rml_sample(1)
sample_2: RmlSample = create_rml_sample(2)
sample_3: RmlSample = create_rml_sample(3)
samples = [sample_1, sample_2, sample_3]
return RmlOrder(
customer="cust000",
project_type=OrderType.RML,
user_id=0,
delivery_type=RmlDeliveryType.FASTQ,
samples=samples,
name="RmlOrder",
)
Original file line number Diff line number Diff line change
Expand Up @@ -81,13 +81,13 @@ def test_validate_well_position_format(valid_microsalt_order: MicrosaltOrder):
assert errors[0].sample_index == 0


def test_validate_well_position_rml_format(valid_rml_order: RmlOrder):
def test_validate_well_position_rml_format(rml_order: RmlOrder):

# GIVEN an order with a sample with an invalid well position
valid_rml_order.samples[0].well_position_rml = "J:4"
# GIVEN a RML order with a sample with an invalid well position
rml_order.samples[0].well_position_rml = "J:4"

# WHEN validating the well position format
errors = validate_well_position_rml_format(order=valid_rml_order)
errors = validate_well_position_rml_format(order=rml_order)

# THEN an error should be returned
assert errors
Expand Down Expand Up @@ -179,17 +179,17 @@ def test_non_required_sample_volume(valid_microsalt_order: MicrosaltOrder):
assert not errors


def test_validate_concentration_required_if_skip_rc(valid_fastq_order: FastqOrder):
def test_validate_concentration_required_if_skip_rc(fastq_order: FastqOrder):

# GIVEN a fastq order trying to skip reception control
valid_fastq_order.skip_reception_control = True
fastq_order.skip_reception_control = True

# GIVEN that one of its samples has no concentration set
valid_fastq_order.samples[0].concentration_ng_ul = None
fastq_order.samples[0].concentration_ng_ul = None

# WHEN validating that the concentration is not missing
errors: list[ConcentrationRequiredError] = validate_concentration_required_if_skip_rc(
order=valid_fastq_order
order=fastq_order
)

# THEN an error should be returned
Expand All @@ -199,24 +199,22 @@ def test_validate_concentration_required_if_skip_rc(valid_fastq_order: FastqOrde
assert isinstance(errors[0], ConcentrationRequiredError)


def test_validate_concentration_interval_if_skip_rc(
valid_fastq_order: FastqOrder, base_store: Store
):
def test_validate_concentration_interval_if_skip_rc(fastq_order: FastqOrder, base_store: Store):

# GIVEN a Fastq order trying to skip reception control
valid_fastq_order.skip_reception_control = True
fastq_order.skip_reception_control = True

# GIVEN that one of the samples has a concentration outside the allowed interval for its
# application
sample = valid_fastq_order.samples[0]
sample = fastq_order.samples[0]
application = base_store.get_application_by_tag(sample.application)
application.sample_concentration_minimum = sample.concentration_ng_ul + 1
base_store.session.add(application)
base_store.commit_to_store()

# WHEN validating that the order's samples' concentrations are within allowed intervals
errors: list[ConcentrationInvalidIfSkipRCError] = validate_concentration_interval_if_skip_rc(
order=valid_fastq_order, store=base_store
order=fastq_order, store=base_store
)

# THEN an error should be returned
Expand All @@ -226,16 +224,16 @@ def test_validate_concentration_interval_if_skip_rc(
assert isinstance(errors[0], ConcentrationInvalidIfSkipRCError)


def test_validate_buffer_skip_rc_condition(valid_fastq_order: FastqOrder):
def test_validate_buffer_skip_rc_condition(fastq_order: FastqOrder):

# GIVEN a Fastq order trying to skip reception control
valid_fastq_order.skip_reception_control = True
fastq_order.skip_reception_control = True

# GIVEN that one of the samples has buffer specified as 'other'
valid_fastq_order.samples[0].elution_buffer = ElutionBuffer.OTHER
fastq_order.samples[0].elution_buffer = ElutionBuffer.OTHER

# WHEN validating that the buffers follow the 'skip reception control' requirements
errors: list[BufferInvalidError] = validate_buffer_skip_rc_condition(order=valid_fastq_order)
errors: list[BufferInvalidError] = validate_buffer_skip_rc_condition(order=fastq_order)

# THEN an error should be returned
assert errors
Expand Down

0 comments on commit ceb17a1

Please sign in to comment.