diff --git a/statsig/statsig.py b/statsig/statsig.py index 68c4586..6011211 100644 --- a/statsig/statsig.py +++ b/statsig/statsig.py @@ -1,4 +1,4 @@ -from typing import Optional +from typing import Dict, List, Optional from . import FeatureGate from .client_initialize_formatter import ClientInitializeResponse @@ -139,6 +139,16 @@ def get_experiment_with_exposure_logging_disabled(user: StatsigUser, experiment: return __instance.get_experiment(user, experiment, log_exposure=False) +def get_experiment_groups(experiment: str) -> List[Dict]: + """ + Returns the group name and return value for each group in the given experiment. + + :param experiment: The name of the experiment + :return: A list of dicts, each containing 'group_name' and 'return_value' for a group + """ + return __instance.get_experiment_groups(experiment) + + def manually_log_experiment_exposure(user: StatsigUser, experiment: str): """ Logs an exposure event for the experiment diff --git a/statsig/statsig_server.py b/statsig/statsig_server.py index 1ab9318..92ebe78 100644 --- a/statsig/statsig_server.py +++ b/statsig/statsig_server.py @@ -2,7 +2,7 @@ import threading import time from concurrent.futures import ThreadPoolExecutor -from typing import Optional, Union +from typing import Dict, List, Optional, Union from . import globals from .config_evaluation import _ConfigEvaluation @@ -348,6 +348,39 @@ def task(): {"configName": experiment_name}, ) + def get_experiment_groups(self, experiment_name: str) -> List[Dict]: + """ + Returns the group name and return value for each group in the given experiment. + + :param experiment_name: The name of the experiment + :return: A list of dicts, each containing 'group_name' and 'return_value' for a group + """ + def task(): + if not self._initialized: + raise StatsigRuntimeError( + "Must call initialize before checking gates/configs/experiments or logging events" + ) + + spec = self._spec_store.get_config(experiment_name) + if spec is None or spec.get("entity") != "experiment": + return [] + + if spec.get("isActive", False) is not True: + return [] + + return [ + { + "group_name": rule.get("groupName"), + "return_value": rule.get("returnValue", {}), + } + for rule in spec.get("rules", []) + if rule.get("isExperimentGroup", False) is True + ] + + return self._errorBoundary.capture( + "get_experiment_groups", task, lambda: [], {"configName": experiment_name} + ) + def manually_log_experiment_exposure(self, user: StatsigUser, experiment_name: str): user = self.__normalize_user(user) result = self._evaluator.get_config(user, experiment_name) diff --git a/tests/test_get_experiment_groups.py b/tests/test_get_experiment_groups.py new file mode 100644 index 0000000..27fb324 --- /dev/null +++ b/tests/test_get_experiment_groups.py @@ -0,0 +1,173 @@ +import json +import os +import unittest +from unittest.mock import patch + +from network_stub import NetworkStub +from statsig import statsig, StatsigOptions, StatsigServer + +with open(os.path.join(os.path.abspath(os.path.dirname(__file__)), '../testdata/download_config_specs.json')) as r: + CONFIG_SPECS_RESPONSE = r.read() + +_api_override = "http://get-experiment-groups-test" +_network_stub = NetworkStub(_api_override) + +_options = StatsigOptions(api=_api_override, disable_diagnostics=True) + + +def _make_specs_with_active_experiment(): + """ + Return a copy of the base config specs with sample_experiment patched to + reflect what production specs look like: + - isActive: True + - experiment group rules have isExperimentGroup: True + - the sizing/allocation rule (empty id) does NOT have isExperimentGroup set + """ + specs = json.loads(CONFIG_SPECS_RESPONSE) + for config in specs["dynamic_configs"]: + if config["name"] == "sample_experiment": + config["isActive"] = True + for rule in config["rules"]: + # Real group rules have a non-empty id and isExperimentGroup: True + if rule.get("id"): + rule["isExperimentGroup"] = True + break + return specs + + +def _setup_network_stub(specs=None): + _network_stub.reset() + _network_stub.stub_request_with_value( + "download_config_specs/.*", 200, + specs if specs is not None else json.loads(CONFIG_SPECS_RESPONSE) + ) + _network_stub.stub_request_with_value("log_event", 202, {}) + + +@patch('requests.Session.request', side_effect=_network_stub.mock) +class TestGetExperimentGroupsOnServer(unittest.TestCase): + """Tests for StatsigServer.get_experiment_groups""" + + @patch('requests.Session.request', side_effect=_network_stub.mock) + def setUp(self, mock_request): + _setup_network_stub(_make_specs_with_active_experiment()) + self._server = StatsigServer() + self._server.initialize("secret-key", _options) + + def tearDown(self): + self._server.shutdown() + + def test_returns_groups_for_known_experiment(self, mock_request): + groups = self._server.get_experiment_groups("sample_experiment") + + self.assertIsInstance(groups, list) + self.assertGreater(len(groups), 0) + + def test_group_shape_has_group_name_and_return_value(self, mock_request): + groups = self._server.get_experiment_groups("sample_experiment") + + for group in groups: + self.assertIn("group_name", group) + self.assertIn("return_value", group) + + def test_group_names_match_spec(self, mock_request): + groups = self._server.get_experiment_groups("sample_experiment") + + group_names = [g["group_name"] for g in groups] + # Exactly the two variant groups — sizing rule must not appear + self.assertEqual(sorted(group_names), ["Control", "Test"]) + + def test_return_values_match_spec(self, mock_request): + groups = self._server.get_experiment_groups("sample_experiment") + + groups_by_name = {g["group_name"]: g["return_value"] for g in groups} + + self.assertEqual( + groups_by_name["Control"], + {"experiment_param": "control", "layer_param": True, "second_layer_param": False}, + ) + self.assertEqual( + groups_by_name["Test"], + {"experiment_param": "test", "layer_param": True, "second_layer_param": True}, + ) + + def test_returns_empty_list_for_unknown_experiment(self, mock_request): + groups = self._server.get_experiment_groups("nonexistent_experiment") + + self.assertEqual(groups, []) + + def test_returns_empty_list_for_dynamic_config(self, mock_request): + # dynamic configs are not experiments; should return [] + groups = self._server.get_experiment_groups("test_config") + + self.assertEqual(groups, []) + + def test_returns_empty_list_for_empty_name(self, mock_request): + groups = self._server.get_experiment_groups("") + + self.assertEqual(groups, []) + + def test_filters_out_sizing_rule_without_is_experiment_group(self, mock_request): + # The experimentSize/allocation rule has no isExperimentGroup field (absent = False). + # It must not appear in the returned groups. + groups = self._server.get_experiment_groups("sample_experiment") + group_names = [g["group_name"] for g in groups] + + self.assertNotIn("experimentSize", group_names) + + def test_returns_empty_list_for_inactive_experiment(self, mock_request): + # An experiment with isActive=False (or absent) should return [] + specs = _make_specs_with_active_experiment() + for config in specs["dynamic_configs"]: + if config["name"] == "sample_experiment": + config["isActive"] = False + break + + _network_stub.reset() + _network_stub.stub_request_with_value("download_config_specs/.*", 200, specs) + _network_stub.stub_request_with_value("log_event", 202, {}) + + server = StatsigServer() + server.initialize("secret-key", _options) + + groups = server.get_experiment_groups("sample_experiment") + self.assertEqual(groups, []) + + server.shutdown() + + +@patch('requests.Session.request', side_effect=_network_stub.mock) +class TestGetExperimentGroupsModuleLevel(unittest.TestCase): + """Tests for the module-level statsig.get_experiment_groups""" + + @classmethod + @patch('requests.Session.request', side_effect=_network_stub.mock) + def setUpClass(cls, mock_request): + _setup_network_stub(_make_specs_with_active_experiment()) + statsig.initialize("secret-key", _options) + + @classmethod + def tearDownClass(cls): + statsig.shutdown() + + def test_module_level_returns_groups(self, mock_request): + groups = statsig.get_experiment_groups("sample_experiment") + + self.assertIsInstance(groups, list) + self.assertGreater(len(groups), 0) + + def test_module_level_group_shape(self, mock_request): + groups = statsig.get_experiment_groups("sample_experiment") + + for group in groups: + self.assertIn("group_name", group) + self.assertIn("return_value", group) + + def test_module_level_returns_empty_for_unknown(self, mock_request): + groups = statsig.get_experiment_groups("nonexistent_experiment") + + self.assertEqual(groups, []) + + +if __name__ == '__main__': + unittest.main()