Add validation for concentration intervals

cg/services/order_validation_service/models/errors.py

@@ -155,6 +155,17 @@ class SubjectIdSameAsSampleNameError(CaseSampleError):
     message: str = "Subject id must be different from the sample name"
 
 
+class InvalidConcentrationIfSkipRCError(CaseSampleError):
+    def __init__(self, case_name: str, sample_name: str, allowed_interval: tuple[int, int]):
+        field: str = "concentration_ng_ul"
+        message: str = (
+            f"Concentration must be between {allowed_interval[0]} ng/μL and {allowed_interval[1]} ng/μL"
+        )
+        super(CaseSampleError, self).__init__(
+            case_name=case_name, sample_name=sample_name, field=field, message=message
+        )
+
+
 class WellPositionMissingError(CaseSampleError):
     field: str = "well_position"
     message: str = "Well position is required for well plates"

cg/services/order_validation_service/workflows/tomte/validation/inter_field/rules.py

@@ -1,5 +1,6 @@
 from cg.services.order_validation_service.models.errors import (
     FatherNotInCaseError,
+    InvalidConcentrationIfSkipRCError,
     InvalidFatherSexError,
     InvalidMotherSexError,
     MotherNotInCaseError,
@@ -23,8 +24,10 @@
     get_mother_sex_errors,
     get_repeated_case_name_errors,
     get_repeated_sample_name_errors,
+    validate_concentration_in_case,
     validate_subject_ids_in_case,
 )
+from cg.store.store import Store
 
 
 def validate_wells_contain_at_most_one_sample(order: TomteOrder) -> list[OccupiedWellError]:
@@ -93,3 +96,15 @@ def validate_subject_ids_different_from_case_names(
         case_errors = validate_subject_ids_in_case(case)
         errors.extend(case_errors)
     return errors
+
+
+def validate_concentration_interval_if_skip_rc(
+    order: TomteOrder, store: Store
+) -> list[InvalidConcentrationIfSkipRCError]:
+    if not order.skip_reception_control:
+        return []
+    errors = []
+    for case in order.cases:
+        case_errors = validate_concentration_in_case(case=case, store=store)
+        errors.extend(case_errors)
+    return errors

cg/services/order_validation_service/workflows/tomte/validation/inter_field/utils.py

@@ -1,9 +1,11 @@
 from collections import Counter
 
+from cg.constants.sample_sources import SourceType
 from cg.constants.subject import Sex
 from cg.models.orders.sample_base import ContainerEnum
 from cg.services.order_validation_service.models.errors import (
     FatherNotInCaseError,
+    InvalidConcentrationIfSkipRCError,
     InvalidFatherSexError,
     InvalidMotherSexError,
     MotherNotInCaseError,
@@ -17,6 +19,8 @@
 from cg.services.order_validation_service.workflows.tomte.models.sample import (
     TomteSample,
 )
+from cg.store.models import Application
+from cg.store.store import Store
 
 
 def _get_errors(colliding_samples: list[tuple[TomteSample, TomteCase]]) -> list[OccupiedWellError]:
@@ -158,3 +162,53 @@ def validate_subject_ids_in_case(case: TomteCase) -> list[SubjectIdSameAsCaseNam
             error = SubjectIdSameAsCaseNameError(case_name=case.name, sample_name=sample.name)
             errors.append(error)
     return errors
+
+
+def validate_concentration_in_case(case: TomteCase, store: Store):
+    errors = []
+    for sample in case.samples:
+        if has_sample_invalid_concentration(sample=sample, store=store):
+            error = create_invalid_concentration_error(
+                case_name=case.name, sample=sample, store=store
+            )
+            errors.append(error)
+    return errors
+
+
+def create_invalid_concentration_error(case_name: str, sample: TomteSample, store: Store):
+    application: Application = store.get_application_by_tag(sample.application)
+    is_cfdna = is_sample_cfdna(sample)
+    allowed_interval = get_concentration_interval(application=application, is_cfdna=is_cfdna)
+    return InvalidConcentrationIfSkipRCError(
+        case_name=case_name, sample_name=sample.name, allowed_interval=allowed_interval
+    )
+
+
+def has_sample_invalid_concentration(sample: TomteSample, store: Store) -> bool:
+    application: Application | None = store.get_application_by_tag(sample.application)
+    return not is_sample_concentration_allowed(sample=sample, application=application)
+
+
+def is_sample_concentration_allowed(sample: TomteSample, application: Application):
+    concentration = sample.concentration_ng_ul
+    is_cfdna = is_sample_cfdna(sample)
+    interval = get_concentration_interval(application=application, is_cfdna=is_cfdna)
+    return is_sample_concentration_within_interval(concentration=concentration, interval=interval)
+
+
+def is_sample_cfdna(sample: TomteSample):
+    source = sample.source
+    return source == SourceType.CELL_FREE_DNA
+
+
+def get_concentration_interval(application: Application, is_cfdna: bool) -> tuple[int, int]:
+    if is_cfdna:
+        return (
+            application.sample_concentration_minimum_cfdna,
+            application.sample_concentration_maximum_cfdna,
+        )
+    return application.sample_concentration_minimum, application.sample_concentration_maximum
+
+
+def is_sample_concentration_within_interval(concentration: float, interval: tuple[int, int]):
+    return interval[0] <= concentration <= interval[1]

cg/services/order_validation_service/workflows/tomte/validation_rules.py

@@ -16,6 +16,7 @@
 )
 from cg.services.order_validation_service.workflows.tomte.validation.inter_field.rules import (
     validate_case_names_not_repeated,
+    validate_concentration_interval_if_skip_rc,
     validate_fathers_are_male,
     validate_fathers_in_same_case_as_children,
     validate_mothers_are_female,
@@ -44,6 +45,7 @@
     validate_application_not_archived,
     validate_buffer_skip_rc_condition,
     validate_case_names_not_repeated,
+    validate_concentration_interval_if_skip_rc,
     validate_concentration_required_if_skip_rc,
     validate_fathers_are_male,
     validate_fathers_in_same_case_as_children,

tests/services/order_validation_service/conftest.py

@@ -183,13 +183,41 @@ def order_with_siblings_as_parents():
     return create_order([case])
 
 
+@pytest.fixture
+def sample_with_invalid_concentration():
+    sample: TomteSample = create_sample(1)
+    sample.concentration_ng_ul = 1
+    return sample
+
+
 @pytest.fixture
 def sample_with_missing_well_position():
     sample = create_sample(1)
     sample.well_position = None
     return sample
 
 
+@pytest.fixture
+def application_with_concentration_interval(base_store: Store) -> Application:
+    return base_store.add_application(
+        tag="RNAPOAR100",
+        prep_category="wts",
+        description="This is an application with concentration interval",
+        percent_kth=100,
+        percent_reads_guaranteed=90,
+        sample_concentration_minimum=50,
+        sample_concentration_maximum=250,
+    )
+
+
+@pytest.fixture
+def order_with_invalid_concentration(sample_with_invalid_concentration) -> TomteOrder:
+    case = create_case([sample_with_invalid_concentration])
+    order = create_order([case])
+    order.skip_reception_control = True
+    return order
+
+
 @pytest.fixture
 def sample_with_missing_container_name() -> TomteSample:
     sample: TomteSample = create_sample(1)

tests/services/order_validation_service/test_tomte_inter_field_validators.py

@@ -3,6 +3,7 @@
     DescendantAsFatherError,
     FatherNotInCaseError,
     InvalidBufferError,
+    InvalidConcentrationIfSkipRCError,
     InvalidFatherSexError,
     OccupiedWellError,
     RepeatedCaseNameError,
@@ -12,20 +13,24 @@
 )
 from cg.services.order_validation_service.validators.inter_field.rules import (
     validate_buffers_are_allowed,
-    validate_subject_ids_different_from_sample_names,
-)
-from cg.services.order_validation_service.validators.inter_field.rules import (
     validate_concentration_required_if_skip_rc,
+    validate_subject_ids_different_from_sample_names,
 )
 from cg.services.order_validation_service.workflows.tomte.models.order import TomteOrder
+from cg.services.order_validation_service.workflows.tomte.models.sample import (
+    TomteSample,
+)
 from cg.services.order_validation_service.workflows.tomte.validation.inter_field.rules import (
     validate_case_names_not_repeated,
+    validate_concentration_interval_if_skip_rc,
     validate_fathers_are_male,
     validate_fathers_in_same_case_as_children,
     validate_pedigree,
     validate_sample_names_not_repeated,
     validate_wells_contain_at_most_one_sample,
 )
+from cg.store.models import Application
+from cg.store.store import Store
 
 
 def test_multiple_samples_in_well_not_allowed(order_with_samples_in_same_well: TomteOrder):
@@ -207,3 +212,27 @@ def test_concentration_required_if_skip_rc(valid_order: TomteOrder):
 
     # THEN the error should concern the missing concentration
     assert isinstance(errors[0], ConcentrationRequiredIfSkipRCError)
+
+
+def test_concentration_not_within_interval_if_skip_rc(
+    order_with_invalid_concentration: TomteOrder,
+    sample_with_invalid_concentration: TomteSample,
+    base_store: Store,
+    application_with_concentration_interval: Application,
+):
+
+    # GIVEN an order skipping reception control
+    # GIVEN that the order has a sample with invalid concentration for its application
+    base_store.session.add(application_with_concentration_interval)
+    base_store.session.commit()
+
+    # WHEN validating that the concentration is within the allowed interval
+    errors = validate_concentration_interval_if_skip_rc(
+        order=order_with_invalid_concentration, store=base_store
+    )
+
+    # THEN an error is returned
+    assert errors
+
+    # THEN the error should concern the application interval
+    assert isinstance(errors[0], InvalidConcentrationIfSkipRCError)