tests: adopted to exclude criterion combination

Author: glichtner

tests/_fixtures/omop_fixture.py

@@ -177,15 +177,15 @@ def celida_recommendation(
             recommendation_id=recommendation_id,
             pi_pair_url="https://example.com",
             pi_pair_name="my_pair",
-            pi_pair_hash=hash("my_pair"),
+            pi_pair_hash=str(hash("my_pair")),
         )
         db_session.add(pi_pair)
         db_session.commit()
 
         criterion = Criterion(
             criterion_id=criterion_id,
             criterion_description="my_description",
-            criterion_hash=hash("my_criterion"),
+            criterion_hash=str(hash("my_criterion")),
         )
         db_session.add(criterion)
         db_session.commit()

tests/execution_engine/converter/action/test_drug_administration.py

@@ -64,8 +64,8 @@ def test_multiple_doses(self):
             ],
         )
 
-        comb = action.to_expression()
-        criteria = list(comb)
+        expr = action.to_expression()
+        criteria = list(expr.args)
 
         assert len(criteria) == 3
         assert all(isinstance(c, DrugExposure) for c in criteria)

tests/execution_engine/omop/cohort/test_cohort_recommendation.py

@@ -1,22 +1,20 @@
 import pytest
 
-from execution_engine.omop.cohort import PopulationInterventionPair
+from execution_engine.omop.cohort import PopulationInterventionPairExpr
 from execution_engine.omop.cohort.recommendation import Recommendation
 from execution_engine.omop.concepts import Concept
 from execution_engine.omop.criterion.visit_occurrence import ActivePatients
+from execution_engine.util import logic
 from tests.mocks.criterion import MockCriterion
 
 
 class TestRecommendation:
 
     def test_serialization(self):
         # Register the mock criterion class
-        from execution_engine.omop.criterion import factory
-
-        factory.register_criterion_class("MockCriterion", MockCriterion)
 
         original = Recommendation(
-            pi_pairs=[],
+            expr=logic.BooleanFunction(),
             base_criterion=MockCriterion("c"),
             name="foo",
             title="bar",
@@ -47,7 +45,7 @@ def test_serialization_with_active_patients(self, concept):
         # specifically because there used to be a problem with the
         # serialization of that combination.
         original = Recommendation(
-            pi_pairs=[],
+            expr=logic.BooleanFunction(),
             base_criterion=ActivePatients(),
             name="foo",
             title="bar",
@@ -60,11 +58,12 @@ def test_serialization_with_active_patients(self, concept):
         deserialized = Recommendation.from_json(json)
         assert original == deserialized
 
-    def test_database_serialization(self, concept):
+    def test_database_serialization(
+        self,
+        concept,
+        db_session,  # use_db_session fixture for a proper database clean up after this test
+    ):
         from execution_engine.execution_engine import ExecutionEngine
-        from execution_engine.omop.criterion.factory import register_criterion_class
-
-        register_criterion_class("MockCriterion", MockCriterion)
 
         e = ExecutionEngine(verbose=False)
 
@@ -74,18 +73,18 @@ def test_database_serialization(self, concept):
         population_criterion = MockCriterion("p")
         intervention_criterion = MockCriterion("i)")
 
-        pi_pair = PopulationInterventionPair(
+        pi_pair = PopulationInterventionPairExpr(
+            population_expr=population_criterion,
+            intervention_expr=intervention_criterion,
             name="foo",
             url="foo",
             base_criterion=base_criterion,
-            population=population_criterion,
-            intervention=intervention_criterion,
         )
 
         pi_pairs = [pi_pair]
 
         recommendation = Recommendation(
-            pi_pairs=pi_pairs,
+            expr=pi_pair,
             base_criterion=ActivePatients(),
             name="foo",
             title="bar",
@@ -105,33 +104,33 @@ def test_database_serialization(self, concept):
             with pytest.raises(ValueError, match=r"Database ID has not been set yet!"):
                 assert pi_pair.id is None
 
-            for criterion in pi_pair.flatten():
-                with pytest.raises(
-                    ValueError, match=r"Database ID has not been set yet!"
-                ):
-                    assert criterion.id is None
+        for criterion in recommendation.atoms():
+            with pytest.raises(ValueError, match=r"Database ID has not been set yet!"):
+                assert criterion.id is None
 
         e.register_recommendation(recommendation)
 
         assert recommendation.id is not None
         assert recommendation.base_criterion.id is not None
+
         for pi_pair in recommendation.population_intervention_pairs():
             assert pi_pair.id is not None
-            for criterion in pi_pair.flatten():
-                assert criterion.id is not None
+
+        for criterion in recommendation.atoms():
+            assert criterion.id is not None
 
         rec_loaded = e.load_recommendation_from_database(url)
 
         assert recommendation == rec_loaded
         assert rec_loaded.id == recommendation.id
-        assert len(rec_loaded._pi_pairs) == len(pi_pairs)
+        assert len(list(rec_loaded.population_intervention_pairs())) == 1
 
-        for pi_pair_loaded, pi_pair in zip(rec_loaded._pi_pairs, pi_pairs):
+        for pi_pair_loaded, pi_pair in zip(
+            list(rec_loaded.population_intervention_pairs()), pi_pairs
+        ):
             assert pi_pair_loaded.id == pi_pair.id
 
-            assert len(pi_pair_loaded.flatten()) == len(pi_pair.flatten())
-
-            for criterion_loaded, criterion in zip(
-                pi_pair_loaded.flatten(), pi_pair.flatten()
-            ):
-                assert criterion.id == criterion_loaded.id
+        for criterion_loaded, criterion in zip(
+            rec_loaded.atoms(), recommendation.atoms()
+        ):
+            assert criterion.id == criterion_loaded.id

tests/execution_engine/omop/cohort/test_population_intervention_pair.py

@@ -1,21 +1,22 @@
 from execution_engine.omop.cohort.population_intervention_pair import (
-    PopulationInterventionPair,
+    PopulationInterventionPairExpr,
 )
+from execution_engine.util import logic
 from tests.mocks.criterion import MockCriterion
 
 
 class TestPopulationInterventionPair:
 
     def test_serialization(self):
         # Register the mock criterion class
-        from execution_engine.omop.criterion import factory
-
-        factory.register_criterion_class("MockCriterion", MockCriterion)
-
-        original = PopulationInterventionPair(
-            name="foo", url="bar", base_criterion=MockCriterion("c")
+        original = PopulationInterventionPairExpr(
+            population_expr=logic.NonSimplifiableAnd(MockCriterion("population")),
+            intervention_expr=logic.NonSimplifiableAnd(MockCriterion("intervention")),
+            name="foo",
+            url="bar",
+            base_criterion=MockCriterion("base"),
         )
 
         json = original.json()
-        deserialized = PopulationInterventionPair.from_json(json)
+        deserialized = PopulationInterventionPairExpr.from_json(json)
         assert original == deserialized

tests/execution_engine/omop/criterion/combination/__init__.py

@@ -1,56 +0,0 @@
-from typing import Any, Dict, Self
-
-from sqlalchemy import Select, select
-
-from execution_engine.omop.criterion.abstract import (
-    Criterion,
-    column_interval_type,
-    observation_end_datetime,
-    observation_start_datetime,
-)
-from execution_engine.util.interval import IntervalType
-
-
-class NoopCriterion(Criterion):
-    """
-    Select patients who are post-surgical in the timeframe between the day of the surgery and 6 days after the surgery.
-    """
-
-    _static = True
-
-    def _create_query(self) -> Select:
-        """
-        Get the SQL Select query for data required by this criterion.
-        """
-        subquery = self.base_query().subquery()
-
-        query = select(
-            subquery.c.person_id,
-            column_interval_type(IntervalType.POSITIVE),
-            observation_start_datetime.label("interval_start"),
-            observation_end_datetime.label("interval_end"),
-        )
-
-        query = self._filter_base_persons(query, c_person_id=subquery.c.person_id)
-        query = self._filter_datetime(query)
-
-        return query
-
-    @classmethod
-    def from_dict(cls, data: Dict[str, Any]) -> Self:
-        """
-        Create an object from a dictionary.
-        """
-        return cls()
-
-    def description(self) -> str:
-        """
-        Get a description of the criterion.
-        """
-        return self.__class__.__name__
-
-    def dict(self) -> dict:
-        """
-        Get a dictionary representation of the object.
-        """
-        return {}