runtime rule NO MATCH ✅

Author: joshua-koehler

mixpanel/flags/local_feature_flags.py

@@ -312,11 +312,26 @@ def _get_assigned_rollout(
             rollout_hash = normalized_hash(str(context_value), salt)
 
             if (rollout_hash < rollout.rollout_percentage
-                and self._is_runtime_evaluation_satisfied(rollout, context)
+                and self._is_runtime_rules_engine_satisfied(rollout, context)
             ):
                 return rollout
 
         return None
+    
+    def _is_runtime_rules_engine_satisfied(self, rollout: Rollout, context: Dict[str, Any]) -> bool:
+        if not rollout.runtime_evaluation_rule:
+            return self._is_runtime_evaluation_satisfied(rollout, context)
+        if not (custom_properties := context.get("custom_properties")):
+            return False
+        if not isinstance(custom_properties, dict):
+            return False
+        import json_logic
+        try:
+            result = json_logic.jsonLogic(rollout.runtime_evaluation_rule, custom_properties)
+            return bool(result)
+        except Exception as e:
+            logger.exception("Error evaluating runtime evaluation rule", e)
+            return False
 
     def _is_runtime_evaluation_satisfied(
         self, rollout: Rollout, context: Dict[str, Any]

mixpanel/flags/test_local_feature_flags.py

@@ -189,7 +189,9 @@ async def test_get_variant_value_returns_variant_when_rollout_percentage_hundred
     # TODO problem test doesn't fail
     @respx.mock
     async def test_get_variant_value_respects_runtime_evaluation_rule_satisfied(self):
-        runtime_eval = {"oops": "sorry"}
+        runtime_eval = {
+            "==": [{"var": "plan"}, "premium"]
+        }
         flag = create_test_flag(runtime_evaluation_rule=runtime_eval)
         await self.setup_flags([flag])
         context = {
@@ -204,7 +206,9 @@ async def test_get_variant_value_respects_runtime_evaluation_rule_satisfied(self
 
     @respx.mock
     async def test_get_variant_value_respects_runtime_evaluation_rule_not_satisfied(self):
-        runtime_eval = {"oops": "sorry"}
+        runtime_eval = {
+            "==": [{"var": "plan"}, "basic"]
+        }
         flag = create_test_flag(runtime_evaluation_rule=runtime_eval)
         await self.setup_flags([flag])
         context = {

mixpanel/flags/types.py

@@ -31,7 +31,7 @@ class VariantOverride(BaseModel):
 class Rollout(BaseModel):
     rollout_percentage: float
     runtime_evaluation_definition: Optional[Dict[str, str]] = None
-    runtime_evaluation_rule: Optional[Dict[str, str]] = None
+    runtime_evaluation_rule: Optional[Dict[Any, Any]] = None
     variant_override: Optional[VariantOverride] = None
     variant_splits: Optional[Dict[str,float]] = None