feat: support for web testing presegmentation

Kaustubh22327 · rohitesh-wingify · commit 9fc53fd64195 · 2026-06-30T12:58:21.000+05:30
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [1.60.0] - 2026-06-29
+
+### Added
+
+- Added support for `campaignVariation` segmentation operator to evaluate web testing campaign assignments. Context can now pass `platformVariables.webTestingCampaigns` (as a JSON object or JSON string) to target users based on their web testing campaign and variation assignments.
+
 ## [1.55.0] - 2026-06-16
 
 ### Added
diff --git a/README.md b/README.md
@@ -141,9 +141,30 @@ context = {
     'user_agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36',
     'ip_address': '1.1.1.1',
     'session_id': 1697123456  # Optional: Custom session ID for web client integration
+    'platformVariables': {
+        # Reference example only:.
+        # In production, fetch campaign assignments using script run in frontend and pass that object to backend as webTestingCampaigns.
+        'webTestingCampaigns': '{"122":"1","130":"2"}'
+    },
+    'session_id': 1697123456  # Optional: Custom session ID for web client integration
 }
 ```
 
+## Web testing pre-segmentation
+
+Server-side flag decisions can align with **Web Testing** (browser) experiments. **Campaign assignments are typically read on the frontend** (for example, VWO cookies) and sent to your server; pass them in `platformVariables.webTestingCampaigns` as a map of **campaign ID -> variation ID** (strings), or as a **JSON string** of that object.
+
+Pre-segment rules in the VWO dashboard can use the **`campaignVariation`** operator:
+
+| Operand pattern | Meaning |
+| --------------- | ------- |
+| `C` | User is in campaign `C` (any variation). |
+| `!C` | User is **not** in campaign `C`. |
+| `C_V` | User is in campaign `C` with variation `V`. |
+| `C_!V` | User is in campaign `C` and assigned variation is **not** `V`. |
+
+Rollout rules evaluate pre-segments from the **first variation** only; AB/testing rules use **campaign-level** segments.
+
 ## Session Management
 
 The SDK provides automatic session management capabilities to enable seamless integration with VWO's web client testing campaigns. Session IDs are automatically generated and managed to connect server-side feature flag decisions with client-side user sessions.
diff --git a/setup.py b/setup.py
@@ -143,7 +143,7 @@ def run(self):
 
 setup(
     name=_brand["name"],
-    version="1.55.0",
+    version="1.60.0",
     description=_brand["description"],
     long_description=long_description,
     long_description_content_type="text/markdown",
diff --git a/tests/unit/packages/segmentation_evaluator/web_testing_segment_test.py b/tests/unit/packages/segmentation_evaluator/web_testing_segment_test.py
@@ -0,0 +1,216 @@
+# Copyright 2024-2026 Wingify Software Pvt. Ltd.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "..", "..")))
+
+from wingify.models.user.context_model import ContextModel
+from wingify.packages.segmentation_evaluator.evaluators.segment_evaluator import (
+    SegmentEvaluator,
+)
+from wingify.packages.segmentation_evaluator.core.segmentation_manager import (
+    SegmentationManager,
+)
+from wingify.packages.segmentation_evaluator.utils.web_testing_segment_util import (
+    evaluate_web_testing_campaign_variation,
+    normalize_web_testing_campaigns_map,
+    parse_web_testing_campaigns_from_context,
+)
+from wingify.services.settings_manager import SettingsManager
+
+
+def _context_with_platform_vars(extra: dict):
+    base = {"id": "user-1", **extra}
+    with patch.object(SettingsManager, "get_instance") as m:
+        m.return_value.get_account_id.return_value = "1"
+        return ContextModel(base)
+
+
+class TestWebTestingSegmentUtil(unittest.TestCase):
+    def test_evaluate_variation_patterns(self):
+        campaign_assignments_map = {"1": "1", "2": "2"}
+        self.assertTrue(evaluate_web_testing_campaign_variation("1_1", campaign_assignments_map)[0])
+        self.assertFalse(evaluate_web_testing_campaign_variation("1_1", campaign_assignments_map)[1])
+        self.assertFalse(evaluate_web_testing_campaign_variation("1_2", campaign_assignments_map)[0])
+        self.assertFalse(evaluate_web_testing_campaign_variation("99_1", campaign_assignments_map)[0])
+        self.assertTrue(evaluate_web_testing_campaign_variation("1_!2", campaign_assignments_map)[0])
+        self.assertFalse(evaluate_web_testing_campaign_variation("1_!1", campaign_assignments_map)[0])
+        self.assertFalse(evaluate_web_testing_campaign_variation("99_!1", campaign_assignments_map)[0])
+        self.assertTrue(evaluate_web_testing_campaign_variation("!99", campaign_assignments_map)[0])
+        self.assertFalse(evaluate_web_testing_campaign_variation("!1", campaign_assignments_map)[0])
+
+    def test_null_map_like_node(self):
+        self.assertTrue(evaluate_web_testing_campaign_variation("!1", None)[0])
+        self.assertFalse(evaluate_web_testing_campaign_variation("1_1", None)[0])
+
+    def test_invalid_encoding(self):
+        evaluation_result_pair = evaluate_web_testing_campaign_variation("bogus", {"1": "1"})
+        self.assertFalse(evaluation_result_pair[0])
+        self.assertTrue(evaluation_result_pair[1])
+
+    def test_multi_digit_ids(self):
+        campaign_assignments_map = {"122": "4"}
+        self.assertTrue(evaluate_web_testing_campaign_variation("122_4", campaign_assignments_map)[0])
+        self.assertTrue(evaluate_web_testing_campaign_variation("122_!1", campaign_assignments_map)[0])
+        self.assertFalse(evaluate_web_testing_campaign_variation("!122", campaign_assignments_map)[0])
+
+    def test_campaign_only_any_variation(self):
+        self.assertTrue(
+            evaluate_web_testing_campaign_variation("100", {"100": "1"})[0]
+        )
+        self.assertTrue(
+            evaluate_web_testing_campaign_variation("100", {"100": "9"})[0]
+        )
+        self.assertFalse(evaluate_web_testing_campaign_variation("100", {})[0])
+        self.assertFalse(
+            evaluate_web_testing_campaign_variation("100", {"99": "1"})[0]
+        )
+        self.assertFalse(
+            evaluate_web_testing_campaign_variation(
+                "100", {"1": "1", "2": "2"}
+            )[0]
+        )
+
+    def test_normalize_coerces_types(self):
+        self.assertEqual(
+            normalize_web_testing_campaigns_map({129: 1, "14": 2}),
+            {"129": "1", "14": "2"},
+        )
+
+
+class TestWebTestingSegmentEvaluator(unittest.TestCase):
+    def _evaluator(self, context: ContextModel):
+        ev = SegmentEvaluator()
+        ev.context = context
+        return ev
+
+    def test_or_json_string_web_testing_campaigns(self):
+        ctx = _context_with_platform_vars(
+            {
+                "platform_variables": {
+                    "web_testing_campaigns": '{"1":"1"}',
+                }
+            }
+        )
+        ev = self._evaluator(ctx)
+        self.assertTrue(ev.is_segmentation_valid({"or": [{"campaignVariation": "1_1"}]}, {}))
+
+    def test_or_object_web_testing_campaigns(self):
+        ctx = _context_with_platform_vars(
+            {"platform_variables": {"web_testing_campaigns": {"1": 1}}}
+        )
+        ev = self._evaluator(ctx)
+        self.assertTrue(ev.is_segmentation_valid({"or": [{"campaignVariation": "1_1"}]}, {}))
+
+    def test_not_in_campaign(self):
+        ctx = _context_with_platform_vars(
+            {"platform_variables": {"web_testing_campaigns": "{}"}}
+        )
+        ev = self._evaluator(ctx)
+        self.assertTrue(ev.is_segmentation_valid({"or": [{"campaignVariation": "!1"}]}, {}))
+
+    def test_nested_not_campaign_variation(self):
+        ctx = _context_with_platform_vars(
+            {"platform_variables": {"web_testing_campaigns": '{"1":"1"}'}}
+        )
+        ev = self._evaluator(ctx)
+        self.assertFalse(
+            ev.is_segmentation_valid({"not": {"campaignVariation": "1_1"}}, {})
+        )
+
+    def test_campaign_id_only(self):
+        ctx = _context_with_platform_vars(
+            {"platform_variables": {"web_testing_campaigns": '{"100":"2"}'}}
+        )
+        ev = self._evaluator(ctx)
+        self.assertTrue(ev.is_segmentation_valid({"or": [{"campaignVariation": "100"}]}, {}))
+
+    def test_operand_trimmed(self):
+        ctx = _context_with_platform_vars(
+            {"platform_variables": {"web_testing_campaigns": '{"1":"1"}'}}
+        )
+        ev = self._evaluator(ctx)
+        self.assertTrue(
+            ev.is_segmentation_valid({"or": [{"campaignVariation": "  1_1  "}]}, {})
+        )
+
+    def test_json_array_rejected(self):
+        ctx = _context_with_platform_vars(
+            {"platform_variables": {"web_testing_campaigns": "[]"}}
+        )
+        ev = self._evaluator(ctx)
+        self.assertFalse(
+            ev.is_segmentation_valid({"or": [{"campaignVariation": "1_1"}]}, {})
+        )
+
+    def test_numeric_operand_coerced_to_string(self):
+        ctx = _context_with_platform_vars(
+            {"platform_variables": {"web_testing_campaigns": '{"1":"1"}'}}
+        )
+        ev = self._evaluator(ctx)
+        self.assertTrue(ev.is_segmentation_valid({"or": [{"campaignVariation": 1}]}, {}))
+        ctx_float = _context_with_platform_vars(
+            {"platform_variables": {"web_testing_campaigns": '{"100":"2"}'}}
+        )
+        ev_float = self._evaluator(ctx_float)
+        self.assertTrue(
+            ev_float.is_segmentation_valid({"or": [{"campaignVariation": 100.0}]}, {})
+        )
+
+
+class TestWebTestingValidateSegmentationPrecheck(unittest.TestCase):
+    def test_precheck_false_when_campaign_variation_without_assignments_feed(self):
+        ctx = _context_with_platform_vars({})
+        segmentation_manager = SegmentationManager.get_instance()
+        segmentation_manager.attach_evaluator()
+        segmentation_manager.evaluator.context = ctx
+        self.assertFalse(
+            segmentation_manager.validate_segmentation(
+                {"not": {"campaignVariation": "1_1"}}, {}
+            )
+        )
+
+    def test_precheck_true_when_campaign_variation_with_empty_assignment_object_string(
+        self,
+    ):
+        ctx = _context_with_platform_vars(
+            {"platform_variables": {"web_testing_campaigns": "{}"}}
+        )
+        segmentation_manager = SegmentationManager.get_instance()
+        segmentation_manager.attach_evaluator()
+        segmentation_manager.evaluator.context = ctx
+        self.assertTrue(
+            segmentation_manager.validate_segmentation(
+                {"or": [{"campaignVariation": "!1"}]}, {}
+            )
+        )
+
+    def test_duplicate_keys_in_json_string_last_value_wins(self):
+        ctx = _context_with_platform_vars(
+            {
+                "platform_variables": {
+                    "web_testing_campaigns": '{"122":"1","122":"2"}',
+                }
+            }
+        )
+        campaigns_map_from_context = parse_web_testing_campaigns_from_context(ctx)
+        self.assertEqual(campaigns_map_from_context, {"122": "2"})
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/wingify/constants/Constants.py b/wingify/constants/Constants.py
@@ -17,7 +17,7 @@ class Constants:
     # TODO: read from setup.py
     sdk_meta = {
         "name": "vwo-fme-python-sdk",
-        "version": "1.55.0",
+        "version": "1.60.0",
     }
 
     SDK_VERSION = sdk_meta["version"]
diff --git a/wingify/models/user/context_model.py b/wingify/models/user/context_model.py
@@ -14,7 +14,8 @@
 
 
 from .context_vwo_model import ContextVWOModel
-from typing import Dict, List
+from typing import Any, Dict, List, Optional
+from ...utils.data_type_util import is_object
 from ...utils.uuid_util import get_uuid
 from ...services.settings_manager import SettingsManager
 from ...utils.function_util import get_current_unix_timestamp
@@ -33,6 +34,11 @@ def __init__(self, context: Dict):
         self._vwo = ContextVWOModel(vwo_context_data) if vwo_context_data else None
         self.post_segmentation_variables = context.get("post_segmentation_variables", [])
         self.bucketingSeed = context.get("bucketingSeed")
+        self.platform_variables: Optional[Dict[str, Any]] = (
+            dict(context.get("platform_variables"))
+            if is_object(context.get("platform_variables"))
+            else None
+        )
         self.session_id = context.get("session_id", None)
         if self.session_id is None:
             self.session_id = get_current_unix_timestamp()
@@ -96,4 +102,12 @@ def set_session_id(self, session_id: int) -> None:
         self.session_id = session_id
     
     def get_bucketing_seed(self) -> str:
-        return self.bucketingSeed
+        return self.bucketingSeed
+    
+    def get_platform_variables(self) -> Optional[Dict[str, Any]]:
+        return self.platform_variables
+
+    def set_platform_variables(self, platform_variables: Optional[Dict[str, Any]]) -> None:
+        self.platform_variables = (
+            dict(platform_variables) if is_object(platform_variables) else None
+        )
diff --git a/wingify/packages/segmentation_evaluator/core/segmentation_manager.py b/wingify/packages/segmentation_evaluator/core/segmentation_manager.py
@@ -24,6 +24,10 @@
 from ....utils.gateway_service_util import get_query_params, get_from_gateway_service
 from ....constants.Constants import Constants
 from ....enums.api_enum import ApiEnum
+from ..utils.web_testing_segment_util import (
+    dsl_contains_campaign_variation_node,
+    is_web_testing_campaigns_absent_or_null,
+)
 
 
 class SegmentationManager:
@@ -115,4 +119,11 @@ def validate_segmentation(self, dsl, properties):
         :param properties: The properties to validate against.
         :return: True if segmentation is valid, otherwise False.
         """
+        # if the dsl contains a campaign variation node, then we need to check if the campaign variation is valid
+        if dsl_contains_campaign_variation_node(dsl):
+            evaluation_context = getattr(self.evaluator, "context", None)
+            if evaluation_context is None or is_web_testing_campaigns_absent_or_null(
+                evaluation_context
+            ):
+                return False
         return self.evaluator.is_segmentation_valid(dsl, properties)
diff --git a/wingify/packages/segmentation_evaluator/enums/segment_operator_value_enum.py b/wingify/packages/segmentation_evaluator/enums/segment_operator_value_enum.py
@@ -34,3 +34,4 @@ class SegmentOperatorValueEnum(Enum):
     IP = "ip_address"
     BROWSER_VERSION = "browser_version"
     OS_VERSION = "os_version"
+    WEB_CAMPAIGN_VARIATION = "campaignVariation"
diff --git a/wingify/packages/segmentation_evaluator/evaluators/segment_evaluator.py b/wingify/packages/segmentation_evaluator/evaluators/segment_evaluator.py
@@ -68,6 +68,10 @@ def is_segmentation_valid(self, dsl, properties):
             return SegmentOperandEvaluator().evaluate_string_operand_dsl(
                 sub_dsl, self.context, SegmentOperatorValueEnum.OS_VERSION.value
             )
+        elif operator == SegmentOperatorValueEnum.WEB_CAMPAIGN_VARIATION.value:
+            return SegmentOperandEvaluator().evaluate_campaign_variation_dsl(
+                sub_dsl, self.context
+            )
         else:
             return False
 
diff --git a/wingify/packages/segmentation_evaluator/evaluators/segment_operand_evaluator.py b/wingify/packages/segmentation_evaluator/evaluators/segment_operand_evaluator.py
@@ -16,6 +16,11 @@
 import re
 from typing import Dict, Any
 from ..utils.segment_util import get_key_value, match_with_regex
+from ..utils.web_testing_segment_util import (
+    coerce_campaign_variation_operand_to_str,
+    parse_web_testing_campaigns_from_context,
+    evaluate_web_testing_campaign_variation,
+)
 from ..enums.segment_operand_regex_enum import SegmentOperandRegexEnum
 from ..enums.segment_operand_value_enum import SegmentOperandValueEnum
 from ..enums.segment_operator_value_enum import SegmentOperatorValueEnum
@@ -84,6 +89,25 @@ def evaluate_custom_variable_dsl(
                 operand_type, processed_values["operand_value"], tag_value
             )
 
+    def evaluate_campaign_variation_dsl(
+        self, dsl_operand_value: Any, context: ContextModel
+    ) -> bool:
+        """
+        Evaluates a campaignVariation DSL operand against the user's web testing campaign assignments.
+
+        :param dsl_operand_value: The raw operand value from the DSL (e.g. "123_1", "!123", 100).
+        :param context: The context object containing platform_variables.web_testing_campaigns.
+        :return: True if the segment condition is satisfied, otherwise False.
+        """
+        operand_str = coerce_campaign_variation_operand_to_str(dsl_operand_value)
+        if operand_str is None:
+            return False
+
+        operand_str = operand_str.strip()
+        campaigns_map = parse_web_testing_campaigns_from_context(context)
+        result, _invalid = evaluate_web_testing_campaign_variation(operand_str, campaigns_map)
+        return result
+
     def evaluate_user_dsl(self, dsl_operand_value, properties):
         """
         Evaluates a user DSL expression to check if a user ID is in a specified List.
diff --git a/wingify/packages/segmentation_evaluator/utils/web_testing_segment_util.py b/wingify/packages/segmentation_evaluator/utils/web_testing_segment_util.py
diff --git a/wingify/resources/error-messages.json b/wingify/resources/error-messages.json

Original file line number	Diff line number	Diff line change
`@@ -17,7 +17,7 @@ class Constants:`
`17`	`17`	`# TODO: read from setup.py`
`18`	`18`	`sdk_meta = {`
`19`	`19`	`"name": "vwo-fme-python-sdk",`
`20`		`- "version": "1.55.0",`
	`20`	`+ "version": "1.60.0",`
`21`	`21`	`}`
`22`	`22`
`23`	`23`	`SDK_VERSION = sdk_meta["version"]`