75 lines
3.4 KiB
Python
75 lines
3.4 KiB
Python
"""Account-bound opt-in for recommended questions after answers."""
|
|
|
|
from types import SimpleNamespace
|
|
|
|
import httpx
|
|
import pytest
|
|
import pytest_asyncio
|
|
from fastapi import FastAPI
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from app.gateway.routers import user_preferences
|
|
from deerflow.persistence.user_preferences.model import UserPreferencesRow
|
|
from deerflow.persistence.user_preferences.sql import UserPreferencesRepository
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def preferences(tmp_path):
|
|
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'preferences.db'}")
|
|
async with engine.begin() as connection:
|
|
await connection.run_sync(UserPreferencesRow.__table__.create)
|
|
store = UserPreferencesRepository(async_sessionmaker(engine, expire_on_commit=False))
|
|
app = FastAPI()
|
|
app.state.user_preferences_store = store
|
|
app.include_router(user_preferences.router)
|
|
|
|
@app.middleware("http")
|
|
async def current_user(request, call_next):
|
|
request.state.user = SimpleNamespace(id=request.headers.get("X-Test-User", "alice"))
|
|
return await call_next(request)
|
|
|
|
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
|
|
yield client, store
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_followup_suggestions_default_off_for_new_and_existing_accounts(preferences):
|
|
client, store = preferences
|
|
response = await client.get("/api/user-preferences")
|
|
assert response.status_code == 200
|
|
assert response.json()["followup_suggestions_enabled"] is False
|
|
|
|
await store.merge_for_user("alice", {"hide_admin_prompt_prefix": True})
|
|
response = await client.get("/api/user-preferences")
|
|
assert response.json() == {"hide_admin_prompt_prefix": True, "followup_suggestions_enabled": False}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_followup_suggestions_opt_in_is_persisted_only_for_current_account(preferences):
|
|
client, store = preferences
|
|
await store.merge_for_user("alice", {"hide_admin_prompt_prefix": True})
|
|
response = await client.put("/api/user-preferences", json={"followup_suggestions_enabled": True})
|
|
assert response.status_code == 200
|
|
assert response.json() == {"hide_admin_prompt_prefix": True, "followup_suggestions_enabled": True}
|
|
assert (await store.get_for_user("alice"))["followup_suggestions_enabled"] is True
|
|
|
|
response = await client.get("/api/user-preferences", headers={"X-Test-User": "bob"})
|
|
assert response.json()["followup_suggestions_enabled"] is False
|
|
assert await store.get_for_user("bob") == {}
|
|
|
|
response = await client.put("/api/user-preferences", json={"followup_suggestions_enabled": False})
|
|
assert response.json() == {"hide_admin_prompt_prefix": True, "followup_suggestions_enabled": False}
|
|
assert (await client.get("/api/user-preferences")).json()["followup_suggestions_enabled"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unrelated_or_empty_updates_preserve_suggestions_preference(preferences):
|
|
client, _ = preferences
|
|
await client.put("/api/user-preferences", json={"followup_suggestions_enabled": True})
|
|
|
|
for patch in ({"hide_admin_prompt_prefix": True}, {}, {"followup_suggestions_enabled": None}):
|
|
response = await client.put("/api/user-preferences", json=patch)
|
|
assert response.status_code == 200
|
|
assert response.json() == {"hide_admin_prompt_prefix": True, "followup_suggestions_enabled": True}
|