electron-rare/conversations
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

✨(backend) add feature flags from posthog

Browse Source 
This adds posthog backend to know whether a feature is enabled
for a user.
More globally, if Posthog is not present feature flags can be
enabled/disabled globally.
This commit is contained in:
Quentin BEY
2025-08-27 23:29:32 +02:00
parent fd06d33d05
commit b6fe71adb0
21 changed files with 659 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+1
View File
@@ -23,6 +23,7 @@ and this project adheres to
- ✨(chat) add document RAG on document uploaded by user #8
- ✨(backend) allow use to stop conversation streaming #14
- 🐛(agent) add the current date in the system prompt #18
- ✨(backend) add feature flags from posthog #13
[unreleased]: https://github.com/numerique-gouv/conversations/compare/HEAD...main
env.d/development/common.e2e.dist
+4
View File
@@ -2,3 +2,7 @@
BURST_THROTTLE_RATES="200/minute"
DJANGO_SERVER_TO_SERVER_API_TOKENS=test-e2e
SUSTAINED_THROTTLE_RATES="200/hour"
# Features
FEATURE_FLAG_WEB_SEARCH=ENABLED
FEATURE_FLAG_DOCUMENT_UPLOAD=ENABLED
src/backend/chat/clients/pydantic_ai.py
+41 -1
View File
@@ -47,6 +47,8 @@ from pydantic_ai.messages import (
from pydantic_ai.models.openai import OpenAIModel
from pydantic_ai.providers.openai import OpenAIProvider
from core.feature_flags.helpers import is_feature_enabled
from chat.agent_rag.document_search.albert_api import AlbertRagDocumentSearch
from chat.ai_sdk_types import (
LanguageModelV1Source,
@@ -107,8 +109,16 @@ class UserIntent(BaseModel):
class AIAgentService:
"""Service class for AI-related operations (Pydantic-AI edition)."""
def __init__(self, conversation):
def __init__(self, conversation, user):
"""
Initialize the AI agent service.
Args:
conversation: The chat conversation instance
user: The authenticated user instance, only used for dynamic feature flags
"""
self.conversation = conversation
self.user = user # authenticated user only
self._last_stop_check = 0
@property
@@ -195,6 +205,14 @@ class AIAgentService:
Raises:
ImproperlyConfigured: If the AI configuration is not set.
"""
if not any(
is_feature_enabled(self.user, feature) for feature in ["web_search", "document_upload"]
):
logger.info(
"No web search or document upload features enabled, skipping intent detection.",
)
return UserIntent()
if missing_settings := [
setting
for setting in (
@@ -229,6 +247,16 @@ class AIAgentService:
result.output.web_search = False
logger.info("Web search backend is disabled, skipping intent detection.")
if not is_feature_enabled(self.user, "document_upload"):
# If document upload is not enabled, we can skip the attachment summary intent
result.output.attachment_summary = False
logger.info("Document upload feature is disabled, skipping attachment summary intent.")
if not is_feature_enabled(self.user, "web_search"):
# If web search is not enabled, we can skip the web search intent
result.output.web_search = False
logger.info("Web search feature is disabled, skipping web search intent.")
return result.output
def parse_input_documents(self, documents: List[BinaryContent]):
@@ -375,6 +403,18 @@ class AIAgentService:
usage = {"promptTokens": 0, "completionTokens": 0}
# Feature flag management
if force_web_search and not is_feature_enabled(self.user, "web_search"):
logger.warning("Web search feature is disabled, ignoring force_web_search.")
force_web_search = False
if any([input_images, input_documents]) and not is_feature_enabled(
self.user, "document_upload"
):
logger.warning("Document upload feature is disabled, ignoring input documents.")
input_images = []
input_documents = []
# Detect the user intent
if not force_web_search:
user_intent: UserIntent = await self._detect_user_intent(user_prompt)
src/backend/chat/tests/clients/pydantic_ai/test_stream_methods.py
+4 -4
View File
@@ -32,7 +32,7 @@ def ui_messages_fixture():
def test_stream_text_delegates_to_async(mock_convert, ui_messages):
"""Test stream_text method delegates to async version."""
conversation = ChatConversationFactory()
service = AIAgentService(conversation)
service = AIAgentService(conversation, user=conversation.owner)
mock_convert.return_value = iter(["Hello", " world"])
result = service.stream_text(ui_messages, force_web_search=True)
@@ -45,7 +45,7 @@ def test_stream_text_delegates_to_async(mock_convert, ui_messages):
def test_stream_data_delegates_to_async(mock_convert, ui_messages):
"""Test stream_data method delegates to async version."""
conversation = ChatConversationFactory()
service = AIAgentService(conversation)
service = AIAgentService(conversation, user=conversation.owner)
mock_convert.return_value = iter(['0:"Hello"\n', 'd:{"finishReason":"stop"}\n'])
result = service.stream_data(ui_messages, force_web_search=False)
@@ -58,7 +58,7 @@ def test_stream_data_delegates_to_async(mock_convert, ui_messages):
async def test_stream_text_async_filters_text_deltas(ui_messages):
"""Test stream_text_async only yields text deltas."""
conversation = await sync_to_async(ChatConversationFactory)()
service = AIAgentService(conversation)
service = AIAgentService(conversation, user=conversation.owner)
# Mock _run_agent to return various delta types
async def mock_run_agent(*args, **kwargs):
@@ -79,7 +79,7 @@ async def test_stream_text_async_filters_text_deltas(ui_messages):
async def test_stream_data_async_formats_as_sdk_events(ui_messages):
"""Test stream_data_async formats events correctly."""
conversation = await sync_to_async(ChatConversationFactory)()
service = AIAgentService(conversation)
service = AIAgentService(conversation, user=conversation.owner)
async def mock_run_agent(*args, **kwargs):
yield {"type": "0", "payload": "Hello"}
src/backend/chat/tests/views/chat/conversations/test_conversation.py
+55
View File
@@ -2,6 +2,7 @@
# pylint: disable=too-many-lines
import json
import logging
from django.utils import timezone
@@ -11,6 +12,7 @@ from freezegun import freeze_time
from rest_framework import status
from core.factories import UserFactory
from core.feature_flags.flags import FeatureToggle
from chat.ai_sdk_types import (
Attachment,
@@ -889,3 +891,56 @@ def test_post_conversation_tool_call_fails(
"vendor_id": None,
},
]
@freeze_time("2025-07-25T10:36:35.297675Z")
@respx.mock
def test_post_conversation_data_protocol_feature_disabled(
api_client,
caplog,
mock_openai_stream,
feature_flags,
):
"""Test posting messages to a conversation using the 'data' protocol."""
feature_flags.web_search = FeatureToggle.DISABLED
feature_flags.document_upload = FeatureToggle.DISABLED
caplog.set_level(logging.INFO)
chat_conversation = ChatConversationFactory()
url = f"/api/v1.0/chats/{chat_conversation.pk}/conversation/?protocol=data"
data = {
"messages": [
{
"id": "yuPoOuBkKA4FnKvk",
"role": "user",
"parts": [{"text": "Hello", "type": "text"}],
"content": "Hello",
"createdAt": "2025-07-03T15:22:17.105Z",
}
]
}
api_client.force_login(chat_conversation.owner)
response = api_client.post(url, data, format="json")
assert response.status_code == status.HTTP_200_OK
assert response.get("Content-Type") == "text/event-stream"
assert response.get("x-vercel-ai-data-stream") == "v1"
assert response.streaming
# Wait for the streaming content to be fully received
response_content = b"".join(response.streaming_content).decode("utf-8")
assert response_content == (
'0:"Hello"\n'
'0:" there"\n'
'd:{"finishReason": "stop", "usage": {"promptTokens": 0, "completionTokens": 0}}\n'
)
assert mock_openai_stream.called
assert (
"No web search or document upload features enabled, skipping intent detection."
in caplog.text
)
assert "User intent detected: {'web_search': False, 'attachment_summary': False}" in caplog.text
src/backend/chat/tests/views/chat/conversations/test_conversation_with_document_search_rag.py
+69
View File
@@ -3,6 +3,7 @@
# pylint: disable=too-many-lines
import base64
import json
import logging
from io import BytesIO
from django.utils import timezone
@@ -14,6 +15,8 @@ import respx
from freezegun import freeze_time
from rest_framework import status
from core.feature_flags.flags import FeatureToggle
from chat.ai_sdk_types import (
Attachment,
LanguageModelV1Source,
@@ -461,3 +464,69 @@ def test_post_conversation_with_document_upload( # noqa:PLR0913 # pylint: disa
"vendor_details": None,
"vendor_id": None,
}
@responses.activate
@respx.mock
@freeze_time("2025-07-25T10:36:35.297675Z")
def test_post_conversation_with_document_upload_feature_disabled( # noqa:PLR0913 # pylint: disable=too-many-arguments,too-many-positional-arguments
api_client,
caplog,
mock_albert_api, # pylint: disable=unused-argument
mock_openai_stream, # pylint: disable=unused-argument
mock_intent_detection_document, # pylint: disable=unused-argument
sample_pdf_content,
feature_flags,
):
"""
Test POST to /api/v1/chats/{pk}/conversation/ with a PDF document while feature is disabled.
"""
feature_flags.web_search = FeatureToggle.DISABLED
feature_flags.document_upload = FeatureToggle.DISABLED
caplog.set_level(logging.WARNING)
chat_conversation = ChatConversationFactory()
api_client.force_authenticate(user=chat_conversation.owner)
pdf_base64 = base64.b64encode(sample_pdf_content.read()).decode("utf-8")
message = UIMessage(
id="1",
role="user",
content="What does the document say?",
parts=[
TextUIPart(
text="What does the document say?",
type="text",
),
],
experimental_attachments=[
Attachment(
name="sample.pdf",
contentType="application/pdf",
url=f"data:application/pdf;base64,{pdf_base64}",
)
],
)
response = api_client.post(
f"/api/v1.0/chats/{chat_conversation.pk}/conversation/",
data={"messages": [message.model_dump(mode="json")]},
format="json",
)
assert response.status_code == status.HTTP_200_OK
assert response.get("Content-Type") == "text/event-stream"
assert response.get("x-vercel-ai-data-stream") == "v1"
assert response.streaming
# Wait for the streaming content to be fully received
response_content = b"".join(response.streaming_content).decode("utf-8")
assert response_content == (
'0:"From the document, I can see that "\n'
"0:\"it says 'Hello PDF'.\"\n"
'd:{"finishReason": "stop", "usage": {"promptTokens": 150, '
'"completionTokens": 25}}\n'
)
# This behavior must be improved in the future to inform the user properly
assert "Document upload feature is disabled, ignoring input documents." in caplog.text
src/backend/chat/tests/views/chat/conversations/test_conversation_with_web_search_rag.py
+54
View File
@@ -2,6 +2,7 @@
# pylint: disable=too-many-lines
import json
import logging
from django.utils import timezone
@@ -11,6 +12,8 @@ import respx
from freezegun import freeze_time
from rest_framework import status
from core.feature_flags.flags import FeatureToggle
from chat.ai_sdk_types import (
LanguageModelV1Source,
SourceUIPart,
@@ -721,3 +724,54 @@ def test_conversation_with_web_search_text_protocol(api_client, mock_openai_stre
part for part in assistant_message.parts if hasattr(part, "type") and part.type == "source"
]
assert len(source_parts) > 0, "Expected source parts in assistant message"
@freeze_time("2025-07-25T10:36:35.297675Z")
@respx.mock
def test_post_conversation_data_protocol_feature_disabled_force_web(
api_client,
caplog,
mock_openai_stream,
feature_flags,
):
"""Test posting messages to a conversation using the 'data' protocol."""
feature_flags.web_search = FeatureToggle.DISABLED
feature_flags.document_upload = FeatureToggle.DISABLED
caplog.set_level(logging.WARNING)
chat_conversation = ChatConversationFactory()
url = (
f"/api/v1.0/chats/{chat_conversation.pk}/conversation/?protocol=data&force_web_search=true"
)
data = {
"messages": [
{
"id": "yuPoOuBkKA4FnKvk",
"role": "user",
"parts": [{"text": "Hello", "type": "text"}],
"content": "Hello",
"createdAt": "2025-07-03T15:22:17.105Z",
}
]
}
api_client.force_login(chat_conversation.owner)
response = api_client.post(url, data, format="json")
assert response.status_code == status.HTTP_200_OK
assert response.get("Content-Type") == "text/event-stream"
assert response.get("x-vercel-ai-data-stream") == "v1"
assert response.streaming
# Wait for the streaming content to be fully received
response_content = b"".join(response.streaming_content).decode("utf-8")
assert response_content == (
'0:"Hello"\n'
'0:" there"\n'
'd:{"finishReason": "stop", "usage": {"promptTokens": 0, "completionTokens": 0}}\n'
)
assert mock_openai_stream.called
assert "Web search feature is disabled, ignoring force_web_search." in caplog.text
src/backend/chat/views.py
+2 -2
View File
@@ -112,7 +112,7 @@ class ChatViewSet( # pylint: disable=too-many-ancestors
if not messages:
return Response({"error": "No messages provided"}, status=status.HTTP_400_BAD_REQUEST)
ai_service = AIAgentService(conversation=conversation)
ai_service = AIAgentService(conversation=conversation, user=self.request.user)
if protocol == "data":
streaming_content = ai_service.stream_data(messages, force_web_search=force_web_search)
else: # Default to 'text' protocol
@@ -159,6 +159,6 @@ class ChatViewSet( # pylint: disable=too-many-ancestors
"""
conversation = self.get_object()
AIAgentService(conversation=conversation).stop_streaming()
AIAgentService(conversation=conversation, user=self.request.user).stop_streaming()
return Response({"status": "OK"}, status=status.HTTP_200_OK)
src/backend/conftest.py
+10
View File
@@ -24,3 +24,13 @@ def no_http_requests(monkeypatch):
raise RuntimeError(f"The test was about to {method} {self.scheme}://{self.host}{url}")
monkeypatch.setattr("urllib3.connectionpool.HTTPConnectionPool.urlopen", urlopen_mock)
@pytest.fixture(name="feature_flags", scope="function")
def feature_flags_fixture(settings):
"""
Ease feature flags setting in tests by working on a copy
to allow proper restore by SettingsWrapper after the test.
"""
settings.FEATURE_FLAGS = settings.FEATURE_FLAGS.model_copy(deep=True)
yield settings.FEATURE_FLAGS
src/backend/conversations/settings.py
+70
View File
@@ -9,16 +9,20 @@ https://docs.djangoproject.com/en/3.1/topics/settings/
For the full list of settings and their values, see
https://docs.djangoproject.com/en/3.1/ref/settings/
"""
# pylint: disable=too-many-lines
import os
import tomllib
from socket import gethostbyname, gethostname
import posthog
import sentry_sdk
from configurations import Configuration, values
from sentry_sdk.integrations.django import DjangoIntegration
from sentry_sdk.integrations.logging import ignore_logger
from core.feature_flags.flags import FeatureFlags, FeatureToggle
# Build paths inside the project like this: BASE_DIR / 'subdir'.
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DATA_DIR = os.getenv("DATA_DIR", os.path.join("/", "data"))
@@ -175,6 +179,7 @@ class Base(Configuration):
"django.middleware.common.CommonMiddleware",
"django.middleware.csrf.CsrfViewMiddleware",
"django.contrib.auth.middleware.AuthenticationMiddleware",
"core.middleware.PostHogMiddleware",
"django.contrib.messages.middleware.MessageMiddleware",
"dockerflow.django.middleware.DockerflowMiddleware",
]
@@ -310,6 +315,7 @@ class Base(Configuration):
)
# Posthog
# Looks like "{'id': 'posthog_key', 'host': 'https://product.conversations.127.0.0.1.nip.io'}"
POSTHOG_KEY = values.DictValue(None, environ_name="POSTHOG_KEY", environ_prefix=None)
# Crisp
@@ -733,6 +739,39 @@ USER QUESTION:
},
}
@property
def FEATURE_FLAGS(self) -> FeatureFlags: # pylint: disable=invalid-name
"""
Return a dictionary of feature flags for the application.
This can be used to enable or disable features dynamically.
"""
features = FeatureFlags(
**{
field_name: FeatureToggle[
values.Value(
field.default.name,
environ_name=f"FEATURE_FLAG_{field_name.upper()}",
environ_prefix=None,
)
]
for field_name, field in FeatureFlags.model_fields.items()
}
)
# Sanity check to ensure that the RAG_WEB_SEARCH_BACKEND and RAG_DOCUMENT_SEARCH_BACKEND
if features.web_search and not self.RAG_WEB_SEARCH_BACKEND:
raise RuntimeError(
"RAG_WEB_SEARCH_BACKEND is not set, but web_search feature flag is enabled."
)
if features.document_upload and not self.RAG_DOCUMENT_SEARCH_BACKEND:
raise RuntimeError(
"RAG_DOCUMENT_SEARCH_BACKEND is not set, "
"but document_upload feature flag is enabled."
)
return features
@classmethod
def post_setup(cls):
"""Post setup configuration.
@@ -754,6 +793,11 @@ USER QUESTION:
# Ignore the logs added by the DockerflowMiddleware
ignore_logger("request.summary")
# Enable Posthog if the key is set
if cls.POSTHOG_KEY:
posthog.api_key = cls.POSTHOG_KEY["id"]
posthog.host = cls.POSTHOG_KEY["host"]
if cls.OIDC_FALLBACK_TO_EMAIL_FOR_IDENTIFICATION and cls.OIDC_ALLOW_DUPLICATE_EMAILS:
raise ValueError(
"Both OIDC_FALLBACK_TO_EMAIL_FOR_IDENTIFICATION and "
@@ -831,10 +875,36 @@ class Test(Base):
AI_ROUTING_MODEL = None
AI_ROUTING_MODEL_API_KEY = None
POSTHOG_KEY = None
def __init__(self):
# pylint: disable=invalid-name
self.INSTALLED_APPS += ["drf_spectacular_sidecar"]
@property
def FEATURE_FLAGS(self) -> FeatureFlags: # pylint: disable=invalid-name
"""
In the test environment, we want to enable all features to test them.
We willingly call the super to validate it works (coverage), but we override all
the feature flags to always enabled.
"""
_feature_flags = super().FEATURE_FLAGS
for field_name in FeatureFlags.model_fields.keys():
setattr(_feature_flags, field_name, FeatureToggle.ENABLED)
return _feature_flags
@classmethod
def post_setup(cls):
"""Post setup configuration."""
super().post_setup()
# Force logger propagation to allow caplog to work
# see https://github.com/pytest-dev/pytest/issues/3697
cls.LOGGING["loggers"]["core"]["propagate"] = True
class ContinuousIntegration(Test):
"""
src/backend/core/api/viewsets.py
+6 -1
View File
@@ -10,6 +10,7 @@ from django.db.models.expressions import RawSQL
from django.utils.text import slugify
import rest_framework as drf
from pydantic import BaseModel
from rest_framework import viewsets
from rest_framework.permissions import AllowAny
from rest_framework.throttling import UserRateThrottle
@@ -208,10 +209,14 @@ class ConfigView(drf.views.APIView):
"LANGUAGES",
"LANGUAGE_CODE",
"SENTRY_DSN",
"FEATURE_FLAGS",
]
dict_settings = {}
for setting in array_settings:
if hasattr(settings, setting):
_setting_value = getattr(settings, setting)
if isinstance(_setting_value, BaseModel):
dict_settings[setting] = _setting_value.model_dump(by_alias=True)
else:
dict_settings[setting] = getattr(settings, setting)
dict_settings["theme_customization"] = self._load_theme_customization()
src/backend/core/feature_flags/__init__.py
View File
src/backend/core/feature_flags/flags.py
+45
View File
@@ -0,0 +1,45 @@
"""Feature flag objects"""
from enum import StrEnum
from django.utils.text import slugify
from pydantic import BaseModel, ConfigDict
class FeatureToggle(StrEnum):
"""
Feature toggle states.
DISABLED: The feature is fully disabled (ie for all users).
DYNAMIC: The feature can be enabled or disabled based on service like Posthog.
ENABLED: The feature is fully enabled (ie for all users).
"""
DISABLED = "disabled"
DYNAMIC = "dynamic"
ENABLED = "enabled"
@property
def is_always_enabled(self) -> bool:
"""Whether the feature is always enabled"""
return self.value == self.ENABLED
@property
def is_always_disabled(self) -> bool:
"""Whether the feature is always disabled"""
return self.value == self.DISABLED
class FeatureFlags(BaseModel):
"""Feature flags container."""
model_config = ConfigDict(
extra="forbid",
alias_generator=lambda x: slugify(x).replace("_", "-"),
populate_by_name=True,
)
# features
web_search: FeatureToggle = FeatureToggle.DISABLED
document_upload: FeatureToggle = FeatureToggle.DISABLED
src/backend/core/feature_flags/helpers.py
+46
View File
@@ -0,0 +1,46 @@
"""Tooling around feature flags"""
import logging
from django.conf import settings
from django.contrib.auth import get_user_model
User = get_user_model()
try:
import posthog
except ImportError:
posthog = None
logger = logging.getLogger(__name__)
def frontend_feature_name(feature_name: str) -> str:
"""
Formats the feature flag name to match the expected format in settings.
This is the reverse of format_feature_flag_name_back.
"""
return feature_name.lower().replace("_", "-")
def is_feature_enabled(
user: User,
feature_name: str,
) -> bool:
"""Whether a feature is enabled or not."""
_settings_value = getattr(settings.FEATURE_FLAGS, feature_name) # might raise on purpose
if _settings_value.is_always_enabled:
return True
if _settings_value.is_always_disabled:
return False
# Then it's dynamic
if posthog is not None:
return posthog.feature_enabled(frontend_feature_name(feature_name), user.email)
# No feature flag manager
logger.warning(
"No feature flag manager found, cannot use dynamic for %s -> disabled",
feature_name,
)
return False
src/backend/core/middleware.py
+58
View File
@@ -0,0 +1,58 @@
"""Custom middleware(s) for the project."""
import json
import logging
from urllib.parse import unquote
from django.conf import settings
from django.core.exceptions import MiddlewareNotUsed
# We are importing posthog here, but it will only be used if the POSTHOG_KEY is set in settings.
# The settings are checked before any attempt to use posthog.
try:
import posthog
except ImportError:
posthog = None
logger = logging.getLogger(__name__)
class PostHogMiddleware:
"""
This middleware is used to alias the user's distinct_id from the PostHog cookie
with their email address when they are authenticated. This allows us to track
users across different sessions and devices.
"""
def __init__(self, get_response):
"""
Initialize the middleware and disable it if PostHog is not configured.
"""
if posthog is None or not settings.POSTHOG_KEY:
raise MiddlewareNotUsed("POSTHOG_KEY must be set in settings to use PostHogMiddleware.")
self.get_response = get_response
def __call__(self, request):
"""
Process the request to handle the PostHog alias.
"""
if posthog is not None and settings.POSTHOG_KEY:
posthog_cookie = request.COOKIES.get(f"ph_{posthog.project_api_key}_posthog")
if posthog_cookie:
try:
cookie_dict = json.loads(unquote(posthog_cookie))
if (
cookie_dict.get("distinct_id")
and request.user
and request.user.is_authenticated
):
posthog.alias(cookie_dict["distinct_id"], request.user.email)
except (json.JSONDecodeError, KeyError):
# If the cookie is malformed or doesn't contain the expected
# keys, we can't do anything with it, so we ignore it.
logger.warning("Malformed PostHog cookie: %s", posthog_cookie)
response = self.get_response(request)
return response
src/backend/core/tests/feature_flags/__init__.py
View File
src/backend/core/tests/feature_flags/test_flags.py
+96
View File
@@ -0,0 +1,96 @@
"""Tests for feature flag models and enums."""
from typing import get_type_hints
import pytest
from pydantic import ValidationError
from core.feature_flags.flags import FeatureFlags, FeatureToggle
def test_is_always_enabled():
"""Test the is_always_enabled property."""
assert FeatureToggle.ENABLED.is_always_enabled is True
assert FeatureToggle.DYNAMIC.is_always_enabled is False
assert FeatureToggle.DISABLED.is_always_enabled is False
def test_is_always_disabled():
"""Test the is_always_disabled property."""
assert FeatureToggle.DISABLED.is_always_disabled is True
assert FeatureToggle.DYNAMIC.is_always_disabled is False
assert FeatureToggle.ENABLED.is_always_disabled is False
def test_defaults():
"""Ensure new instance has the declared defaults."""
flags = FeatureFlags()
assert flags.web_search is FeatureToggle.DISABLED
assert flags.document_upload is FeatureToggle.DISABLED
@pytest.mark.parametrize(
"field,value",
[
("web_search", FeatureToggle.ENABLED),
("web_search", "enabled"),
],
)
def test_assign_valid_values(field, value):
"""Assignment via attribute or alias must accept valid values."""
flags = FeatureFlags(**{field: value})
assert flags.web_search == FeatureToggle(value)
@pytest.mark.parametrize(
"field,value",
[
("web_search", "not-a-state"),
("document_upload", 123),
("extra_field", "anything"), # extra="forbid"
],
)
def test_reject_invalid_values(field, value):
"""Bad values or extra keys must raise ValidationError."""
data = {field: value}
with pytest.raises(ValidationError):
FeatureFlags(**data)
def test_populate_by_name_and_alias():
"""Both snake_case and kebab-case aliases work."""
from_snake = FeatureFlags(web_search="enabled")
from_kebab = FeatureFlags(**{"web-search": "disabled"})
assert from_snake.web_search is FeatureToggle.ENABLED
assert from_kebab.web_search is FeatureToggle.DISABLED
def test_model_config_forbid_extra():
"""Extra keys are rejected."""
with pytest.raises(ValidationError):
FeatureFlags(unknown_flag="enabled")
def test_round_trip_serialization():
"""JSON round-trip keeps values intact."""
original = FeatureFlags(
web_search=FeatureToggle.DYNAMIC,
document_upload=FeatureToggle.ENABLED,
)
raw = original.model_dump_json()
restored = FeatureFlags.model_validate_json(raw)
assert restored == original
assert raw == ('{"web_search":"dynamic","document_upload":"enabled"}')
raw_alias = original.model_dump_json(by_alias=True)
restored_alias = FeatureFlags.model_validate_json(raw_alias)
assert restored_alias == original
assert raw_alias == ('{"web-search":"dynamic","document-upload":"enabled"}')
def test_all_fields_are_feature_toggle():
"""Static guarantee that every declared flag is a FeatureToggle."""
hints = get_type_hints(FeatureFlags)
for name, typ in hints.items():
assert typ is FeatureToggle, f"{name} is not FeatureToggle"
src/backend/core/tests/feature_flags/test_helpers.py
+92
View File
@@ -0,0 +1,92 @@
"""Tests for feature flag helpers."""
import logging
from unittest.mock import patch
import pytest
from core.factories import UserFactory
from core.feature_flags.flags import FeatureToggle
from core.feature_flags.helpers import frontend_feature_name, is_feature_enabled
pytestmark = pytest.mark.django_db()
@pytest.mark.parametrize(
("raw", "expected"),
[
("web_search", "web-search"),
("document_upload", "document-upload"),
("SNAKE_CASE", "snake-case"),
("already-dashed", "already-dashed"),
],
)
def test_frontend_feature_name(raw: str, expected: str):
"""Test the feature name formatting function."""
assert frontend_feature_name(raw) == expected
def test_is_feature_enabled_always_enabled(feature_flags):
"""Test that a feature that is always enabled returns True."""
feature_flags.web_search = FeatureToggle.ENABLED
feature_flags.document_upload = FeatureToggle.DISABLED
user = UserFactory()
assert is_feature_enabled(user, "web_search") is True
def test_is_feature_enabled_always_disabled(feature_flags):
"""Test that a feature that is always disabled returns False."""
feature_flags.web_search = FeatureToggle.ENABLED
feature_flags.document_upload = FeatureToggle.DISABLED
user = UserFactory()
assert is_feature_enabled(user, "document_upload") is False
@patch("core.feature_flags.helpers.posthog")
def test_is_feature_enabled_dynamic_posthog_true(mock_posthog, feature_flags):
"""Test that a dynamic feature returns the value from PostHog when PostHog is available."""
feature_flags.web_search = FeatureToggle.DYNAMIC
user = UserFactory()
mock_posthog.feature_enabled.return_value = True
assert is_feature_enabled(user, "web_search") is True
mock_posthog.feature_enabled.assert_called_once_with(
"web-search",
user.email,
)
@patch("core.feature_flags.helpers.posthog")
def test_is_feature_enabled_dynamic_posthog_false(mock_posthog, feature_flags):
"""Test that a dynamic feature returns the value from PostHog when PostHog is available."""
feature_flags.web_search = FeatureToggle.DYNAMIC
user = UserFactory()
mock_posthog.feature_enabled.return_value = False
assert is_feature_enabled(user, "web_search") is False
@patch("core.feature_flags.helpers.posthog", None)
def test_is_feature_enabled_dynamic_no_posthog(caplog, feature_flags):
"""Test that a dynamic feature falls back to settings when PostHog is not available."""
caplog.set_level(logging.WARNING, logger="core")
feature_flags.web_search = FeatureToggle.DYNAMIC
user = UserFactory()
assert is_feature_enabled(user, "web_search") is False
assert len(caplog.records) == 1
record = caplog.records[0]
assert record.levelno == logging.WARNING
assert record.message == (
"No feature flag manager found, cannot use dynamic for web_search -> disabled"
)
def test_is_feature_enabled_missing_flag_raises_attribute_error():
"""Test that requesting an unknown feature flag raises an AttributeError."""
user = UserFactory()
with pytest.raises(AttributeError):
is_feature_enabled(user, "unknown_feature")
src/backend/core/tests/test_api_config.py
+1
View File
@@ -44,6 +44,7 @@ def test_api_config(is_authenticated):
assert response.json() == {
"CRISP_WEBSITE_ID": "123",
"ENVIRONMENT": "test",
"FEATURE_FLAGS": {"document-upload": "enabled", "web-search": "enabled"},
"FRONTEND_CSS_URL": "http://testcss/",
"FRONTEND_HOMEPAGE_FEATURE_ENABLED": True,
"FRONTEND_THEME": "test-theme",
src/backend/pyproject.toml
+1
View File
@@ -54,6 +54,7 @@ dependencies = [
"mozilla-django-oidc==4.0.1",
"nested-multipart-parser==1.5.0",
"openai==1.99.1", # see https://github.com/pydantic/pydantic-ai/issues/2476
"posthog==6.4.1",
"pydantic==2.11.7",
"pydantic-ai==0.6.2",
"psycopg[binary]==3.2.9",
src/frontend/apps/e2e/__tests__/app-conversations/common.ts
+4
View File
@@ -3,6 +3,10 @@ import { Page, expect } from '@playwright/test';
export const CONFIG = {
CRISP_WEBSITE_ID: null,
ENVIRONMENT: 'development',
FEATURE_FLAGS: {
'document-upload': 'enabled',
'web-search': 'enabled',
},
FRONTEND_CSS_URL: null,
FRONTEND_HOMEPAGE_FEATURE_ENABLED: true,
FRONTEND_THEME: null,