Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# Ensure variant-choosing is deterministic.
self.assertEqual(variant, three_variants_more._choose_variant(bucket))
# If previously we had a variant, we should still have the same one
# now.
if previous_variant:
self.assertEqual(variant, previous_variant)
for experiment in (control_only, three_variants, three_variants_more):
for variant, percentage in experiment.variants.items():
count = counters[experiment.name][variant]
scaled_percentage = float(count) / (experiment.num_buckets / 100)
self.assertEqual(scaled_percentage, percentage)
# Test boundary conditions around the maximum percentage allowed for
# variants.
fifty_fifty = parse_experiment(
{
"id": 1,
"name": "fifty_fifty",
"owner": "test",
"type": "r2",
"version": "1",
"start_ts": time.time() - THIRTY_DAYS,
"stop_ts": time.time() + THIRTY_DAYS,
"experiment": {"variants": {"control_1": 50, "control_2": 50}},
}
)
almost_fifty_fifty = parse_experiment(
{
"id": 1,
"name": "almost_fifty_fifty",
"owner": "test",
def test_is_newer_than(self):
cfg = {
"id": 1,
"name": "test_feature",
"type": "feature_flag",
"version": "1",
"start_ts": time.time() - THIRTY_DAYS,
"stop_ts": time.time() + THIRTY_DAYS,
"experiment": {
"targeting": {"logged_in": [True]},
"newer_than": int(time.time()) - THIRTY_DAYS,
"variants": {"active": 100},
},
}
feature_flag = parse_experiment(cfg)
self.assertEqual(
feature_flag.variant(
user_id=self.user_id, logged_in=self.user_logged_in, user_created=int(time.time())
),
"active",
)
self.assertNotEqual(
feature_flag.variant(user_id=self.user_id, logged_in=self.user_logged_in), "active"
)
def test_user_in(self):
cfg = {
"id": 1,
"name": "test_feature",
"type": "feature_flag",
"version": "1",
"start_ts": time.time() - THIRTY_DAYS,
"stop_ts": time.time() + THIRTY_DAYS,
"experiment": {
"overrides": {
"user_name": {"Gary": "active", "dave": "active", "ALL_UPPERCASE": "active"}
},
"variants": {"active": 0},
},
}
feature_flag = parse_experiment(cfg)
self.assertEqual(
feature_flag.variant(
user_id=self.user_id, logged_in=self.user_logged_in, user_name="Gary"
),
"active",
)
self.assertEqual(
feature_flag.variant(
user_id=self.user_id, logged_in=self.user_logged_in, user_name=self.user_name
),
"active",
)
all_uppercase_id = "t2_f00d"
all_uppercase_name = "ALL_UPPERCASE"
self.assertEqual(
feature_flag.variant(
def test_calculate_bucket_value(self):
cfg = {
"id": 1,
"name": "test",
"owner": "test",
"type": "r2",
"version": "1",
"start_ts": time.time() - THIRTY_DAYS,
"stop_ts": time.time() + THIRTY_DAYS,
"experiment": {"variants": {"control_1": 10, "control_2": 10}},
}
experiment = parse_experiment(cfg)
experiment.num_buckets = 1000
self.assertEqual(experiment._calculate_bucket("t2_1"), int(236))
cfg = {
"id": 1,
"name": "test",
"owner": "test",
"type": "r2",
"version": "1",
"start_ts": time.time() - THIRTY_DAYS,
"stop_ts": time.time() + THIRTY_DAYS,
"experiment": {"seed": "test-seed", "variants": {"control_1": 10, "control_2": 10}},
}
seeded_experiment = parse_experiment(cfg)
self.assertNotEqual(seeded_experiment.seed, experiment.seed)
self.assertIsNot(seeded_experiment.seed, None)
seeded_experiment.num_buckets = 1000
def test_disable_returns_forced_variant(self):
cfg = {
"id": 1,
"name": "test",
"owner": "test",
"type": "r2",
"version": "1",
"start_ts": time.time() - THIRTY_DAYS,
"stop_ts": time.time() + THIRTY_DAYS,
"enabled": False,
"experiment": {"id": 1, "name": "test", "variants": {"control_1": 10, "control_2": 10}},
}
experiment = parse_experiment(cfg)
self.assertTrue(isinstance(experiment, ForcedVariantExperiment))
def test_gold_disabled(self):
cfg = {
"id": 1,
"name": "test_feature",
"type": "feature_flag",
"version": "1",
"start_ts": time.time() - THIRTY_DAYS,
"stop_ts": time.time() + THIRTY_DAYS,
"experiment": {
"overrides": {"user_roles": {"gold": "active"}},
"variants": {"active": 0},
},
}
feature_flag = parse_experiment(cfg)
self.assertNotEqual(
feature_flag.variant(user_id=self.user_id, logged_in=self.user_logged_in), "active"
)
self.assertNotEqual(
feature_flag.variant(
user_id=self.user_id, logged_in=self.user_logged_in, user_roles=[]
),
"active",
)
self.assertNotEqual(
feature_flag.variant(
user_id=self.user_id, logged_in=self.user_logged_in, user_roles=["admin"]
),
"active",
)
def test_start_ts_and_stop_ts_ignore_expires(self):
expires = (datetime.now() - timedelta(days=30)).strftime(ISO_DATE_FMT)
cfg = {
"id": 1,
"name": "test",
"owner": "test",
"type": "r2",
"version": "1",
"start_ts": time.time() - THIRTY_DAYS,
"stop_ts": time.time() + THIRTY_DAYS,
"expires": expires,
"enabled": True,
"experiment": {"id": 1, "name": "test", "variants": {"control_1": 10, "control_2": 10}},
}
experiment = parse_experiment(cfg)
self.assertFalse(isinstance(experiment, ForcedVariantExperiment))
def test_choose_variant(self):
control_only = parse_experiment(
{
"id": 1,
"name": "control_only",
"owner": "test",
"type": "r2",
"version": "1",
"start_ts": time.time() - THIRTY_DAYS,
"stop_ts": time.time() + THIRTY_DAYS,
"experiment": {"variants": {"control_1": 10, "control_2": 10}},
}
)
three_variants = parse_experiment(
{
"id": 1,
"name": "three_variants",
"owner": "test",
"type": "r2",
"version": "1",
"start_ts": time.time() - THIRTY_DAYS,
"stop_ts": time.time() + THIRTY_DAYS,
"experiment": {
"variants": {"remove_vote_counters": 5, "control_1": 10, "control_2": 5}
},
}
)
three_variants_more = parse_experiment(
{
"id": 1,
def create_simple_experiment():
cfg = get_simple_config()
experiment = parse_experiment(cfg)
return experiment
def _get_experiment(self, name: str) -> Optional[Experiment]:
if name not in self._experiment_cache:
experiment_config = self._get_config(name)
if not experiment_config:
experiment = None
else:
try:
experiment = parse_experiment(experiment_config)
except Exception as err:
logger.error("Invalid configuration for experiment %s: %s", name, err)
return None
self._experiment_cache[name] = experiment
return self._experiment_cache[name]