test_cmab_ups_exclusion.py

# Copyright 2026, Optimizely
# 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.

"""
Tests to verify that CMAB experiments are excluded from User Profile Service.

CMAB experiments should not use UPS for sticky bucketing because:
- UPS maintains decisions across experiment lifetime without considering TTL
- UPS doesn't account for changing user attributes
- This contradicts CMAB's dynamic decision-making nature
"""

import unittest
from unittest import mock
from optimizely import decision_service, entities, user_profile
from optimizely.cmab.cmab_service import DefaultCmabService


class TestCmabUpsExclusion(unittest.TestCase):
    """Test suite to verify CMAB experiments are excluded from UPS."""

    def test_cmab_experiment_skips_ups_lookup(self):
        """
        Test that get_variation skips UPS lookup for CMAB experiments.

        This test verifies that the condition `not experiment.cmab` prevents
        calling get_stored_variation when processing CMAB experiments.
        """
        # Create mocks
        mock_logger = mock.MagicMock()
        mock_ups = mock.MagicMock(spec=user_profile.UserProfileService)
        mock_cmab_service = mock.MagicMock(spec=DefaultCmabService)
        mock_project_config = mock.MagicMock()
        mock_user_context = mock.MagicMock()
        mock_user_context.user_id = 'test_user'
        mock_user_context.get_user_attributes.return_value = {}

        # Create decision service
        ds = decision_service.DecisionService(
            mock_logger,
            mock_ups,
            mock_cmab_service
        )

        # Create a CMAB experiment
        variation = entities.Variation('1', 'var1')
        cmab_experiment = entities.Experiment(
            '123',
            'cmab_test',
            'Running',
            ['1'],
            [variation],
            {},
            [],
            '1',
            cmab={'attributeIds': []}
        )

        # Create user profile tracker with a stored variation
        user_profile_tracker = user_profile.UserProfileTracker(
            'test_user',
            mock_ups,
            mock_logger
        )
        user_profile_tracker.user_profile = user_profile.UserProfile(
            'test_user',
            {'123': {'variation_id': 'stored_variation'}}
        )

        # Mock get_stored_variation to verify it's not called for CMAB
        # Also mock audience helper to pass audience conditions
        with mock.patch('optimizely.helpers.audience.does_user_meet_audience_conditions',
                       return_value=(True, [])):
            with mock.patch.object(
                ds,
                'get_stored_variation',
                return_value=None
            ) as mock_get_stored:
                # Mock CMAB decision
                mock_cmab_service.get_decision.return_value = (
                    {'variation_id': '1', 'cmab_uuid': 'uuid-123'},
                    []
                )
                mock_project_config.get_variation_from_id.return_value = variation

                # Mock bucketer to say user is in traffic
                with mock.patch.object(ds.bucketer, 'bucket_to_entity_id') as m:
                    m.return_value = ('1', [])

                    # Get variation
                    result = ds.get_variation(
                        mock_project_config,
                        cmab_experiment,
                        mock_user_context,
                        user_profile_tracker
                    )

                    # Verify get_stored_variation was NOT called
                    # because experiment.cmab is True
                    mock_get_stored.assert_not_called()

    def test_cmab_experiment_skips_ups_save(self):
        """
        Test that get_variation skips UPS save for CMAB experiments.

        This test verifies that the condition `not experiment.cmab` prevents
        calling update_user_profile when processing CMAB experiments.
        """
        # Create mocks
        mock_logger = mock.MagicMock()
        mock_ups = mock.MagicMock(spec=user_profile.UserProfileService)
        mock_cmab_service = mock.MagicMock(spec=DefaultCmabService)
        mock_project_config = mock.MagicMock()
        mock_user_context = mock.MagicMock()
        mock_user_context.user_id = 'test_user'
        mock_user_context.get_user_attributes.return_value = {}

        # Create decision service
        ds = decision_service.DecisionService(
            mock_logger,
            mock_ups,
            mock_cmab_service
        )

        # Create a CMAB experiment with variation
        variation = entities.Variation('1', 'var1')
        cmab_experiment = entities.Experiment(
            '123',
            'cmab_test',
            'Running',
            ['1'],
            [variation],
            {},
            [],
            '1',
            cmab={'attributeIds': []}
        )

        # Mock get_variation_from_id to return the variation
        mock_project_config.get_variation_from_id.return_value = variation

        # Create user profile tracker
        user_profile_tracker = user_profile.UserProfileTracker(
            'test_user',
            mock_ups,
            mock_logger
        )

        # Mock CMAB decision
        mock_cmab_service.get_decision.return_value = (
            {'variation_id': '1', 'cmab_uuid': 'uuid-123'},
            []
        )

        # Mock audience helper to pass audience conditions
        with mock.patch('optimizely.helpers.audience.does_user_meet_audience_conditions',
                       return_value=(True, [])):
            # Mock bucketer to say user is in traffic
            with mock.patch.object(ds.bucketer, 'bucket_to_entity_id') as m:
                m.return_value = ('1', [])

                # Get variation
                result = ds.get_variation(
                    mock_project_config,
                    cmab_experiment,
                    mock_user_context,
                    user_profile_tracker
                )

                # Verify profile was NOT updated
                self.assertFalse(
                    user_profile_tracker.profile_updated,
                    "CMAB should not update user profile"
                )

    def test_non_cmab_experiment_uses_ups_lookup(self):
        """
        Test that regular experiments still use UPS lookup.
        """
        # Create mocks
        mock_logger = mock.MagicMock()
        mock_ups = mock.MagicMock(spec=user_profile.UserProfileService)
        mock_cmab_service = mock.MagicMock(spec=DefaultCmabService)
        mock_project_config = mock.MagicMock()
        mock_user_context = mock.MagicMock()
        mock_user_context.user_id = 'test_user'

        # Create decision service
        ds = decision_service.DecisionService(
            mock_logger,
            mock_ups,
            mock_cmab_service
        )

        # Create a regular (non-CMAB) experiment
        variation = entities.Variation('1', 'var1')
        regular_experiment = entities.Experiment(
            '123',
            'regular_test',
            'Running',
            ['1'],
            [variation],
            {},
            [],
            '1'
        )

        # Create user profile tracker with stored variation
        user_profile_tracker = user_profile.UserProfileTracker(
            'test_user',
            mock_ups,
            mock_logger
        )

        # Mock get_stored_variation to return the variation
        with mock.patch.object(
            ds,
            'get_stored_variation',
            return_value=variation
        ) as mock_get_stored:
            # Get variation
            result = ds.get_variation(
                mock_project_config,
                regular_experiment,
                mock_user_context,
                user_profile_tracker
            )

            # Verify get_stored_variation WAS called for regular experiments
            mock_get_stored.assert_called_once()

            # Verify returned variation is the stored one
            self.assertEqual(result['variation'], variation)

    def test_non_cmab_experiment_uses_ups_save(self):
        """
        Test that regular experiments still save to UPS.

        This verifies our changes don't break existing UPS behavior.
        """
        # Create mocks
        mock_logger = mock.MagicMock()
        mock_ups = mock.MagicMock(spec=user_profile.UserProfileService)
        mock_cmab_service = mock.MagicMock(spec=DefaultCmabService)
        mock_project_config = mock.MagicMock()
        mock_user_context = mock.MagicMock()
        mock_user_context.user_id = 'test_user'
        mock_user_context.get_user_attributes.return_value = {}

        # Create decision service
        ds = decision_service.DecisionService(
            mock_logger,
            mock_ups,
            mock_cmab_service
        )

        # Create a regular experiment with variation
        variation = entities.Variation('1', 'var1')
        regular_experiment = entities.Experiment(
            '123',
            'regular_test',
            'Running',
            ['1'],
            [variation],
            {},
            [],
            '1'
        )

        # Create user profile tracker
        user_profile_tracker = user_profile.UserProfileTracker(
            'test_user',
            mock_ups,
            mock_logger
        )

        # Mock audience helper to pass audience conditions
        with mock.patch('optimizely.helpers.audience.does_user_meet_audience_conditions',
                       return_value=(True, [])):
            # Mock bucketer to return variation
            with mock.patch.object(
                ds.bucketer,
                'bucket',
                return_value=(variation, [])
            ):
                # Get variation
                result = ds.get_variation(
                    mock_project_config,
                    regular_experiment,
                    mock_user_context,
                    user_profile_tracker
                )

                # Verify profile WAS updated for regular experiments
                self.assertTrue(
                    user_profile_tracker.profile_updated,
                    "Regular experiments should update user profile"
                )


if __name__ == '__main__':
    unittest.main()