Fix clashing ticket ids

cg/store/base.py

@@ -6,7 +6,12 @@
 from sqlalchemy import and_, func
 from sqlalchemy.orm import Query, Session
 
-from cg.store.models import Analysis, Application, ApplicationLimitations, ApplicationVersion
+from cg.store.models import (
+    Analysis,
+    Application,
+    ApplicationLimitations,
+    ApplicationVersion,
+)
 from cg.store.models import Base as ModelBase
 from cg.store.models import (
     Case,
@@ -74,6 +79,12 @@ def _get_join_sample_family_query(self) -> Query:
         """Return a join sample case relationship query."""
         return self._get_query(table=Sample).join(Case.links).join(CaseSample.sample)
 
+    def _get_join_sample_case_order_query(self) -> Query:
+        """Return a query joining sample, cases_sample, case and order. Selects from sample."""
+        return (
+            self._get_query(table=Sample).join(Case.links).join(CaseSample.sample).join(Case.orders)
+        )
+
     def _get_join_sample_application_version_query(self) -> Query:
         """Return join sample to application version query."""
         return (

cg/store/crud/read.py

@@ -313,25 +313,26 @@ def filter_cases_with_samples(self, case_ids: list[str]) -> list[str]:
 
     def get_cases_by_ticket_id(self, ticket_id: str) -> list[Case]:
         """Return cases associated with a given ticket id."""
-        return apply_case_filter(
-            cases=self._get_query(table=Case),
-            filter_functions=[CaseFilter.BY_TICKET],
-            ticket_id=ticket_id,
-        ).all()
+        if order := apply_order_filters(
+            orders=self._get_query(table=Order),
+            filters=[OrderFilter.BY_TICKET_ID],
+            ticket_id=int(ticket_id),
+        ).first():
+            return order.cases
+        return []
 
     def get_customer_id_from_ticket(self, ticket: str) -> str:
         """Returns the customer related to given ticket."""
-        cases: list[Case] = self.get_cases_by_ticket_id(ticket_id=ticket)
-        if not cases:
-            raise ValueError(f"No case found for ticket {ticket}")
-        return cases[0].customer.internal_id
+        if order := self.get_order_by_ticket_id(int(ticket)):
+            return order.customer.internal_id
+        raise ValueError(f"No order found for ticket {ticket}")
 
     def get_samples_from_ticket(self, ticket: str) -> list[Sample]:
         """Returns the samples related to given ticket."""
-        return apply_case_filter(
-            cases=self._get_join_sample_family_query(),
-            filter_functions=[CaseFilter.BY_TICKET],
-            ticket_id=ticket,
+        return apply_order_filters(
+            orders=self._get_join_sample_case_order_query(),
+            filters=[OrderFilter.BY_TICKET_ID],
+            ticket_id=int(ticket),
         ).all()
 
     def get_latest_ticket_from_case(self, case_id: str) -> str:

cg/store/filters/status_case_filters.py

@@ -96,11 +96,6 @@ def filter_cases_by_priority(cases: Query, priority: str, **kwargs) -> Query:
     return cases.filter(Case.priority == priority)
 
 
-def filter_cases_by_ticket_id(cases: Query, ticket_id: str, **kwargs) -> Query:
-    """Filter cases with matching ticket id."""
-    return cases.filter(Case.tickets.contains(ticket_id))
-
-
 def filter_cases_for_analysis(cases: Query, **kwargs) -> Query:
     """Filter cases in need of analysis by:
     1. Action set to analyze or
@@ -289,7 +284,6 @@ class CaseFilter(Enum):
     BY_WORKFLOWS: Callable = filter_cases_by_workflows
     BY_WORKFLOW_SEARCH: Callable = filter_cases_by_workflow_search
     BY_PRIORITY: Callable = filter_cases_by_priority
-    BY_TICKET: Callable = filter_cases_by_ticket_id
     FOR_ANALYSIS: Callable = filter_cases_for_analysis
     HAS_INACTIVE_ANALYSIS: Callable = filter_inactive_analysis_cases
     HAS_SEQUENCE: Callable = filter_cases_has_sequence

tests/store/filters/test_status_cases_filters.py

@@ -12,7 +12,6 @@
     filter_cases_by_entry_id,
     filter_cases_by_name,
     filter_cases_by_priority,
-    filter_cases_by_ticket_id,
     filter_cases_by_workflow_search,
     filter_cases_for_analysis,
     filter_cases_has_sequence,
@@ -704,36 +703,6 @@ def test_filter_running_cases_only_running_cases(store_with_multiple_cases_and_s
     assert active_cases.count() == cases_query.count()
 
 
-def test_filter_cases_by_ticket_no_matching_ticket(
-    store_with_multiple_cases_and_samples: Store, non_existent_id: str
-):
-    """Test that no cases are returned when filtering by a non-existent ticket."""
-    # GIVEN a store containing cases with no matching ticket id
-    cases_query: Query = store_with_multiple_cases_and_samples._get_query(table=Case)
-
-    # WHEN filtering cases by a non-existent ticket
-    filtered_cases: Query = filter_cases_by_ticket_id(cases=cases_query, ticket_id=non_existent_id)
-
-    # THEN the query should return no cases
-    assert filtered_cases.count() == 0
-
-
-def test_filter_cases_by_ticket_matching_ticket(
-    store_with_multiple_cases_and_samples: Store, ticket_id: str
-):
-    """Test that cases are returned when filtering by an existing ticket id."""
-    # GIVEN a store containing cases with a matching ticket id
-    cases_query: Query = store_with_multiple_cases_and_samples._get_query(table=Case)
-
-    # WHEN filtering cases by an existing ticket id
-    filtered_cases: Query = filter_cases_by_ticket_id(cases=cases_query, ticket_id=ticket_id)
-
-    # THEN the query should return cases with the matching ticket
-    assert filtered_cases.count() > 0
-    for case in filtered_cases:
-        assert ticket_id in case.tickets
-
-
 def test_filter_cases_by_customer_entry_ids(store_with_multiple_cases_and_samples: Store):
     """Test that cases are returned when filtering by customer entry ids."""
     # GIVEN a store containing cases with customer ids

tests/store/filters/test_status_order_filters.py

@@ -0,0 +1,51 @@
+from sqlalchemy.orm import Query
+
+from cg.constants import Workflow
+from cg.store.filters.status_order_filters import filter_orders_by_ticket_id
+from cg.store.models import Order
+from cg.store.store import Store
+
+
+def test_filter_orders_by_ticket_no_matching_ticket(base_store: Store, non_existent_id: str):
+    """Test that no cases are returned when filtering by a non-existent ticket."""
+    # GIVEN a store containing orders with no matching ticket id
+    order = Order(
+        id=1,
+        customer_id=1,
+        ticket_id=1,
+        workflow=Workflow.MIP_DNA,
+    )
+    base_store.session.add(order)
+    base_store.session.commit()
+    order_query: Query = base_store._get_query(table=Order)
+
+    # WHEN filtering orders by a non-existent ticket
+    filtered_orders: Query = filter_orders_by_ticket_id(orders=order_query, ticket_id=2)
+
+    # THEN the query should return no order
+    assert filtered_orders.count() == 0
+
+
+def test_filter_orders_by_ticket_id_matching_ticket(base_store: Store, ticket_id: str):
+    """Test that the order is returned when filtering by an existing ticket id."""
+
+    order = Order(
+        id=1,
+        customer_id=1,
+        ticket_id=int(ticket_id),
+        workflow=Workflow.MIP_DNA,
+    )
+    base_store.session.add(order)
+    base_store.session.commit()
+
+    # GIVEN a store containing an order with a matching ticket id
+    order_query: Query = base_store._get_query(table=Order)
+
+    # WHEN filtering orders by an existing ticket id
+    filtered_orders: Query = filter_orders_by_ticket_id(
+        orders=order_query, ticket_id=int(ticket_id)
+    )
+
+    # THEN the query should return cases with the matching ticket
+    assert filtered_orders.count() == 1
+    assert filtered_orders.first().ticket_id == int(ticket_id)

tests/store_helpers.py

@@ -552,6 +552,11 @@ def ensure_case_from_dict(
     ):
         """Load a case with samples and link relations from a dictionary."""
         customer_obj = StoreHelpers.ensure_customer(store)
+        order = store.get_order_by_ticket_id(ticket_id=int(case_info["tickets"])) or Order(
+            ticket_id=int(case_info["tickets"]),
+            customer_id=customer_obj.id,
+            workflow=case_info.get("data_analysis", Workflow.MIP_DNA),
+        )
         case = Case(
             name=case_info["name"],
             panels=case_info["panels"],
@@ -563,7 +568,7 @@ def ensure_case_from_dict(
             action=case_info.get("action"),
             tickets=case_info["tickets"],
         )
-
+        case.orders.append(order)
         case = StoreHelpers.add_case(store, case_obj=case, customer_id=customer_obj.internal_id)
 
         app_tag = app_tag or "WGSPCFC030"