PROD.2026/Lotty
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(experiments): added experiments business logic

Browse Source
This commit is contained in:
ITQ
2026-02-23 10:54:04 +03:00
parent 68c6ca5267
commit d87671e49a
12 changed files with 2136 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/backend/apps/experiments/__init__.py
View File
src/backend/apps/experiments/apps.py
+5
View File
@@ -0,0 +1,5 @@
from django.apps import AppConfig
class ExperimentsConfig(AppConfig):
name = "apps.experiments"
src/backend/apps/experiments/migrations/0001_initial.py
+95
View File
@@ -0,0 +1,95 @@
# Generated by Django 5.2.11 on 2026-02-14 09:55
import django.core.validators
import uuid
from decimal import Decimal
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
]
operations = [
migrations.CreateModel(
name='Approval',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('comment', models.TextField(blank=True, verbose_name='comment')),
('created_at', models.DateTimeField(auto_now_add=True, verbose_name='created at')),
],
options={
'verbose_name': 'approval',
'verbose_name_plural': 'approvals',
'ordering': ['-created_at'],
},
),
migrations.CreateModel(
name='Experiment',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('name', models.CharField(max_length=200, verbose_name='name')),
('description', models.TextField(blank=True, verbose_name='description')),
('hypothesis', models.TextField(blank=True, verbose_name='hypothesis')),
('status', models.CharField(choices=[('draft', 'Draft'), ('in_review', 'In review'), ('approved', 'Approved'), ('running', 'Running'), ('paused', 'Paused'), ('completed', 'Completed'), ('archived', 'Archived'), ('rejected', 'Rejected')], db_index=True, default='draft', max_length=20, verbose_name='status')),
('version', models.PositiveIntegerField(default=1, verbose_name='version')),
('traffic_allocation', models.DecimalField(decimal_places=2, default=Decimal('100.00'), help_text='Percentage of eligible users included in experiment', max_digits=5, validators=[django.core.validators.MinValueValidator(Decimal('0.01')), django.core.validators.MaxValueValidator(Decimal('100.00'))], verbose_name='traffic allocation %')),
('targeting_rules', models.TextField(blank=True, help_text='DSL expression for targeting', verbose_name='targeting rules')),
('created_at', models.DateTimeField(auto_now_add=True, verbose_name='created at')),
('updated_at', models.DateTimeField(auto_now=True, verbose_name='updated at')),
],
options={
'verbose_name': 'experiment',
'verbose_name_plural': 'experiments',
'ordering': ['-created_at'],
},
),
migrations.CreateModel(
name='ExperimentLog',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('log_type', models.CharField(choices=[('status_change', 'Status change'), ('approval', 'Approval'), ('rejection', 'Rejection'), ('review_requested', 'Review requested'), ('guardrail_triggered', 'Guardrail triggered'), ('completed', 'Completed'), ('comment', 'Comment')], max_length=30, verbose_name='log type')),
('comment', models.TextField(blank=True, verbose_name='comment')),
('metadata', models.JSONField(blank=True, default=dict, verbose_name='metadata')),
('created_at', models.DateTimeField(auto_now_add=True, verbose_name='created at')),
],
options={
'verbose_name': 'experiment log',
'verbose_name_plural': 'experiment logs',
'ordering': ['-created_at'],
},
),
migrations.CreateModel(
name='ExperimentOutcome',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('outcome', models.CharField(choices=[('rollout', 'Rollout winner'), ('rollback', 'Rollback'), ('no_effect', 'No effect')], max_length=20, verbose_name='outcome')),
('rationale', models.TextField(verbose_name='rationale')),
('created_at', models.DateTimeField(auto_now_add=True, verbose_name='created at')),
],
options={
'verbose_name': 'experiment outcome',
'verbose_name_plural': 'experiment outcomes',
},
),
migrations.CreateModel(
name='Variant',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('name', models.CharField(max_length=100, verbose_name='name')),
('value', models.CharField(help_text='Value returned when this variant is selected', max_length=500, verbose_name='value')),
('weight', models.DecimalField(decimal_places=2, max_digits=5, validators=[django.core.validators.MinValueValidator(Decimal('0.00')), django.core.validators.MaxValueValidator(Decimal('100.00'))], verbose_name='weight %')),
('is_control', models.BooleanField(default=False, verbose_name='is control')),
('created_at', models.DateTimeField(auto_now_add=True, verbose_name='created at')),
('updated_at', models.DateTimeField(auto_now=True, verbose_name='updated at')),
],
options={
'verbose_name': 'variant',
'verbose_name_plural': 'variants',
'ordering': ['name'],
},
),
]
src/backend/apps/experiments/migrations/0002_initial.py
+86
View File
@@ -0,0 +1,86 @@
# Generated by Django 5.2.11 on 2026-02-14 09:55
import django.db.models.deletion
from django.conf import settings
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
('experiments', '0001_initial'),
('flags', '0001_initial'),
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.AddField(
model_name='approval',
name='approver',
field=models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='experiment_approvals', to=settings.AUTH_USER_MODEL, verbose_name='approver'),
),
migrations.AddField(
model_name='experiment',
name='flag',
field=models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='experiments', to='flags.featureflag', verbose_name='feature flag'),
),
migrations.AddField(
model_name='experiment',
name='owner',
field=models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='owned_experiments', to=settings.AUTH_USER_MODEL, verbose_name='owner'),
),
migrations.AddField(
model_name='experiment',
name='previous_version',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='next_versions', to='experiments.experiment', verbose_name='previous version'),
),
migrations.AddField(
model_name='approval',
name='experiment',
field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='approvals', to='experiments.experiment', verbose_name='experiment'),
),
migrations.AddField(
model_name='experimentlog',
name='experiment',
field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='logs', to='experiments.experiment', verbose_name='experiment'),
),
migrations.AddField(
model_name='experimentlog',
name='user',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='experiment_logs', to=settings.AUTH_USER_MODEL, verbose_name='user'),
),
migrations.AddField(
model_name='experimentoutcome',
name='decided_by',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='experiment_decisions', to=settings.AUTH_USER_MODEL, verbose_name='decided by'),
),
migrations.AddField(
model_name='experimentoutcome',
name='experiment',
field=models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='outcome', to='experiments.experiment', verbose_name='experiment'),
),
migrations.AddField(
model_name='variant',
name='experiment',
field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='variants', to='experiments.experiment', verbose_name='experiment'),
),
migrations.AddField(
model_name='experimentoutcome',
name='winning_variant',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='wins', to='experiments.variant', verbose_name='winning variant'),
),
migrations.AddIndex(
model_name='experiment',
index=models.Index(fields=['flag', 'status'], name='idx_experiment_flag_status'),
),
migrations.AddConstraint(
model_name='experiment',
constraint=models.UniqueConstraint(condition=models.Q(('status__in', ('running', 'paused'))), fields=('flag',), name='unique_active_experiment_per_flag'),
),
migrations.AddConstraint(
model_name='approval',
constraint=models.UniqueConstraint(fields=('experiment', 'approver'), name='unique_approval_per_user'),
),
]
src/backend/apps/experiments/migrations/__init__.py
View File
src/backend/apps/experiments/models.py
+557
View File
@@ -0,0 +1,557 @@
from decimal import Decimal
from typing import Any, override
from django.conf import settings
from django.core.exceptions import ValidationError
from django.core.validators import MaxValueValidator, MinValueValidator
from django.db import models
from django.utils.translation import gettext_lazy as _
from apps.core.models import BaseModel
class ExperimentStatus(models.TextChoices):
DRAFT = "draft", _("Draft")
IN_REVIEW = "in_review", _("In review")
APPROVED = "approved", _("Approved")
RUNNING = "running", _("Running")
PAUSED = "paused", _("Paused")
COMPLETED = "completed", _("Completed")
ARCHIVED = "archived", _("Archived")
REJECTED = "rejected", _("Rejected")
ALLOWED_TRANSITIONS: dict[str, tuple[str, ...]] = {
ExperimentStatus.DRAFT: (ExperimentStatus.IN_REVIEW,),
ExperimentStatus.IN_REVIEW: (
ExperimentStatus.APPROVED,
ExperimentStatus.DRAFT,
ExperimentStatus.REJECTED,
),
ExperimentStatus.APPROVED: (ExperimentStatus.RUNNING,),
ExperimentStatus.RUNNING: (
ExperimentStatus.PAUSED,
ExperimentStatus.COMPLETED,
),
ExperimentStatus.PAUSED: (
ExperimentStatus.RUNNING,
ExperimentStatus.COMPLETED,
),
ExperimentStatus.COMPLETED: (ExperimentStatus.ARCHIVED,),
ExperimentStatus.ARCHIVED: (),
ExperimentStatus.REJECTED: (ExperimentStatus.DRAFT,),
}
ACTIVE_STATUSES = (ExperimentStatus.RUNNING, ExperimentStatus.PAUSED)
STARTED_STATUSES = (
ExperimentStatus.RUNNING,
ExperimentStatus.PAUSED,
ExperimentStatus.COMPLETED,
ExperimentStatus.ARCHIVED,
)
FROZEN_AFTER_START: tuple[str, ...] = (
"traffic_allocation",
"targeting_rules",
"flag",
)
EDITABLE_IN_DRAFT: tuple[str, ...] = (
"name",
"description",
"hypothesis",
"traffic_allocation",
"targeting_rules",
"flag",
)
class ConflictPolicy(models.TextChoices):
MUTUAL_EXCLUSION = "mutual_exclusion", _("Mutual exclusion")
PRIORITY = "priority", _("Priority tiers")
class OutcomeType(models.TextChoices):
ROLLOUT = "rollout", _("Rollout winner")
ROLLBACK = "rollback", _("Rollback")
NO_EFFECT = "no_effect", _("No effect")
class TransitionValidator:
def __init__(self, instance: "Experiment") -> None:
self.instance = instance
def __call__(self) -> None:
if not self.instance.pk:
return
try:
old = Experiment.objects.get(pk=self.instance.pk)
except Experiment.DoesNotExist:
return
if old.status == self.instance.status:
return
allowed = ALLOWED_TRANSITIONS.get(old.status, ())
if self.instance.status not in allowed:
raise ValidationError(
{
"status": (
f"Transition from '{old.status}' to "
f"'{self.instance.status}' is not allowed. "
f"Allowed: {', '.join(allowed) or 'none'}."
)
}
)
class FrozenFieldsValidator:
def __init__(self, instance: "Experiment") -> None:
self.instance = instance
def __call__(self) -> None:
if not self.instance.pk:
return
try:
old = Experiment.objects.get(pk=self.instance.pk)
except Experiment.DoesNotExist:
return
if old.status not in STARTED_STATUSES:
return
errors: dict[str, str] = {}
for field_name in FROZEN_AFTER_START:
old_value = getattr(old, field_name)
new_value = getattr(self.instance, field_name)
if hasattr(old_value, "pk"):
changed = getattr(old_value, "pk", None) != getattr(
new_value, "pk", None
)
else:
changed = old_value != new_value
if changed:
errors[field_name] = (
f"Field '{field_name}' cannot be modified after "
f"the experiment has been started "
f"(current status: '{old.status}')."
)
if errors:
raise ValidationError(errors)
class UniqueActiveFlagValidator:
def __init__(self, instance: "Experiment") -> None:
self.instance = instance
def __call__(self) -> None:
if self.instance.status not in ACTIVE_STATUSES:
return
conflict = (
Experiment.objects.filter(
flag=self.instance.flag,
status__in=ACTIVE_STATUSES,
)
.exclude(pk=self.instance.pk)
.first()
)
if conflict:
raise ValidationError(
{
"flag": (
f"Flag '{self.instance.flag.key}' already has an "
f"active experiment: '{conflict.name}' "
f"(status: {conflict.status})."
)
}
)
class Experiment(BaseModel):
VALIDATORS = (
TransitionValidator,
FrozenFieldsValidator,
UniqueActiveFlagValidator,
)
flag = models.ForeignKey(
"flags.FeatureFlag",
on_delete=models.PROTECT,
related_name="experiments",
verbose_name=_("feature flag"),
)
name = models.CharField(
max_length=200,
verbose_name=_("name"),
)
description = models.TextField(
blank=True,
verbose_name=_("description"),
)
hypothesis = models.TextField(
blank=True,
verbose_name=_("hypothesis"),
)
status = models.CharField(
max_length=20,
choices=ExperimentStatus.choices,
default=ExperimentStatus.DRAFT,
db_index=True,
verbose_name=_("status"),
)
version = models.PositiveIntegerField(
default=1,
verbose_name=_("version"),
)
previous_version = models.ForeignKey(
"self",
null=True,
blank=True,
on_delete=models.SET_NULL,
related_name="next_versions",
verbose_name=_("previous version"),
)
traffic_allocation = models.DecimalField(
max_digits=5,
decimal_places=2,
default=Decimal("100.00"),
validators=[
MinValueValidator(Decimal("0.01")),
MaxValueValidator(Decimal("100.00")),
],
verbose_name=_("traffic allocation %"),
help_text=_("Percentage of eligible users included in experiment"),
)
targeting_rules = models.TextField(
blank=True,
verbose_name=_("targeting rules"),
help_text=_("DSL expression for targeting"),
)
owner = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.PROTECT,
related_name="owned_experiments",
verbose_name=_("owner"),
)
created_at = models.DateTimeField(
auto_now_add=True,
verbose_name=_("created at"),
)
updated_at = models.DateTimeField(
auto_now=True,
verbose_name=_("updated at"),
)
class Meta:
verbose_name = _("experiment")
verbose_name_plural = _("experiments")
ordering = ["-created_at"]
indexes = [
models.Index(
fields=["flag", "status"],
name="idx_experiment_flag_status",
),
]
constraints = [
models.UniqueConstraint(
fields=["flag"],
condition=models.Q(status__in=ACTIVE_STATUSES),
name="unique_active_experiment_per_flag",
),
]
@override
def __str__(self) -> str:
return f"{self.name} (v{self.version}, {self.status})"
@override
def save(self, *args: Any, **kwargs: Any) -> None:
for validator_cls in self.VALIDATORS:
validator_cls(self)()
super().save(*args, **kwargs)
class Variant(BaseModel):
experiment = models.ForeignKey(
Experiment,
on_delete=models.CASCADE,
related_name="variants",
verbose_name=_("experiment"),
)
name = models.CharField(
max_length=100,
verbose_name=_("name"),
)
value = models.CharField(
max_length=500,
verbose_name=_("value"),
help_text=_("Value returned when this variant is selected"),
)
weight = models.DecimalField(
max_digits=5,
decimal_places=2,
validators=[
MinValueValidator(Decimal("0.00")),
MaxValueValidator(Decimal("100.00")),
],
verbose_name=_("weight %"),
)
is_control = models.BooleanField(
default=False,
verbose_name=_("is control"),
)
created_at = models.DateTimeField(
auto_now_add=True,
verbose_name=_("created at"),
)
updated_at = models.DateTimeField(
auto_now=True,
verbose_name=_("updated at"),
)
class Meta:
verbose_name = _("variant")
verbose_name_plural = _("variants")
ordering = ["name"]
@override
def __str__(self) -> str:
control = " [control]" if self.is_control else ""
return f"{self.name}: {self.weight}%{control}"
@override
def save(self, *args: Any, **kwargs: Any) -> None:
VariantFrozenValidator(self)()
super().save(*args, **kwargs)
class VariantFrozenValidator:
def __init__(self, instance: Variant) -> None:
self.instance = instance
def __call__(self) -> None:
experiment = self.instance.experiment
if experiment.status in STARTED_STATUSES:
raise ValidationError(
{
"experiment": (
"Variants cannot be modified after the experiment "
f"has been started (status: '{experiment.status}')."
)
}
)
class LogType(models.TextChoices):
STATUS_CHANGE = "status_change", _("Status change")
APPROVAL = "approval", _("Approval")
REJECTION = "rejection", _("Rejection")
REVIEW_REQUESTED = "review_requested", _("Review requested")
GUARDRAIL_TRIGGERED = "guardrail_triggered", _("Guardrail triggered")
COMPLETED = "completed", _("Completed")
COMMENT = "comment", _("Comment")
class ExperimentLog(BaseModel):
experiment = models.ForeignKey(
Experiment,
on_delete=models.CASCADE,
related_name="logs",
verbose_name=_("experiment"),
)
log_type = models.CharField(
max_length=30,
choices=LogType.choices,
verbose_name=_("log type"),
)
user = models.ForeignKey(
settings.AUTH_USER_MODEL,
null=True,
blank=True,
on_delete=models.SET_NULL,
related_name="experiment_logs",
verbose_name=_("user"),
)
comment = models.TextField(
blank=True,
verbose_name=_("comment"),
)
metadata = models.JSONField(
default=dict,
blank=True,
verbose_name=_("metadata"),
)
created_at = models.DateTimeField(
auto_now_add=True,
verbose_name=_("created at"),
)
class Meta:
verbose_name = _("experiment log")
verbose_name_plural = _("experiment logs")
ordering = ["-created_at"]
@override
def __str__(self) -> str:
return f"[{self.log_type}] {self.experiment.name}"
class Approval(BaseModel):
experiment = models.ForeignKey(
Experiment,
on_delete=models.CASCADE,
related_name="approvals",
verbose_name=_("experiment"),
)
approver = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.PROTECT,
related_name="experiment_approvals",
verbose_name=_("approver"),
)
comment = models.TextField(
blank=True,
verbose_name=_("comment"),
)
created_at = models.DateTimeField(
auto_now_add=True,
verbose_name=_("created at"),
)
class Meta:
verbose_name = _("approval")
verbose_name_plural = _("approvals")
ordering = ["-created_at"]
constraints = [
models.UniqueConstraint(
fields=["experiment", "approver"],
name="unique_approval_per_user",
),
]
@override
def __str__(self) -> str:
return f"Approval by {self.approver} for {self.experiment}"
class ConflictDomain(BaseModel):
name = models.CharField(
max_length=200,
unique=True,
verbose_name=_("name"),
)
description = models.TextField(
blank=True,
verbose_name=_("description"),
)
policy = models.CharField(
max_length=30,
choices=ConflictPolicy.choices,
default=ConflictPolicy.MUTUAL_EXCLUSION,
verbose_name=_("conflict policy"),
)
max_concurrent = models.PositiveIntegerField(
default=1,
verbose_name=_("max concurrent experiments"),
)
created_at = models.DateTimeField(
auto_now_add=True,
verbose_name=_("created at"),
)
updated_at = models.DateTimeField(
auto_now=True,
verbose_name=_("updated at"),
)
class Meta:
verbose_name = _("conflict domain")
verbose_name_plural = _("conflict domains")
ordering = ["name"]
@override
def __str__(self) -> str:
return f"{self.name} ({self.policy})"
class ExperimentConflictDomain(BaseModel):
experiment = models.ForeignKey(
Experiment,
on_delete=models.CASCADE,
related_name="conflict_memberships",
verbose_name=_("experiment"),
)
conflict_domain = models.ForeignKey(
ConflictDomain,
on_delete=models.CASCADE,
related_name="experiment_memberships",
verbose_name=_("conflict domain"),
)
priority = models.IntegerField(
default=0,
verbose_name=_("priority"),
help_text=_("Higher value wins in priority-based resolution"),
)
created_at = models.DateTimeField(
auto_now_add=True,
verbose_name=_("created at"),
)
class Meta:
verbose_name = _("experiment conflict domain")
verbose_name_plural = _("experiment conflict domains")
ordering = ["-priority"]
constraints = [
models.UniqueConstraint(
fields=["experiment", "conflict_domain"],
name="unique_experiment_conflict_domain",
),
]
@override
def __str__(self) -> str:
return (
f"{self.experiment.name} in {self.conflict_domain.name} "
f"(priority={self.priority})"
)
class ExperimentOutcome(BaseModel):
experiment = models.OneToOneField(
Experiment,
on_delete=models.CASCADE,
related_name="outcome",
verbose_name=_("experiment"),
)
outcome = models.CharField(
max_length=20,
choices=OutcomeType.choices,
verbose_name=_("outcome"),
)
winning_variant = models.ForeignKey(
Variant,
null=True,
blank=True,
on_delete=models.SET_NULL,
related_name="wins",
verbose_name=_("winning variant"),
)
rationale = models.TextField(
verbose_name=_("rationale"),
)
decided_by = models.ForeignKey(
settings.AUTH_USER_MODEL,
null=True,
blank=True,
on_delete=models.SET_NULL,
related_name="experiment_decisions",
verbose_name=_("decided by"),
)
created_at = models.DateTimeField(
auto_now_add=True,
verbose_name=_("created at"),
)
class Meta:
verbose_name = _("experiment outcome")
verbose_name_plural = _("experiment outcomes")
@override
def __str__(self) -> str:
return f"{self.experiment.name}: {self.outcome}"
src/backend/apps/experiments/selectors.py
+69
View File
@@ -0,0 +1,69 @@
from uuid import UUID
from django.db.models import QuerySet
from apps.experiments.models import (
Approval,
Experiment,
ExperimentLog,
ExperimentStatus,
Variant,
)
def experiment_list(
*,
status: str | None = None,
flag_id: UUID | None = None,
owner_id: UUID | None = None,
search: str | None = None,
) -> QuerySet[Experiment]:
qs = Experiment.objects.select_related("flag", "owner").prefetch_related(
"variants"
)
if status:
qs = qs.filter(status=status)
if flag_id:
qs = qs.filter(flag_id=flag_id)
if owner_id:
qs = qs.filter(owner_id=owner_id)
if search:
qs = qs.filter(name__icontains=search)
return qs
def experiment_get(experiment_id: UUID) -> Experiment | None:
return (
Experiment.objects.select_related("flag", "owner")
.prefetch_related("variants", "approvals__approver", "logs")
.filter(pk=experiment_id)
.first()
)
def experiment_logs(experiment_id: UUID) -> QuerySet[ExperimentLog]:
return ExperimentLog.objects.filter(
experiment_id=experiment_id,
).select_related("user")
def experiment_approvals(experiment_id: UUID) -> QuerySet[Approval]:
return Approval.objects.filter(
experiment_id=experiment_id,
).select_related("approver")
def variant_list(experiment_id: UUID) -> QuerySet[Variant]:
return Variant.objects.filter(experiment_id=experiment_id)
def active_experiment_for_flag(flag_id: UUID) -> Experiment | None:
return (
Experiment.objects.filter(
flag_id=flag_id,
status=ExperimentStatus.RUNNING,
)
.select_related("flag")
.prefetch_related("variants")
.first()
)
src/backend/apps/experiments/services.py
+495
View File
@@ -0,0 +1,495 @@
from decimal import Decimal
from typing import Any
from django.core.exceptions import ValidationError
from django.db import transaction
from apps.experiments.models import (
ACTIVE_STATUSES,
ALLOWED_TRANSITIONS,
Approval,
Experiment,
ExperimentLog,
ExperimentOutcome,
ExperimentStatus,
LogType,
OutcomeType,
Variant,
)
from apps.flags.models import FeatureFlag, validate_value_for_type
from apps.notifications.services import (
NotificationPayload,
notification_enqueue,
)
from apps.reviews.selectors import (
can_user_approve_experimenter,
get_min_approvals_for_experimenter,
)
from apps.users.models import User
from config.errors import ForbiddenError
def _notify(event_type: str, experiment: Experiment, extra: dict[str, Any] | None = None) -> None:
notification_enqueue(
event_type,
NotificationPayload(
title=f"{event_type.replace('_', ' ').title()}",
body=f"Experiment '{experiment.name}' — {event_type.replace('_', ' ')}.",
event_type=event_type,
experiment_id=str(experiment.pk),
experiment_name=experiment.name,
extra=extra or {},
),
)
def ensure_owner_or_admin(experiment: Experiment, user: User) -> None:
if user.is_admin_role:
return
if experiment.owner_id != user.pk:
raise ForbiddenError(
"Only the experiment owner or an admin can perform this action."
)
@transaction.atomic
def experiment_create(
*,
flag: FeatureFlag,
name: str,
owner: User,
description: str = "",
hypothesis: str = "",
traffic_allocation: Decimal = Decimal("100.00"),
targeting_rules: str = "",
) -> Experiment:
experiment = Experiment(
flag=flag,
name=name,
owner=owner,
description=description,
hypothesis=hypothesis,
traffic_allocation=traffic_allocation,
targeting_rules=targeting_rules,
status=ExperimentStatus.DRAFT,
version=1,
)
experiment.save()
ExperimentLog.objects.create(
experiment=experiment,
log_type=LogType.STATUS_CHANGE,
user=owner,
metadata={"to_status": ExperimentStatus.DRAFT},
)
return experiment
@transaction.atomic
def experiment_update(
*,
experiment: Experiment,
user: User,
**fields: Any,
) -> Experiment:
ensure_owner_or_admin(experiment, user)
allowed_fields = {
"name",
"description",
"hypothesis",
"traffic_allocation",
"targeting_rules",
}
for key in fields:
if key not in allowed_fields:
raise ValidationError(
{key: f"Field '{key}' cannot be updated via this endpoint."}
)
for key, value in fields.items():
if value is not None:
setattr(experiment, key, value)
experiment.save()
return experiment
@transaction.atomic
def variant_create(
*,
experiment: Experiment,
user: User,
name: str,
value: str,
weight: Decimal,
is_control: bool = False,
) -> Variant:
ensure_owner_or_admin(experiment, user)
validate_value_for_type(value, experiment.flag.value_type)
variant = Variant(
experiment=experiment,
name=name,
value=value,
weight=weight,
is_control=is_control,
)
variant.save()
_validate_variant_invariants(experiment)
return variant
@transaction.atomic
def variant_update(
*,
variant: Variant,
user: User,
name: str | None = None,
value: str | None = None,
weight: Decimal | None = None,
is_control: bool | None = None,
) -> Variant:
ensure_owner_or_admin(variant.experiment, user)
if name is not None:
variant.name = name
if value is not None:
validate_value_for_type(value, variant.experiment.flag.value_type)
variant.value = value
if weight is not None:
variant.weight = weight
if is_control is not None:
variant.is_control = is_control
variant.save()
_validate_variant_invariants(variant.experiment)
return variant
@transaction.atomic
def variant_delete(*, variant: Variant, user: User) -> None:
ensure_owner_or_admin(variant.experiment, user)
experiment = variant.experiment
if experiment.status != ExperimentStatus.DRAFT:
raise ValidationError(
{
"experiment": (
"Variants can only be deleted while the experiment "
f"is in '{ExperimentStatus.DRAFT}' status."
)
}
)
variant.delete()
def _validate_variant_invariants(experiment: Experiment) -> None:
variants = experiment.variants.all()
if not variants.exists():
return
control_count = variants.filter(is_control=True).count()
if control_count > 1:
raise ValidationError(
{"is_control": "Exactly one variant must be the control."}
)
total_weight = sum(v.weight for v in variants)
if total_weight > experiment.traffic_allocation:
raise ValidationError(
{
"weight": (
f"Sum of variant weights ({total_weight}%) exceeds "
f"traffic allocation ({experiment.traffic_allocation}%)."
)
}
)
def _validate_experiment_ready_for_review(experiment: Experiment) -> None:
variants = experiment.variants.all()
if variants.count() < 2:
raise ValidationError(
{"variants": "At least two variants are required."}
)
control_count = variants.filter(is_control=True).count()
if control_count != 1:
raise ValidationError(
{"is_control": "Exactly one variant must be the control."}
)
total_weight = sum(v.weight for v in variants)
if total_weight != experiment.traffic_allocation:
raise ValidationError(
{
"weight": (
f"Sum of variant weights ({total_weight}%) must equal "
f"traffic allocation ({experiment.traffic_allocation}%)."
)
}
)
def _validate_no_active_flag_conflict(experiment: Experiment) -> None:
conflict = (
Experiment.objects.filter(
flag=experiment.flag,
status__in=ACTIVE_STATUSES,
)
.exclude(pk=experiment.pk)
.first()
)
if conflict:
raise ValidationError(
{
"flag": (
f"Flag '{experiment.flag.key}' already has an active "
f"experiment: '{conflict.name}'."
)
}
)
def _transition(
experiment: Experiment,
new_status: str,
user: User,
log_type: str = LogType.STATUS_CHANGE,
comment: str = "",
metadata: dict[str, Any] | None = None,
) -> Experiment:
old_status = experiment.status
allowed = ALLOWED_TRANSITIONS.get(old_status, ())
if new_status not in allowed:
raise ValidationError(
{
"status": (
f"Transition from '{old_status}' to '{new_status}' "
f"is not allowed."
)
}
)
experiment.status = new_status
experiment.save(update_fields=["status", "updated_at"])
ExperimentLog.objects.create(
experiment=experiment,
log_type=log_type,
user=user,
comment=comment,
metadata={
"from_status": old_status,
"to_status": new_status,
**(metadata or {}),
},
)
return experiment
@transaction.atomic
def experiment_submit_for_review(
*, experiment: Experiment, user: User
) -> Experiment:
ensure_owner_or_admin(experiment, user)
_validate_experiment_ready_for_review(experiment)
experiment = _transition(
experiment,
ExperimentStatus.IN_REVIEW,
user,
log_type=LogType.REVIEW_REQUESTED,
)
_notify("review_requested", experiment)
return experiment
@transaction.atomic
def experiment_approve(
*, experiment: Experiment, approver: User, comment: str = ""
) -> Experiment:
if experiment.status != ExperimentStatus.IN_REVIEW:
raise ValidationError(
{"status": "Experiment must be in 'in_review' status to approve."}
)
if not can_user_approve_experimenter(approver, experiment.owner):
raise ValidationError(
{
"approver": (
f"User '{approver.username}' is not authorized to "
f"approve experiments for '{experiment.owner.username}'."
)
}
)
if experiment.approvals.filter(approver=approver).exists():
raise ValidationError(
{"approver": "You have already approved this experiment."}
)
Approval.objects.create(
experiment=experiment,
approver=approver,
comment=comment,
)
ExperimentLog.objects.create(
experiment=experiment,
log_type=LogType.APPROVAL,
user=approver,
comment=comment,
metadata={"approver": str(approver.pk)},
)
approval_count = experiment.approvals.count()
min_approvals = get_min_approvals_for_experimenter(experiment.owner)
if approval_count >= min_approvals:
experiment = _transition(
experiment,
ExperimentStatus.APPROVED,
approver,
metadata={"approval_count": approval_count},
)
_notify("review_approved", experiment)
return experiment
return experiment
@transaction.atomic
def experiment_reject(
*, experiment: Experiment, user: User, comment: str = ""
) -> Experiment:
experiment = _transition(
experiment,
ExperimentStatus.REJECTED,
user,
log_type=LogType.REJECTION,
comment=comment,
)
_notify("review_rejected", experiment)
return experiment
@transaction.atomic
def experiment_request_changes(
*, experiment: Experiment, user: User, comment: str = ""
) -> Experiment:
experiment.approvals.all().delete()
return _transition(
experiment,
ExperimentStatus.DRAFT,
user,
comment=comment,
metadata={"action": "request_changes"},
)
@transaction.atomic
def experiment_start(*, experiment: Experiment, user: User) -> Experiment:
ensure_owner_or_admin(experiment, user)
_validate_no_active_flag_conflict(experiment)
experiment = _transition(
experiment,
ExperimentStatus.RUNNING,
user,
)
_notify("experiment_started", experiment)
return experiment
@transaction.atomic
def experiment_pause(
*, experiment: Experiment, user: User, comment: str = ""
) -> Experiment:
ensure_owner_or_admin(experiment, user)
experiment = _transition(
experiment,
ExperimentStatus.PAUSED,
user,
comment=comment,
)
_notify("experiment_paused", experiment, extra={"comment": comment})
return experiment
@transaction.atomic
def experiment_resume(*, experiment: Experiment, user: User) -> Experiment:
ensure_owner_or_admin(experiment, user)
_validate_no_active_flag_conflict(experiment)
return _transition(
experiment,
ExperimentStatus.RUNNING,
user,
)
@transaction.atomic
def experiment_complete(
*,
experiment: Experiment,
user: User,
outcome: str,
rationale: str,
winning_variant_id: str | None = None,
) -> Experiment:
ensure_owner_or_admin(experiment, user)
valid_outcomes = {c[0] for c in OutcomeType.choices}
if outcome not in valid_outcomes:
raise ValidationError(
{
"outcome": (
f"Invalid outcome '{outcome}'. "
f"Must be one of: {', '.join(sorted(valid_outcomes))}."
)
}
)
if not rationale.strip():
raise ValidationError(
{"rationale": "Rationale is required when completing."}
)
winning_variant = None
if outcome == OutcomeType.ROLLOUT:
if not winning_variant_id:
raise ValidationError(
{"winning_variant_id": "Winner must be specified for rollout."}
)
winning_variant = experiment.variants.filter(
pk=winning_variant_id
).first()
if not winning_variant:
raise ValidationError(
{"winning_variant_id": "Variant not found in this experiment."}
)
experiment = _transition(
experiment,
ExperimentStatus.COMPLETED,
user,
log_type=LogType.COMPLETED,
comment=rationale,
metadata={
"outcome": outcome,
"winning_variant_id": str(winning_variant.pk)
if winning_variant
else None,
},
)
ExperimentOutcome.objects.create(
experiment=experiment,
outcome=outcome,
winning_variant=winning_variant,
rationale=rationale,
decided_by=user,
)
_notify(
"experiment_completed",
experiment,
extra={"outcome": outcome, "rationale": rationale},
)
return experiment
@transaction.atomic
def experiment_archive(*, experiment: Experiment, user: User) -> Experiment:
ensure_owner_or_admin(experiment, user)
return _transition(
experiment,
ExperimentStatus.ARCHIVED,
user,
)
@transaction.atomic
def experiment_reopen(*, experiment: Experiment, user: User) -> Experiment:
ensure_owner_or_admin(experiment, user)
experiment.approvals.all().delete()
return _transition(
experiment,
ExperimentStatus.DRAFT,
user,
metadata={"action": "reopen_from_rejected"},
)
src/backend/apps/experiments/tests/__init__.py
View File
src/backend/apps/experiments/tests/helpers.py
+65
View File
@@ -0,0 +1,65 @@
from decimal import Decimal
from apps.experiments.models import Experiment, Variant
from apps.experiments.services import (
experiment_create,
variant_create,
)
from apps.flags.models import FeatureFlagType
from apps.flags.services import feature_flag_create
from apps.users.tests.helpers import make_user
def make_flag(suffix="", value_type=FeatureFlagType.STRING, default="a"):
return feature_flag_create(
key=f"flag{suffix}",
name=f"Flag {suffix}",
value_type=value_type,
default_value=default,
)
def make_experiment(
flag=None,
owner=None,
suffix="",
name=None,
**kwargs,
) -> Experiment:
if not flag:
flag = make_flag(suffix=suffix)
if not owner:
owner = make_user(
username=f"owner{suffix}",
email=f"owner{suffix}@lotty.local",
)
return experiment_create(
flag=flag,
name=name or f"Exp{suffix}",
owner=owner,
**kwargs,
)
def add_two_variants(
experiment: Experiment,
weight_a: Decimal = Decimal("50.00"),
weight_b: Decimal = Decimal("50.00"),
) -> tuple[Variant, Variant]:
v_control = variant_create(
experiment=experiment,
user=experiment.owner,
name="control",
value="a",
weight=weight_a,
is_control=True,
)
v_treatment = variant_create(
experiment=experiment,
user=experiment.owner,
name="treatment",
value="b",
weight=weight_b,
is_control=False,
)
return v_control, v_treatment
src/backend/apps/experiments/tests/test_models.py
+179
View File
@@ -0,0 +1,179 @@
from decimal import Decimal
from django.core.exceptions import ValidationError
from django.test import TestCase
from apps.experiments.models import (
ALLOWED_TRANSITIONS,
FROZEN_AFTER_START,
STARTED_STATUSES,
Experiment,
ExperimentStatus,
Variant,
)
from apps.experiments.tests.helpers import (
add_two_variants,
make_experiment,
make_flag,
)
from apps.users.tests.helpers import make_user
class TransitionValidatorTest(TestCase):
def test_all_allowed_transitions_succeed(self) -> None:
exp = make_experiment(suffix="_tr")
for from_status, targets in ALLOWED_TRANSITIONS.items():
for target in targets:
Experiment.objects.filter(pk=exp.pk).update(status=from_status)
exp.refresh_from_db()
exp.status = target
exp.save()
exp.refresh_from_db()
self.assertEqual(exp.status, target)
def test_forbidden_transition_raises(self) -> None:
exp = make_experiment(suffix="_fb")
Experiment.objects.filter(pk=exp.pk).update(
status=ExperimentStatus.DRAFT
)
exp.refresh_from_db()
exp.status = ExperimentStatus.RUNNING
with self.assertRaises(ValidationError) as ctx:
exp.save()
self.assertIn("status", ctx.exception.message_dict)
def test_same_status_save_allowed(self) -> None:
exp = make_experiment(suffix="_ss")
exp.name = "Updated name"
exp.save()
exp.refresh_from_db()
self.assertEqual(exp.name, "Updated name")
class FrozenFieldsValidatorTest(TestCase):
def test_frozen_fields_after_start(self) -> None:
exp = make_experiment(
suffix="_ff",
traffic_allocation=Decimal("100.00"),
)
add_two_variants(exp)
Experiment.objects.filter(pk=exp.pk).update(
status=ExperimentStatus.RUNNING
)
exp.refresh_from_db()
for field_name in FROZEN_AFTER_START:
exp_copy = Experiment.objects.get(pk=exp.pk)
if field_name == "traffic_allocation":
exp_copy.traffic_allocation = Decimal("10.00")
elif field_name == "targeting_rules":
exp_copy.targeting_rules = "changed"
elif field_name == "flag":
exp_copy.flag = make_flag(suffix=f"_alt_{field_name}")
with self.assertRaises(
ValidationError, msg=f"Expected freeze on {field_name}"
):
exp_copy.save()
def test_non_frozen_fields_editable_after_start(self) -> None:
exp = make_experiment(suffix="_nf")
Experiment.objects.filter(pk=exp.pk).update(
status=ExperimentStatus.RUNNING
)
exp.refresh_from_db()
exp.name = "New Name"
exp.description = "New Desc"
exp.save()
exp.refresh_from_db()
self.assertEqual(exp.name, "New Name")
def test_frozen_fields_editable_in_draft(self) -> None:
exp = make_experiment(suffix="_dft")
exp.traffic_allocation = Decimal("25.00")
exp.targeting_rules = "age > 18"
exp.save()
exp.refresh_from_db()
self.assertEqual(exp.traffic_allocation, Decimal("25.00"))
def test_frozen_in_all_started_statuses(self) -> None:
for status in STARTED_STATUSES:
exp = make_experiment(suffix=f"_s{status}")
Experiment.objects.filter(pk=exp.pk).update(status=status)
exp.refresh_from_db()
exp.traffic_allocation = Decimal("1.00")
with self.assertRaises(
ValidationError, msg=f"Expected freeze in {status}"
):
exp.save()
class UniqueActiveFlagValidatorTest(TestCase):
def test_two_running_experiments_same_flag_rejected(self) -> None:
flag = make_flag(suffix="_uaf")
owner = make_user(username="uaf_owner", email="uaf_owner@lotty.local")
exp1 = make_experiment(flag=flag, owner=owner, suffix="_u1", name="E1")
Experiment.objects.filter(pk=exp1.pk).update(
status=ExperimentStatus.RUNNING
)
exp2 = make_experiment(flag=flag, owner=owner, suffix="_u2", name="E2")
Experiment.objects.filter(pk=exp2.pk).update(
status=ExperimentStatus.APPROVED
)
exp2.refresh_from_db()
exp2.status = ExperimentStatus.RUNNING
with self.assertRaises(ValidationError) as ctx:
exp2.save()
self.assertIn("flag", ctx.exception.message_dict)
def test_different_flags_both_running_allowed(self) -> None:
owner = make_user(username="dfb_owner", email="dfb_owner@lotty.local")
exp1 = make_experiment(owner=owner, suffix="_df1")
exp2 = make_experiment(owner=owner, suffix="_df2")
Experiment.objects.filter(pk=exp1.pk).update(
status=ExperimentStatus.RUNNING
)
Experiment.objects.filter(pk=exp2.pk).update(
status=ExperimentStatus.APPROVED
)
exp2.refresh_from_db()
exp2.status = ExperimentStatus.RUNNING
exp2.save()
exp2.refresh_from_db()
self.assertEqual(exp2.status, ExperimentStatus.RUNNING)
class VariantFrozenValidatorTest(TestCase):
def test_variant_not_editable_after_start(self) -> None:
exp = make_experiment(suffix="_vf")
_vc, vt = add_two_variants(exp)
Experiment.objects.filter(pk=exp.pk).update(
status=ExperimentStatus.RUNNING
)
vt.refresh_from_db()
vt.name = "changed"
with self.assertRaises(ValidationError):
vt.save()
def test_variant_editable_in_draft(self) -> None:
exp = make_experiment(suffix="_ved")
_vc, vt = add_two_variants(exp)
vt.name = "renamed"
vt.save()
vt.refresh_from_db()
self.assertEqual(vt.name, "renamed")
def test_new_variant_rejected_after_start(self) -> None:
exp = make_experiment(suffix="_vnr")
add_two_variants(exp)
Experiment.objects.filter(pk=exp.pk).update(
status=ExperimentStatus.RUNNING
)
exp.refresh_from_db()
with self.assertRaises(ValidationError):
Variant(
experiment=exp,
name="extra",
value="c",
weight=Decimal("10.00"),
).save()
src/backend/apps/experiments/tests/test_services.py
+585
View File
@@ -0,0 +1,585 @@
from decimal import Decimal
from typing import override
from django.core.exceptions import ValidationError
from django.test import TestCase
from apps.experiments.models import (
Approval,
Experiment,
ExperimentLog,
ExperimentOutcome,
ExperimentStatus,
LogType,
OutcomeType,
)
from apps.experiments.services import (
ensure_owner_or_admin,
experiment_approve,
experiment_archive,
experiment_complete,
experiment_create,
experiment_pause,
experiment_reject,
experiment_reopen,
experiment_request_changes,
experiment_resume,
experiment_start,
experiment_submit_for_review,
experiment_update,
variant_create,
variant_delete,
variant_update,
)
from apps.experiments.tests.helpers import (
add_two_variants,
make_experiment,
make_flag,
)
from apps.reviews.services import approver_group_create, review_settings_update
from apps.reviews.tests.helpers import (
make_admin,
make_approver,
make_experimenter,
make_viewer,
)
from config.errors import ForbiddenError
class ExperimentCreateTest(TestCase):
def test_creates_with_defaults(self) -> None:
exp = make_experiment(suffix="_cw")
self.assertEqual(exp.status, ExperimentStatus.DRAFT)
self.assertEqual(exp.version, 1)
self.assertEqual(exp.traffic_allocation, Decimal("100.00"))
def test_creates_log_entry(self) -> None:
exp = make_experiment(suffix="_cl")
logs = ExperimentLog.objects.filter(experiment=exp)
self.assertEqual(logs.count(), 1)
self.assertEqual(logs.first().log_type, LogType.STATUS_CHANGE)
class ExperimentUpdateTest(TestCase):
def test_updates_allowed_fields(self) -> None:
exp = make_experiment(suffix="_uf")
exp = experiment_update(
experiment=exp,
user=exp.owner,
name="Updated",
description="desc",
)
exp.refresh_from_db()
self.assertEqual(exp.name, "Updated")
self.assertEqual(exp.description, "desc")
def test_rejects_disallowed_field(self) -> None:
exp = make_experiment(suffix="_rd")
with self.assertRaises(ValidationError):
experiment_update(
experiment=exp,
user=exp.owner,
status="running",
)
class VariantCrudTest(TestCase):
def test_create_variants(self) -> None:
exp = make_experiment(suffix="_vc")
vc, vt = add_two_variants(exp)
self.assertTrue(vc.is_control)
self.assertFalse(vt.is_control)
def test_total_weight_exceeds_allocation_raises(self) -> None:
exp = make_experiment(
suffix="_twe", traffic_allocation=Decimal("50.00")
)
variant_create(
experiment=exp,
user=exp.owner,
name="control",
value="a",
weight=Decimal("30.00"),
is_control=True,
)
with self.assertRaises(ValidationError) as ctx:
variant_create(
experiment=exp,
user=exp.owner,
name="treatment",
value="b",
weight=Decimal("30.00"),
)
self.assertIn("weight", ctx.exception.message_dict)
def test_two_controls_raises(self) -> None:
exp = make_experiment(suffix="_2c")
variant_create(
experiment=exp,
user=exp.owner,
name="c1",
value="a",
weight=Decimal("20.00"),
is_control=True,
)
with self.assertRaises(ValidationError) as ctx:
variant_create(
experiment=exp,
user=exp.owner,
name="c2",
value="b",
weight=Decimal("20.00"),
is_control=True,
)
self.assertIn("is_control", ctx.exception.message_dict)
def test_update_variant(self) -> None:
exp = make_experiment(suffix="_vu")
_vc, vt = add_two_variants(exp)
vt = variant_update(variant=vt, user=exp.owner, name="renamed")
self.assertEqual(vt.name, "renamed")
def test_delete_variant_in_draft(self) -> None:
exp = make_experiment(suffix="_vd")
_vc, vt = add_two_variants(exp)
variant_delete(variant=vt, user=exp.owner)
self.assertEqual(exp.variants.count(), 1)
def test_delete_variant_not_in_draft_raises(self) -> None:
exp = make_experiment(suffix="_vnd")
_vc, vt = add_two_variants(exp)
Experiment.objects.filter(pk=exp.pk).update(
status=ExperimentStatus.IN_REVIEW
)
vt.refresh_from_db()
with self.assertRaises(ValidationError):
variant_delete(variant=vt, user=exp.owner)
class SubmitForReviewTest(TestCase):
def test_submit_with_valid_variants(self) -> None:
exp = make_experiment(suffix="_sr")
add_two_variants(exp)
exp = experiment_submit_for_review(experiment=exp, user=exp.owner)
self.assertEqual(exp.status, ExperimentStatus.IN_REVIEW)
def test_submit_without_variants_raises(self) -> None:
exp = make_experiment(suffix="_snv")
with self.assertRaises(ValidationError) as ctx:
experiment_submit_for_review(experiment=exp, user=exp.owner)
self.assertIn("variants", ctx.exception.message_dict)
def test_submit_without_control_raises(self) -> None:
exp = make_experiment(suffix="_snc")
variant_create(
experiment=exp,
user=exp.owner,
name="v1",
value="a",
weight=Decimal("50.00"),
is_control=False,
)
variant_create(
experiment=exp,
user=exp.owner,
name="v2",
value="b",
weight=Decimal("50.00"),
is_control=False,
)
with self.assertRaises(ValidationError) as ctx:
experiment_submit_for_review(experiment=exp, user=exp.owner)
self.assertIn("is_control", ctx.exception.message_dict)
def test_submit_weights_not_equal_allocation_raises(self) -> None:
exp = make_experiment(suffix="_sne")
variant_create(
experiment=exp,
user=exp.owner,
name="c",
value="a",
weight=Decimal("30.00"),
is_control=True,
)
variant_create(
experiment=exp,
user=exp.owner,
name="t",
value="b",
weight=Decimal("20.00"),
)
with self.assertRaises(ValidationError) as ctx:
experiment_submit_for_review(experiment=exp, user=exp.owner)
self.assertIn("weight", ctx.exception.message_dict)
def test_submit_one_variant_raises(self) -> None:
exp = make_experiment(suffix="_s1v")
variant_create(
experiment=exp,
user=exp.owner,
name="c",
value="a",
weight=Decimal("100.00"),
is_control=True,
)
with self.assertRaises(ValidationError) as ctx:
experiment_submit_for_review(experiment=exp, user=exp.owner)
self.assertIn("variants", ctx.exception.message_dict)
class ApprovalFlowTest(TestCase):
@override
def setUp(self) -> None:
self.experimenter = make_experimenter("_af")
self.approver1 = make_approver("_af1")
self.approver2 = make_approver("_af2")
self.flag = make_flag(suffix="_af")
approver_group_create(
experimenter=self.experimenter,
approver_ids=[
str(self.approver1.pk),
str(self.approver2.pk),
],
min_approvals=2,
)
self.exp = experiment_create(
flag=self.flag,
name="AF Exp",
owner=self.experimenter,
)
add_two_variants(self.exp)
self.exp = experiment_submit_for_review(
experiment=self.exp, user=self.experimenter
)
def test_first_approval_stays_in_review(self) -> None:
exp = experiment_approve(
experiment=self.exp,
approver=self.approver1,
comment="lgtm",
)
self.assertEqual(exp.status, ExperimentStatus.IN_REVIEW)
self.assertEqual(Approval.objects.filter(experiment=exp).count(), 1)
def test_second_approval_transitions_to_approved(self) -> None:
experiment_approve(
experiment=self.exp,
approver=self.approver1,
)
exp = experiment_approve(
experiment=self.exp,
approver=self.approver2,
)
self.assertEqual(exp.status, ExperimentStatus.APPROVED)
def test_duplicate_approval_raises(self) -> None:
experiment_approve(
experiment=self.exp,
approver=self.approver1,
)
with self.assertRaises(ValidationError) as ctx:
experiment_approve(
experiment=self.exp,
approver=self.approver1,
)
self.assertIn("approver", ctx.exception.message_dict)
def test_unauthorized_approver_raises(self) -> None:
viewer = make_viewer("_unauth")
with self.assertRaises(ValidationError) as ctx:
experiment_approve(
experiment=self.exp,
approver=viewer,
)
self.assertIn("approver", ctx.exception.message_dict)
class RejectAndReqChangesTest(TestCase):
@override
def setUp(self) -> None:
self.experimenter = make_experimenter("_rr")
self.approver1 = make_approver("_rr1")
self.approver2 = make_approver("_rr2")
self.flag = make_flag(suffix="_rr")
approver_group_create(
experimenter=self.experimenter,
approver_ids=[
str(self.approver1.pk),
str(self.approver2.pk),
],
min_approvals=2,
)
self.exp = experiment_create(
flag=self.flag,
name="RR Exp",
owner=self.experimenter,
)
add_two_variants(self.exp)
self.exp = experiment_submit_for_review(
experiment=self.exp, user=self.experimenter
)
def test_reject(self) -> None:
exp = experiment_reject(
experiment=self.exp,
user=self.approver1,
comment="not ready",
)
self.assertEqual(exp.status, ExperimentStatus.REJECTED)
def test_request_changes_clears_approvals(self) -> None:
experiment_approve(
experiment=self.exp,
approver=self.approver1,
)
self.exp.refresh_from_db()
exp = experiment_request_changes(
experiment=self.exp, user=self.approver2, comment="fix please"
)
self.assertEqual(exp.status, ExperimentStatus.DRAFT)
self.assertEqual(Approval.objects.filter(experiment=exp).count(), 0)
def test_reopen_from_rejected(self) -> None:
exp = experiment_reject(
experiment=self.exp,
user=self.approver1,
)
exp = experiment_reopen(experiment=exp, user=self.experimenter)
self.assertEqual(exp.status, ExperimentStatus.DRAFT)
class LifecycleFlowTest(TestCase):
@override
def setUp(self) -> None:
self.experimenter = make_experimenter("_lf")
self.approver = make_approver("_lf")
self.flag = make_flag(suffix="_lf")
review_settings_update(
default_min_approvals=1, allow_any_approver=True
)
self.exp = experiment_create(
flag=self.flag,
name="LF Exp",
owner=self.experimenter,
)
self.vc, self.vt = add_two_variants(self.exp)
def _approve_and_start(self) -> Experiment:
exp = experiment_submit_for_review(
experiment=self.exp, user=self.experimenter
)
exp = experiment_approve(experiment=exp, approver=self.approver)
return experiment_start(experiment=exp, user=self.experimenter)
def test_full_lifecycle_to_complete(self) -> None:
exp = self._approve_and_start()
self.assertEqual(exp.status, ExperimentStatus.RUNNING)
exp = experiment_pause(experiment=exp, user=self.experimenter)
self.assertEqual(exp.status, ExperimentStatus.PAUSED)
exp = experiment_resume(experiment=exp, user=self.experimenter)
self.assertEqual(exp.status, ExperimentStatus.RUNNING)
exp = experiment_complete(
experiment=exp,
user=self.experimenter,
outcome=OutcomeType.ROLLOUT,
rationale="Treatment won",
winning_variant_id=str(self.vt.pk),
)
self.assertEqual(exp.status, ExperimentStatus.COMPLETED)
outcome = ExperimentOutcome.objects.get(experiment=exp)
self.assertEqual(outcome.outcome, OutcomeType.ROLLOUT)
self.assertEqual(outcome.winning_variant, self.vt)
def test_complete_rollback(self) -> None:
exp = self._approve_and_start()
exp = experiment_complete(
experiment=exp,
user=self.experimenter,
outcome=OutcomeType.ROLLBACK,
rationale="No improvement",
)
self.assertEqual(exp.status, ExperimentStatus.COMPLETED)
def test_complete_rollout_without_winner_raises(self) -> None:
exp = self._approve_and_start()
with self.assertRaises(ValidationError) as ctx:
experiment_complete(
experiment=exp,
user=self.experimenter,
outcome=OutcomeType.ROLLOUT,
rationale="Treatment won",
)
self.assertIn("winning_variant_id", ctx.exception.message_dict)
def test_complete_empty_rationale_raises(self) -> None:
exp = self._approve_and_start()
with self.assertRaises(ValidationError) as ctx:
experiment_complete(
experiment=exp,
user=self.experimenter,
outcome=OutcomeType.ROLLBACK,
rationale=" ",
)
self.assertIn("rationale", ctx.exception.message_dict)
def test_archive_after_complete(self) -> None:
exp = self._approve_and_start()
exp = experiment_complete(
experiment=exp,
user=self.experimenter,
outcome=OutcomeType.NO_EFFECT,
rationale="No effect observed",
)
exp = experiment_archive(experiment=exp, user=self.experimenter)
self.assertEqual(exp.status, ExperimentStatus.ARCHIVED)
def test_start_with_competing_active_raises(self) -> None:
exp1 = self._approve_and_start()
self.assertEqual(exp1.status, ExperimentStatus.RUNNING)
exp2 = experiment_create(
flag=self.flag,
name="LF Exp 2",
owner=self.experimenter,
)
add_two_variants(exp2)
exp2 = experiment_submit_for_review(
experiment=exp2, user=self.experimenter
)
exp2 = experiment_approve(experiment=exp2, approver=self.approver)
with self.assertRaises(ValidationError) as ctx:
experiment_start(experiment=exp2, user=self.experimenter)
self.assertIn("flag", ctx.exception.message_dict)
def test_start_from_draft_raises(self) -> None:
with self.assertRaises(ValidationError):
experiment_start(experiment=self.exp, user=self.experimenter)
def test_audit_log_tracks_all_transitions(self) -> None:
exp = self._approve_and_start()
exp = experiment_complete(
experiment=exp,
user=self.experimenter,
outcome=OutcomeType.ROLLBACK,
rationale="done",
)
logs = ExperimentLog.objects.filter(experiment=exp).order_by(
"created_at"
)
log_types = list(logs.values_list("log_type", flat=True))
self.assertIn(LogType.STATUS_CHANGE, log_types)
self.assertIn(LogType.REVIEW_REQUESTED, log_types)
self.assertIn(LogType.APPROVAL, log_types)
self.assertIn(LogType.COMPLETED, log_types)
class OwnershipPermissionTest(TestCase):
@override
def setUp(self) -> None:
self.owner = make_experimenter("_own")
self.other = make_experimenter("_oth")
self.admin = make_admin("_adm")
self.flag = make_flag(suffix="_op")
self.exp = experiment_create(
flag=self.flag,
name="Ownership Exp",
owner=self.owner,
)
add_two_variants(self.exp)
def test_owner_can_update(self) -> None:
exp = experiment_update(
experiment=self.exp, user=self.owner, name="Renamed"
)
self.assertEqual(exp.name, "Renamed")
def test_admin_can_update(self) -> None:
exp = experiment_update(
experiment=self.exp, user=self.admin, name="Admin Renamed"
)
self.assertEqual(exp.name, "Admin Renamed")
def test_other_experimenter_cannot_update(self) -> None:
with self.assertRaises(ForbiddenError):
experiment_update(
experiment=self.exp, user=self.other, name="Hijack"
)
def test_other_experimenter_cannot_create_variant(self) -> None:
with self.assertRaises(ForbiddenError):
variant_create(
experiment=self.exp,
user=self.other,
name="v3",
value="c",
weight=Decimal("10.00"),
)
def test_other_experimenter_cannot_update_variant(self) -> None:
_vc, vt = self.exp.variants.all()[0], self.exp.variants.all()[1]
with self.assertRaises(ForbiddenError):
variant_update(variant=vt, user=self.other, name="hijack")
def test_other_experimenter_cannot_delete_variant(self) -> None:
vt = self.exp.variants.filter(is_control=False).first()
with self.assertRaises(ForbiddenError):
variant_delete(variant=vt, user=self.other)
def test_other_experimenter_cannot_submit_for_review(self) -> None:
with self.assertRaises(ForbiddenError):
experiment_submit_for_review(
experiment=self.exp, user=self.other
)
def test_admin_can_submit_for_review(self) -> None:
exp = experiment_submit_for_review(
experiment=self.exp, user=self.admin
)
self.assertEqual(exp.status, ExperimentStatus.IN_REVIEW)
def test_other_experimenter_cannot_start(self) -> None:
review_settings_update(
default_min_approvals=1, allow_any_approver=True
)
approver = make_approver("_ops")
exp = experiment_submit_for_review(
experiment=self.exp, user=self.owner
)
exp = experiment_approve(
experiment=exp, approver=approver, comment="ok"
)
with self.assertRaises(ForbiddenError):
experiment_start(experiment=exp, user=self.other)
def test_other_experimenter_cannot_pause(self) -> None:
review_settings_update(
default_min_approvals=1, allow_any_approver=True
)
approver = make_approver("_opp")
exp = experiment_submit_for_review(
experiment=self.exp, user=self.owner
)
exp = experiment_approve(
experiment=exp, approver=approver, comment="ok"
)
exp = experiment_start(experiment=exp, user=self.owner)
with self.assertRaises(ForbiddenError):
experiment_pause(experiment=exp, user=self.other)
def test_admin_can_pause(self) -> None:
review_settings_update(
default_min_approvals=1, allow_any_approver=True
)
approver = make_approver("_opa")
exp = experiment_submit_for_review(
experiment=self.exp, user=self.owner
)
exp = experiment_approve(
experiment=exp, approver=approver, comment="ok"
)
exp = experiment_start(experiment=exp, user=self.owner)
exp = experiment_pause(experiment=exp, user=self.admin)
self.assertEqual(exp.status, ExperimentStatus.PAUSED)