Merge pull request #492 from christianlouis/copilot/fix-pricing-page-issues
fix: merge main, resolve test failures, and patch CodeQL CWE-312 sensitive data logging
This commit is contained in:
Christian Krakau-Louis
+166
-13
@@ -375,6 +375,7 @@ class TestOAuthCallback:
|
||||
|
||||
mock_request = MagicMock(spec=Request)
|
||||
mock_request.session = {}
|
||||
mock_db = MagicMock()
|
||||
|
||||
userinfo = {
|
||||
"email": "test@example.com",
|
||||
@@ -388,11 +389,12 @@ class TestOAuthCallback:
|
||||
with (
|
||||
patch("app.auth.oauth") as mock_oauth,
|
||||
patch("app.auth.settings") as mock_settings,
|
||||
patch("app.auth._ensure_user_profile"),
|
||||
):
|
||||
mock_oauth.authentik = mock_authentik
|
||||
mock_settings.admin_group_name = "admin"
|
||||
|
||||
result = await oauth_callback(mock_request)
|
||||
result = await oauth_callback(mock_request, db=mock_db)
|
||||
|
||||
assert isinstance(result, RedirectResponse)
|
||||
assert result.status_code == status.HTTP_302_FOUND
|
||||
@@ -407,6 +409,7 @@ class TestOAuthCallback:
|
||||
|
||||
mock_request = MagicMock(spec=Request)
|
||||
mock_request.session = {}
|
||||
mock_db = MagicMock()
|
||||
|
||||
userinfo = {"email": "test@example.com", "name": "Test User"}
|
||||
|
||||
@@ -416,11 +419,12 @@ class TestOAuthCallback:
|
||||
with (
|
||||
patch("app.auth.oauth") as mock_oauth,
|
||||
patch("app.auth.settings") as mock_settings,
|
||||
patch("app.auth._ensure_user_profile"),
|
||||
):
|
||||
mock_oauth.authentik = mock_authentik
|
||||
mock_settings.admin_group_name = "admin"
|
||||
|
||||
result = await oauth_callback(mock_request)
|
||||
result = await oauth_callback(mock_request, db=mock_db)
|
||||
|
||||
# Gravatar should be added
|
||||
assert "picture" in mock_request.session["user"]
|
||||
@@ -433,6 +437,7 @@ class TestOAuthCallback:
|
||||
|
||||
mock_request = MagicMock(spec=Request)
|
||||
mock_request.session = {}
|
||||
mock_db = MagicMock()
|
||||
|
||||
userinfo = {
|
||||
"email": "test@example.com",
|
||||
@@ -446,11 +451,12 @@ class TestOAuthCallback:
|
||||
with (
|
||||
patch("app.auth.oauth") as mock_oauth,
|
||||
patch("app.auth.settings") as mock_settings,
|
||||
patch("app.auth._ensure_user_profile"),
|
||||
):
|
||||
mock_oauth.authentik = mock_authentik
|
||||
mock_settings.admin_group_name = "admin"
|
||||
|
||||
result = await oauth_callback(mock_request)
|
||||
result = await oauth_callback(mock_request, db=mock_db)
|
||||
|
||||
# Custom picture should be preserved, not replaced with Gravatar
|
||||
assert mock_request.session["user"]["picture"] == "https://example.com/custom-pic.jpg"
|
||||
@@ -462,6 +468,7 @@ class TestOAuthCallback:
|
||||
|
||||
mock_request = MagicMock(spec=Request)
|
||||
mock_request.session = {}
|
||||
mock_db = MagicMock()
|
||||
|
||||
userinfo = {
|
||||
"email": "admin@example.com",
|
||||
@@ -475,11 +482,12 @@ class TestOAuthCallback:
|
||||
with (
|
||||
patch("app.auth.oauth") as mock_oauth,
|
||||
patch("app.auth.settings") as mock_settings,
|
||||
patch("app.auth._ensure_user_profile"),
|
||||
):
|
||||
mock_oauth.authentik = mock_authentik
|
||||
mock_settings.admin_group_name = "admin"
|
||||
|
||||
result = await oauth_callback(mock_request)
|
||||
result = await oauth_callback(mock_request, db=mock_db)
|
||||
|
||||
# User should be marked as admin
|
||||
assert mock_request.session["user"]["is_admin"] is True
|
||||
@@ -491,6 +499,7 @@ class TestOAuthCallback:
|
||||
|
||||
mock_request = MagicMock(spec=Request)
|
||||
mock_request.session = {}
|
||||
mock_db = MagicMock()
|
||||
|
||||
userinfo = {
|
||||
"email": "user@example.com",
|
||||
@@ -504,11 +513,12 @@ class TestOAuthCallback:
|
||||
with (
|
||||
patch("app.auth.oauth") as mock_oauth,
|
||||
patch("app.auth.settings") as mock_settings,
|
||||
patch("app.auth._ensure_user_profile"),
|
||||
):
|
||||
mock_oauth.authentik = mock_authentik
|
||||
mock_settings.admin_group_name = "admin"
|
||||
|
||||
result = await oauth_callback(mock_request)
|
||||
result = await oauth_callback(mock_request, db=mock_db)
|
||||
|
||||
# User should not be marked as admin
|
||||
assert mock_request.session["user"]["is_admin"] is False
|
||||
@@ -520,6 +530,7 @@ class TestOAuthCallback:
|
||||
|
||||
mock_request = MagicMock(spec=Request)
|
||||
mock_request.session = {}
|
||||
mock_db = MagicMock()
|
||||
|
||||
mock_authentik = MagicMock()
|
||||
mock_authentik.authorize_access_token = AsyncMock(return_value={})
|
||||
@@ -527,7 +538,7 @@ class TestOAuthCallback:
|
||||
with patch("app.auth.oauth") as mock_oauth:
|
||||
mock_oauth.authentik = mock_authentik
|
||||
|
||||
result = await oauth_callback(mock_request)
|
||||
result = await oauth_callback(mock_request, db=mock_db)
|
||||
|
||||
assert isinstance(result, RedirectResponse)
|
||||
assert "/login?error=Failed+to+retrieve+user+information" in result.headers["location"]
|
||||
@@ -539,6 +550,7 @@ class TestOAuthCallback:
|
||||
|
||||
mock_request = MagicMock(spec=Request)
|
||||
mock_request.session = {"redirect_after_login": "/protected/page"}
|
||||
mock_db = MagicMock()
|
||||
|
||||
userinfo = {"email": "test@example.com", "name": "Test User"}
|
||||
|
||||
@@ -548,11 +560,12 @@ class TestOAuthCallback:
|
||||
with (
|
||||
patch("app.auth.oauth") as mock_oauth,
|
||||
patch("app.auth.settings") as mock_settings,
|
||||
patch("app.auth._ensure_user_profile"),
|
||||
):
|
||||
mock_oauth.authentik = mock_authentik
|
||||
mock_settings.admin_group_name = "admin"
|
||||
|
||||
result = await oauth_callback(mock_request)
|
||||
result = await oauth_callback(mock_request, db=mock_db)
|
||||
|
||||
assert isinstance(result, RedirectResponse)
|
||||
assert result.headers["location"] == "/protected/page"
|
||||
@@ -566,6 +579,7 @@ class TestOAuthCallback:
|
||||
|
||||
mock_request = MagicMock(spec=Request)
|
||||
mock_request.session = {}
|
||||
mock_db = MagicMock()
|
||||
|
||||
mock_authentik = MagicMock()
|
||||
mock_authentik.authorize_access_token = AsyncMock(side_effect=Exception("OAuth error"))
|
||||
@@ -573,19 +587,158 @@ class TestOAuthCallback:
|
||||
with patch("app.auth.oauth") as mock_oauth:
|
||||
mock_oauth.authentik = mock_authentik
|
||||
|
||||
result = await oauth_callback(mock_request)
|
||||
result = await oauth_callback(mock_request, db=mock_db)
|
||||
|
||||
assert isinstance(result, RedirectResponse)
|
||||
assert "/login?error=Authentication+failed" in result.headers["location"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_oauth_callback_creates_user_profile(self):
|
||||
"""Test OAuth callback auto-creates a UserProfile for the authenticated user."""
|
||||
from app.auth import oauth_callback
|
||||
|
||||
mock_request = MagicMock(spec=Request)
|
||||
mock_request.session = {}
|
||||
mock_db = MagicMock()
|
||||
|
||||
userinfo = {
|
||||
"sub": "oauth-sub-abc123",
|
||||
"email": "new@example.com",
|
||||
"name": "New User",
|
||||
"preferred_username": "newuser",
|
||||
}
|
||||
|
||||
mock_authentik = MagicMock()
|
||||
mock_authentik.authorize_access_token = AsyncMock(return_value={"userinfo": userinfo})
|
||||
|
||||
with (
|
||||
patch("app.auth.oauth") as mock_oauth,
|
||||
patch("app.auth.settings") as mock_settings,
|
||||
patch("app.auth._ensure_user_profile") as mock_ensure,
|
||||
):
|
||||
mock_oauth.authentik = mock_authentik
|
||||
mock_settings.admin_group_name = "admin"
|
||||
|
||||
await oauth_callback(mock_request, db=mock_db)
|
||||
|
||||
# _ensure_user_profile should be called with the db and user_data
|
||||
mock_ensure.assert_called_once()
|
||||
call_args = mock_ensure.call_args
|
||||
assert call_args[0][0] is mock_db
|
||||
assert call_args[0][1]["sub"] == "oauth-sub-abc123"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestEnsureUserProfile:
|
||||
"""Tests for _ensure_user_profile() helper."""
|
||||
|
||||
def test_creates_profile_for_new_user(self):
|
||||
"""New user_id should insert a UserProfile row."""
|
||||
from app.auth import _ensure_user_profile
|
||||
from app.models import UserProfile
|
||||
|
||||
mock_db = MagicMock()
|
||||
mock_db.query.return_value.filter.return_value.first.return_value = None
|
||||
|
||||
user_data = {
|
||||
"sub": "sub-xyz",
|
||||
"email": "alice@example.com",
|
||||
"name": "Alice",
|
||||
"preferred_username": "alice",
|
||||
}
|
||||
|
||||
_ensure_user_profile(mock_db, user_data)
|
||||
|
||||
mock_db.add.assert_called_once()
|
||||
added_profile = mock_db.add.call_args[0][0]
|
||||
assert isinstance(added_profile, UserProfile)
|
||||
assert added_profile.user_id == "sub-xyz"
|
||||
assert added_profile.display_name == "Alice"
|
||||
mock_db.commit.assert_called_once()
|
||||
|
||||
def test_skips_existing_profile(self):
|
||||
"""Existing profile should not be overwritten."""
|
||||
from app.auth import _ensure_user_profile
|
||||
from app.models import UserProfile
|
||||
|
||||
mock_db = MagicMock()
|
||||
existing = UserProfile(user_id="sub-xyz", display_name="Old Name")
|
||||
mock_db.query.return_value.filter.return_value.first.return_value = existing
|
||||
|
||||
user_data = {"sub": "sub-xyz", "name": "New Name"}
|
||||
|
||||
_ensure_user_profile(mock_db, user_data)
|
||||
|
||||
mock_db.add.assert_not_called()
|
||||
mock_db.commit.assert_not_called()
|
||||
|
||||
def test_uses_preferred_username_fallback(self):
|
||||
"""Falls back to preferred_username when sub is absent."""
|
||||
from app.auth import _ensure_user_profile
|
||||
|
||||
mock_db = MagicMock()
|
||||
mock_db.query.return_value.filter.return_value.first.return_value = None
|
||||
|
||||
user_data = {"preferred_username": "bob", "name": "Bob"}
|
||||
|
||||
_ensure_user_profile(mock_db, user_data)
|
||||
|
||||
added_profile = mock_db.add.call_args[0][0]
|
||||
assert added_profile.user_id == "bob"
|
||||
|
||||
def test_uses_email_fallback(self):
|
||||
"""Falls back to email when sub and preferred_username are absent."""
|
||||
from app.auth import _ensure_user_profile
|
||||
|
||||
mock_db = MagicMock()
|
||||
mock_db.query.return_value.filter.return_value.first.return_value = None
|
||||
|
||||
user_data = {"email": "carol@example.com", "name": "Carol"}
|
||||
|
||||
_ensure_user_profile(mock_db, user_data)
|
||||
|
||||
added_profile = mock_db.add.call_args[0][0]
|
||||
assert added_profile.user_id == "carol@example.com"
|
||||
|
||||
def test_no_op_when_no_identifier(self):
|
||||
"""Does nothing and logs a warning when no identifier is found."""
|
||||
from app.auth import _ensure_user_profile
|
||||
|
||||
mock_db = MagicMock()
|
||||
|
||||
_ensure_user_profile(mock_db, {})
|
||||
|
||||
mock_db.add.assert_not_called()
|
||||
mock_db.commit.assert_not_called()
|
||||
|
||||
def test_handles_db_exception_gracefully(self):
|
||||
"""DB errors are caught; a rollback is issued and no exception propagates."""
|
||||
from app.auth import _ensure_user_profile
|
||||
|
||||
mock_db = MagicMock()
|
||||
mock_db.query.return_value.filter.return_value.first.return_value = None
|
||||
mock_db.commit.side_effect = Exception("DB error")
|
||||
|
||||
# Should not raise
|
||||
_ensure_user_profile(mock_db, {"sub": "sub-error-test"})
|
||||
|
||||
mock_db.rollback.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestAuthFunction:
|
||||
"""Tests for auth() function (local authentication)."""
|
||||
|
||||
def _make_mock_db(self):
|
||||
"""Create a mock DB that returns None for LocalUser queries (no local users)."""
|
||||
mock_db = MagicMock()
|
||||
# query().filter().first() returns None → no LocalUser found
|
||||
mock_db.query.return_value.filter.return_value.first.return_value = None
|
||||
return mock_db
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_auth_success(self):
|
||||
"""Test successful local authentication."""
|
||||
"""Test successful local authentication (admin fallback)."""
|
||||
from app.auth import auth
|
||||
|
||||
mock_request = MagicMock(spec=Request)
|
||||
@@ -597,7 +750,7 @@ class TestAuthFunction:
|
||||
mock_settings.admin_username = "testadmin"
|
||||
mock_settings.admin_password = "testpass"
|
||||
|
||||
result = await auth(mock_request)
|
||||
result = await auth(mock_request, db=self._make_mock_db())
|
||||
|
||||
assert isinstance(result, RedirectResponse)
|
||||
assert result.status_code == 302
|
||||
@@ -620,7 +773,7 @@ class TestAuthFunction:
|
||||
mock_settings.admin_username = "testadmin"
|
||||
mock_settings.admin_password = "testpass"
|
||||
|
||||
result = await auth(mock_request)
|
||||
result = await auth(mock_request, db=self._make_mock_db())
|
||||
|
||||
assert isinstance(result, RedirectResponse)
|
||||
assert "/login?error=Invalid+username+or+password" in result.headers["location"]
|
||||
@@ -640,7 +793,7 @@ class TestAuthFunction:
|
||||
mock_settings.admin_username = "testadmin"
|
||||
mock_settings.admin_password = "testpass"
|
||||
|
||||
result = await auth(mock_request)
|
||||
result = await auth(mock_request, db=self._make_mock_db())
|
||||
|
||||
assert isinstance(result, RedirectResponse)
|
||||
assert "/login?error=Invalid+username+or+password" in result.headers["location"]
|
||||
@@ -659,7 +812,7 @@ class TestAuthFunction:
|
||||
mock_settings.admin_username = "testadmin"
|
||||
mock_settings.admin_password = "testpass"
|
||||
|
||||
result = await auth(mock_request)
|
||||
result = await auth(mock_request, db=self._make_mock_db())
|
||||
|
||||
assert isinstance(result, RedirectResponse)
|
||||
assert result.headers["location"] == "/settings"
|
||||
|
||||
@@ -270,6 +270,7 @@ class TestAuthEndpoint:
|
||||
with patch("app.auth.settings") as mock_settings:
|
||||
mock_settings.admin_username = "admin"
|
||||
mock_settings.admin_password = "secret123"
|
||||
mock_settings.multi_user_enabled = False
|
||||
|
||||
from app.auth import auth
|
||||
|
||||
@@ -296,6 +297,7 @@ class TestAuthEndpoint:
|
||||
with patch("app.auth.settings") as mock_settings:
|
||||
mock_settings.admin_username = "admin"
|
||||
mock_settings.admin_password = "secret123"
|
||||
mock_settings.multi_user_enabled = False
|
||||
|
||||
from app.auth import auth
|
||||
|
||||
@@ -317,6 +319,7 @@ class TestAuthEndpoint:
|
||||
with patch("app.auth.settings") as mock_settings:
|
||||
mock_settings.admin_username = "admin"
|
||||
mock_settings.admin_password = "secret123"
|
||||
mock_settings.multi_user_enabled = False
|
||||
|
||||
from app.auth import auth
|
||||
|
||||
|
||||
@@ -0,0 +1,486 @@
|
||||
"""Tests for the Stripe billing API endpoints.
|
||||
|
||||
Covers:
|
||||
- POST /api/billing/create-checkout-session
|
||||
- POST /api/billing/create-portal-session
|
||||
- POST /api/billing/webhook (all event types)
|
||||
- GET /api/billing/success
|
||||
- Internal helpers: _handle_stripe_event, _on_checkout_completed,
|
||||
_on_subscription_updated, _on_subscription_deleted
|
||||
"""
|
||||
|
||||
import json
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from sqlalchemy.pool import StaticPool
|
||||
|
||||
from app.api.billing import (
|
||||
_handle_stripe_event,
|
||||
_on_checkout_completed,
|
||||
_on_subscription_deleted,
|
||||
_on_subscription_updated,
|
||||
_resolve_plan_id_from_price,
|
||||
_resolve_user_id_from_customer,
|
||||
)
|
||||
from app.database import Base, get_db
|
||||
from app.models import SubscriptionPlan, UserProfile
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def bill_engine():
|
||||
"""In-memory SQLite engine for billing tests."""
|
||||
engine = create_engine(
|
||||
"sqlite:///:memory:",
|
||||
connect_args={"check_same_thread": False},
|
||||
poolclass=StaticPool,
|
||||
)
|
||||
Base.metadata.create_all(bind=engine)
|
||||
yield engine
|
||||
Base.metadata.drop_all(bind=engine)
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def bill_session(bill_engine):
|
||||
"""DB session for one test."""
|
||||
Session = sessionmaker(bind=bill_engine)
|
||||
session = Session()
|
||||
yield session
|
||||
session.close()
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def bill_client(bill_engine):
|
||||
"""TestClient with DB dependency overridden and a logged-in session."""
|
||||
from app.main import app
|
||||
|
||||
Session = sessionmaker(bind=bill_engine)
|
||||
|
||||
def override_get_db():
|
||||
db = Session()
|
||||
try:
|
||||
yield db
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
app.dependency_overrides[get_db] = override_get_db
|
||||
with TestClient(app, base_url="http://localhost", raise_server_exceptions=False) as client:
|
||||
yield client
|
||||
app.dependency_overrides.pop(get_db, None)
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def starter_plan(bill_session):
|
||||
"""A SubscriptionPlan with Stripe price IDs in the DB."""
|
||||
plan = SubscriptionPlan(
|
||||
plan_id="starter",
|
||||
name="Starter",
|
||||
price_monthly=9.0,
|
||||
price_yearly=90.0,
|
||||
trial_days=0,
|
||||
stripe_price_id_monthly="price_monthly_starter",
|
||||
stripe_price_id_yearly="price_yearly_starter",
|
||||
)
|
||||
bill_session.add(plan)
|
||||
bill_session.commit()
|
||||
return plan
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def user_profile(bill_session):
|
||||
"""A UserProfile for user1@example.com."""
|
||||
profile = UserProfile(
|
||||
user_id="user1@example.com",
|
||||
display_name="Test User",
|
||||
stripe_customer_id=None,
|
||||
)
|
||||
bill_session.add(profile)
|
||||
bill_session.commit()
|
||||
return profile
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: _get_stripe returns None when not configured
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_get_stripe_returns_none_when_not_configured():
|
||||
"""_get_stripe returns None when stripe_secret_key is not set."""
|
||||
from app.api.billing import _get_stripe
|
||||
|
||||
with patch("app.api.billing.settings") as mock_settings:
|
||||
mock_settings.stripe_secret_key = None
|
||||
result = _get_stripe()
|
||||
assert result is None
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_get_stripe_returns_client_when_configured():
|
||||
"""_get_stripe returns a StripeClient when key is configured."""
|
||||
from app.api.billing import _get_stripe
|
||||
|
||||
with patch("app.api.billing.settings") as mock_settings:
|
||||
mock_settings.stripe_secret_key = "sk_test_fake"
|
||||
result = _get_stripe()
|
||||
assert result is not None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: create-checkout-session
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_create_checkout_session_stripe_not_configured(bill_client):
|
||||
"""POST /api/billing/create-checkout-session returns 503 when Stripe not set."""
|
||||
with patch("app.api.billing._get_stripe", return_value=None):
|
||||
resp = bill_client.post(
|
||||
"/api/billing/create-checkout-session",
|
||||
json={"plan_id": "starter", "billing_cycle": "monthly"},
|
||||
)
|
||||
assert resp.status_code == 503
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_create_checkout_session_plan_not_found(bill_client):
|
||||
"""POST /api/billing/create-checkout-session returns 404 for unknown plan."""
|
||||
mock_client = MagicMock()
|
||||
with patch("app.api.billing._get_stripe", return_value=mock_client):
|
||||
resp = bill_client.post(
|
||||
"/api/billing/create-checkout-session",
|
||||
json={"plan_id": "nonexistent", "billing_cycle": "monthly"},
|
||||
)
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_create_checkout_session_no_price_id(bill_client, bill_session):
|
||||
"""POST /api/billing/create-checkout-session returns 404 when price ID not set."""
|
||||
plan = SubscriptionPlan(
|
||||
plan_id="noprice",
|
||||
name="No Price",
|
||||
price_monthly=5.0,
|
||||
price_yearly=50.0,
|
||||
trial_days=0,
|
||||
stripe_price_id_monthly=None,
|
||||
stripe_price_id_yearly=None,
|
||||
)
|
||||
bill_session.add(plan)
|
||||
bill_session.commit()
|
||||
|
||||
mock_client = MagicMock()
|
||||
with patch("app.api.billing._get_stripe", return_value=mock_client):
|
||||
resp = bill_client.post(
|
||||
"/api/billing/create-checkout-session",
|
||||
json={"plan_id": "noprice", "billing_cycle": "monthly"},
|
||||
)
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_create_checkout_session_success(bill_client, starter_plan, user_profile):
|
||||
"""POST /api/billing/create-checkout-session returns checkout_url on success."""
|
||||
mock_client = MagicMock()
|
||||
mock_customer = MagicMock()
|
||||
mock_customer.id = "cus_test123"
|
||||
mock_session = MagicMock()
|
||||
mock_session.id = "cs_test456"
|
||||
mock_session.url = "https://checkout.stripe.com/test"
|
||||
|
||||
mock_client.customers.create.return_value = mock_customer
|
||||
mock_client.checkout.sessions.create.return_value = mock_session
|
||||
|
||||
with (
|
||||
patch("app.api.billing._get_stripe", return_value=mock_client),
|
||||
patch("app.api.billing.get_current_owner_id", return_value="user1@example.com"),
|
||||
):
|
||||
resp = bill_client.post(
|
||||
"/api/billing/create-checkout-session",
|
||||
json={"plan_id": "starter", "billing_cycle": "monthly"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "checkout_url" in data
|
||||
assert data["checkout_url"] == "https://checkout.stripe.com/test"
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_create_checkout_session_yearly(bill_client, starter_plan, user_profile):
|
||||
"""POST /api/billing/create-checkout-session uses yearly price ID for yearly cycle."""
|
||||
mock_client = MagicMock()
|
||||
mock_customer = MagicMock()
|
||||
mock_customer.id = "cus_test123"
|
||||
mock_session = MagicMock()
|
||||
mock_session.id = "cs_test456"
|
||||
mock_session.url = "https://checkout.stripe.com/yearly"
|
||||
|
||||
mock_client.customers.create.return_value = mock_customer
|
||||
mock_client.checkout.sessions.create.return_value = mock_session
|
||||
|
||||
with (
|
||||
patch("app.api.billing._get_stripe", return_value=mock_client),
|
||||
patch("app.api.billing.get_current_owner_id", return_value="user1@example.com"),
|
||||
):
|
||||
resp = bill_client.post(
|
||||
"/api/billing/create-checkout-session",
|
||||
json={"plan_id": "starter", "billing_cycle": "yearly"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
# Verify yearly price ID was used
|
||||
call_params = mock_client.checkout.sessions.create.call_args[1]["params"]
|
||||
assert call_params["line_items"][0]["price"] == "price_yearly_starter"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: create-portal-session
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_create_portal_session_stripe_not_configured(bill_client):
|
||||
"""POST /api/billing/create-portal-session returns 503 when not configured."""
|
||||
with patch("app.api.billing._get_stripe", return_value=None):
|
||||
resp = bill_client.post("/api/billing/create-portal-session", json={})
|
||||
assert resp.status_code == 503
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_create_portal_session_no_customer(bill_client, user_profile):
|
||||
"""POST /api/billing/create-portal-session returns 404 when no Stripe customer."""
|
||||
mock_client = MagicMock()
|
||||
with (
|
||||
patch("app.api.billing._get_stripe", return_value=mock_client),
|
||||
patch("app.api.billing.get_current_owner_id", return_value="user1@example.com"),
|
||||
):
|
||||
resp = bill_client.post("/api/billing/create-portal-session", json={})
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_create_portal_session_success(bill_client, bill_session, user_profile):
|
||||
"""POST /api/billing/create-portal-session returns portal_url on success."""
|
||||
user_profile.stripe_customer_id = "cus_existing"
|
||||
bill_session.commit()
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_portal = MagicMock()
|
||||
mock_portal.url = "https://billing.stripe.com/portal/test"
|
||||
mock_client.billing_portal.sessions.create.return_value = mock_portal
|
||||
|
||||
with (
|
||||
patch("app.api.billing._get_stripe", return_value=mock_client),
|
||||
patch("app.api.billing.get_current_owner_id", return_value="user1@example.com"),
|
||||
):
|
||||
resp = bill_client.post("/api/billing/create-portal-session", json={})
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["portal_url"] == "https://billing.stripe.com/portal/test"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: webhook
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_webhook_not_configured(bill_client):
|
||||
"""POST /api/billing/webhook returns 503 when billing not configured."""
|
||||
with patch("app.api.billing.settings") as mock_settings:
|
||||
mock_settings.stripe_secret_key = None
|
||||
mock_settings.stripe_webhook_secret = None
|
||||
resp = bill_client.post(
|
||||
"/api/billing/webhook",
|
||||
content=b"{}",
|
||||
headers={"content-type": "application/json"},
|
||||
)
|
||||
assert resp.status_code == 503
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_webhook_invalid_signature(bill_client):
|
||||
"""POST /api/billing/webhook returns 400 on invalid Stripe signature."""
|
||||
import stripe
|
||||
|
||||
with (
|
||||
patch("app.api.billing.settings") as mock_settings,
|
||||
patch("stripe.Webhook.construct_event", side_effect=stripe.SignatureVerificationError("bad", "sig")),
|
||||
):
|
||||
mock_settings.stripe_secret_key = "sk_test_fake"
|
||||
mock_settings.stripe_webhook_secret = "whsec_test"
|
||||
resp = bill_client.post(
|
||||
"/api/billing/webhook",
|
||||
content=b'{"type":"test"}',
|
||||
headers={"stripe-signature": "bad_sig", "content-type": "application/json"},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_webhook_checkout_completed(bill_client, bill_session, starter_plan, user_profile):
|
||||
"""POST /api/billing/webhook activates plan on checkout.session.completed."""
|
||||
payload = json.dumps(
|
||||
{
|
||||
"type": "checkout.session.completed",
|
||||
"data": {
|
||||
"object": {
|
||||
"customer": "cus_new",
|
||||
"metadata": {
|
||||
"docuelevate_user_id": "user1@example.com",
|
||||
"plan_id": "starter",
|
||||
"billing_cycle": "monthly",
|
||||
},
|
||||
}
|
||||
},
|
||||
}
|
||||
).encode()
|
||||
|
||||
with (
|
||||
patch("app.api.billing.settings") as mock_settings,
|
||||
patch("stripe.Event.construct_from", return_value=json.loads(payload)),
|
||||
):
|
||||
mock_settings.stripe_secret_key = "sk_test_fake"
|
||||
mock_settings.stripe_webhook_secret = None
|
||||
resp = bill_client.post(
|
||||
"/api/billing/webhook",
|
||||
content=payload,
|
||||
headers={"content-type": "application/json"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
|
||||
bill_session.expire_all()
|
||||
profile = bill_session.query(UserProfile).filter(UserProfile.user_id == "user1@example.com").first()
|
||||
assert profile.subscription_tier == "starter"
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_webhook_subscription_deleted(bill_client, bill_session, user_profile):
|
||||
"""POST /api/billing/webhook downgrades to free on subscription deleted."""
|
||||
user_profile.stripe_customer_id = "cus_del"
|
||||
user_profile.subscription_tier = "starter"
|
||||
bill_session.commit()
|
||||
|
||||
payload = json.dumps(
|
||||
{
|
||||
"type": "customer.subscription.deleted",
|
||||
"data": {"object": {"customer": "cus_del"}},
|
||||
}
|
||||
).encode()
|
||||
|
||||
with (
|
||||
patch("app.api.billing.settings") as mock_settings,
|
||||
patch("stripe.Event.construct_from", return_value=json.loads(payload)),
|
||||
):
|
||||
mock_settings.stripe_secret_key = "sk_test_fake"
|
||||
mock_settings.stripe_webhook_secret = None
|
||||
resp = bill_client.post(
|
||||
"/api/billing/webhook",
|
||||
content=payload,
|
||||
headers={"content-type": "application/json"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
|
||||
bill_session.expire_all()
|
||||
profile = bill_session.query(UserProfile).filter(UserProfile.user_id == "user1@example.com").first()
|
||||
assert profile.subscription_tier == "free"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Unit tests: internal helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_resolve_user_id_from_customer(bill_session, user_profile):
|
||||
"""_resolve_user_id_from_customer returns user_id for known Stripe customer."""
|
||||
user_profile.stripe_customer_id = "cus_known"
|
||||
bill_session.commit()
|
||||
result = _resolve_user_id_from_customer(bill_session, "cus_known")
|
||||
assert result == "user1@example.com"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_resolve_user_id_from_customer_unknown(bill_session):
|
||||
"""_resolve_user_id_from_customer returns None for unknown customer."""
|
||||
result = _resolve_user_id_from_customer(bill_session, "cus_unknown")
|
||||
assert result is None
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_resolve_plan_id_from_price(bill_session, starter_plan):
|
||||
"""_resolve_plan_id_from_price finds plan by monthly price ID."""
|
||||
result = _resolve_plan_id_from_price(bill_session, "price_monthly_starter")
|
||||
assert result == "starter"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_resolve_plan_id_from_price_yearly(bill_session, starter_plan):
|
||||
"""_resolve_plan_id_from_price finds plan by yearly price ID."""
|
||||
result = _resolve_plan_id_from_price(bill_session, "price_yearly_starter")
|
||||
assert result == "starter"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_resolve_plan_id_from_price_unknown(bill_session):
|
||||
"""_resolve_plan_id_from_price returns None for unknown price."""
|
||||
result = _resolve_plan_id_from_price(bill_session, "price_unknown")
|
||||
assert result is None
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_on_checkout_completed_missing_user_id(bill_session):
|
||||
"""_on_checkout_completed does nothing when user_id is absent."""
|
||||
data = {"metadata": {}, "customer": "cus_test"}
|
||||
_on_checkout_completed(bill_session, data) # Should not raise
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_on_subscription_updated_no_items(bill_session, user_profile):
|
||||
"""_on_subscription_updated does nothing when items list is empty."""
|
||||
user_profile.stripe_customer_id = "cus_upd"
|
||||
bill_session.commit()
|
||||
data = {"customer": "cus_upd", "items": {"data": []}}
|
||||
_on_subscription_updated(bill_session, data) # Should not raise
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_on_subscription_deleted_unknown_customer(bill_session):
|
||||
"""_on_subscription_deleted does nothing for unknown customer."""
|
||||
data = {"customer": "cus_nobody"}
|
||||
_on_subscription_deleted(bill_session, data) # Should not raise
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_handle_stripe_event_unhandled_type(bill_session):
|
||||
"""_handle_stripe_event logs but does not raise for unknown event types."""
|
||||
event = {"type": "unknown.event.type", "data": {"object": {}}}
|
||||
_handle_stripe_event(bill_session, event) # Should not raise
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_handle_stripe_event_payment_failed(bill_session):
|
||||
"""_handle_stripe_event handles invoice.payment_failed without raising."""
|
||||
event = {
|
||||
"type": "invoice.payment_failed",
|
||||
"data": {"object": {"customer": "cus_fail"}},
|
||||
}
|
||||
_handle_stripe_event(bill_session, event) # Should not raise
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: billing success page
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_billing_success_page(bill_client):
|
||||
"""GET /api/billing/success returns 200 for logged-in user."""
|
||||
resp = bill_client.get("/api/billing/success")
|
||||
assert resp.status_code == 200
|
||||
assert b"subscription" in resp.content.lower() or b"success" in resp.content.lower()
|
||||
@@ -0,0 +1,677 @@
|
||||
"""Tests for local user authentication: signup, email verification, and password reset.
|
||||
|
||||
Covers:
|
||||
- POST /api/auth/signup (success, disabled, SMTP missing, password mismatch, conflicts)
|
||||
- GET /verify-email (valid token, invalid token, expired token)
|
||||
- POST /api/auth/resend-verification
|
||||
- POST /api/auth/request-password-reset
|
||||
- POST /api/auth/reset-password
|
||||
- GET /signup (page route)
|
||||
- GET /verify-email-sent (page route)
|
||||
- GET /reset-password (page route)
|
||||
- app/utils/local_auth utility functions
|
||||
- auth() login flow with LocalUser
|
||||
"""
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from sqlalchemy.pool import StaticPool
|
||||
|
||||
from app.config import settings
|
||||
from app.database import Base, get_db
|
||||
from app.models import LocalUser, UserProfile
|
||||
from app.utils.local_auth import (
|
||||
build_session_user,
|
||||
generate_token,
|
||||
hash_password,
|
||||
is_token_expired,
|
||||
verify_password,
|
||||
)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_TEST_DB_URL = "sqlite:///:memory:"
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def la_engine():
|
||||
"""In-memory SQLite engine for local auth tests."""
|
||||
engine = create_engine(
|
||||
_TEST_DB_URL,
|
||||
connect_args={"check_same_thread": False},
|
||||
poolclass=StaticPool,
|
||||
)
|
||||
Base.metadata.create_all(bind=engine)
|
||||
yield engine
|
||||
Base.metadata.drop_all(bind=engine)
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def la_session(la_engine):
|
||||
"""DB session for one test."""
|
||||
Session = sessionmaker(bind=la_engine)
|
||||
session = Session()
|
||||
yield session
|
||||
session.close()
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def la_client(la_engine):
|
||||
"""TestClient with DB dependency overridden."""
|
||||
from app.main import app
|
||||
|
||||
Session = sessionmaker(bind=la_engine)
|
||||
|
||||
def override_get_db():
|
||||
db = Session()
|
||||
try:
|
||||
yield db
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
app.dependency_overrides[get_db] = override_get_db
|
||||
with TestClient(app, base_url="http://localhost", raise_server_exceptions=True) as client:
|
||||
yield client
|
||||
app.dependency_overrides.pop(get_db, None)
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def active_user(la_session):
|
||||
"""A fully active LocalUser in the DB."""
|
||||
user = LocalUser(
|
||||
email="active@example.com",
|
||||
username="activeuser",
|
||||
display_name="Active User",
|
||||
hashed_password=hash_password("password123"),
|
||||
is_active=True,
|
||||
)
|
||||
la_session.add(user)
|
||||
la_session.add(UserProfile(user_id="active@example.com", display_name="Active User", onboarding_completed=True))
|
||||
la_session.commit()
|
||||
return user
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def pending_user(la_session):
|
||||
"""A LocalUser with a pending email verification token."""
|
||||
token = "validtoken123"
|
||||
user = LocalUser(
|
||||
email="pending@example.com",
|
||||
username="pendinguser",
|
||||
hashed_password=hash_password("password123"),
|
||||
is_active=False,
|
||||
email_verification_token=token,
|
||||
email_verification_sent_at=datetime.now(tz=timezone.utc),
|
||||
)
|
||||
la_session.add(user)
|
||||
la_session.commit()
|
||||
return user
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Unit tests: local_auth utilities
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_hash_and_verify_password():
|
||||
"""hash_password produces a bcrypt hash that verify_password validates."""
|
||||
plain = "super$ecret99"
|
||||
hashed = hash_password(plain)
|
||||
assert hashed != plain
|
||||
assert verify_password(plain, hashed) is True
|
||||
assert verify_password("wrong", hashed) is False
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_verify_password_bad_hash_returns_false():
|
||||
"""verify_password returns False for a non-bcrypt string."""
|
||||
assert verify_password("any", "notahash") is False
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_generate_token_unique():
|
||||
"""generate_token returns distinct non-empty strings."""
|
||||
tokens = {generate_token() for _ in range(10)}
|
||||
assert len(tokens) == 10
|
||||
for t in tokens:
|
||||
assert len(t) > 20
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_is_token_expired_none():
|
||||
"""None sent_at is treated as expired."""
|
||||
assert is_token_expired(None) is True
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_is_token_expired_old():
|
||||
"""Token sent more than 24 h ago is expired."""
|
||||
old = datetime.now(tz=timezone.utc) - timedelta(hours=25)
|
||||
assert is_token_expired(old) is True
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_is_token_expired_fresh():
|
||||
"""Token sent recently is not expired."""
|
||||
fresh = datetime.now(tz=timezone.utc) - timedelta(hours=1)
|
||||
assert is_token_expired(fresh) is False
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_build_session_user():
|
||||
"""build_session_user returns the expected dict structure."""
|
||||
user = MagicMock()
|
||||
user.email = "u@example.com"
|
||||
user.username = "uname"
|
||||
user.display_name = "Display Name"
|
||||
user.is_admin = False
|
||||
with patch("app.auth.get_gravatar_url", return_value="https://gravatar.com/test"):
|
||||
result = build_session_user(user)
|
||||
assert result["email"] == "u@example.com"
|
||||
assert result["preferred_username"] == "uname"
|
||||
assert result["name"] == "Display Name"
|
||||
assert result["is_admin"] is False
|
||||
assert result["auth_method"] == "local"
|
||||
assert "picture" in result
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Integration tests: signup
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_signup_disabled(la_client):
|
||||
"""POST /api/auth/signup returns 403 when allow_local_signup is False."""
|
||||
with patch("app.api.local_auth.settings") as mock_settings:
|
||||
mock_settings.allow_local_signup = False
|
||||
mock_settings.email_host = "smtp.example.com"
|
||||
resp = la_client.post(
|
||||
"/api/auth/signup",
|
||||
json={
|
||||
"email": "a@example.com",
|
||||
"username": "auser",
|
||||
"password": "password1",
|
||||
"password_confirm": "password1",
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_signup_smtp_not_configured(la_client):
|
||||
"""POST /api/auth/signup returns 503 when SMTP is not configured."""
|
||||
with patch("app.api.local_auth.settings") as mock_settings:
|
||||
mock_settings.allow_local_signup = True
|
||||
mock_settings.multi_user_enabled = True
|
||||
mock_settings.email_host = None
|
||||
resp = la_client.post(
|
||||
"/api/auth/signup",
|
||||
json={
|
||||
"email": "a@example.com",
|
||||
"username": "auser",
|
||||
"password": "password1",
|
||||
"password_confirm": "password1",
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 503
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_signup_password_mismatch(la_client):
|
||||
"""POST /api/auth/signup returns 422 when passwords do not match."""
|
||||
with patch("app.api.local_auth.settings") as mock_settings:
|
||||
mock_settings.allow_local_signup = True
|
||||
mock_settings.multi_user_enabled = True
|
||||
mock_settings.email_host = "smtp.example.com"
|
||||
resp = la_client.post(
|
||||
"/api/auth/signup",
|
||||
json={
|
||||
"email": "a@example.com",
|
||||
"username": "auser",
|
||||
"password": "password1",
|
||||
"password_confirm": "different1",
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_signup_success(la_client):
|
||||
"""POST /api/auth/signup creates user and returns 201."""
|
||||
with (
|
||||
patch("app.api.local_auth.settings") as mock_settings,
|
||||
patch("app.api.local_auth.send_verification_email") as mock_send,
|
||||
):
|
||||
mock_settings.allow_local_signup = True
|
||||
mock_settings.multi_user_enabled = True
|
||||
mock_settings.email_host = "smtp.example.com"
|
||||
mock_settings.version = "test"
|
||||
resp = la_client.post(
|
||||
"/api/auth/signup",
|
||||
json={
|
||||
"email": "new@example.com",
|
||||
"username": "newuser",
|
||||
"password": "password1",
|
||||
"password_confirm": "password1",
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
assert "Verification email sent" in resp.json()["message"]
|
||||
mock_send.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_signup_duplicate_email(la_client, active_user):
|
||||
"""POST /api/auth/signup returns 409 when email already registered."""
|
||||
with (
|
||||
patch("app.api.local_auth.settings") as mock_settings,
|
||||
patch("app.api.local_auth.send_verification_email"),
|
||||
):
|
||||
mock_settings.allow_local_signup = True
|
||||
mock_settings.multi_user_enabled = True
|
||||
mock_settings.email_host = "smtp.example.com"
|
||||
resp = la_client.post(
|
||||
"/api/auth/signup",
|
||||
json={
|
||||
"email": "active@example.com",
|
||||
"username": "otheruser",
|
||||
"password": "password1",
|
||||
"password_confirm": "password1",
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 409
|
||||
assert "Email" in resp.json()["detail"]
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_signup_duplicate_username(la_client, active_user):
|
||||
"""POST /api/auth/signup returns 409 when username already taken."""
|
||||
with (
|
||||
patch("app.api.local_auth.settings") as mock_settings,
|
||||
patch("app.api.local_auth.send_verification_email"),
|
||||
):
|
||||
mock_settings.allow_local_signup = True
|
||||
mock_settings.multi_user_enabled = True
|
||||
mock_settings.email_host = "smtp.example.com"
|
||||
resp = la_client.post(
|
||||
"/api/auth/signup",
|
||||
json={
|
||||
"email": "different@example.com",
|
||||
"username": "activeuser",
|
||||
"password": "password1",
|
||||
"password_confirm": "password1",
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 409
|
||||
assert "Username" in resp.json()["detail"]
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_signup_smtp_failure_cleans_up(la_client, la_session):
|
||||
"""POST /api/auth/signup cleans up user records if email send fails."""
|
||||
with (
|
||||
patch("app.api.local_auth.settings") as mock_settings,
|
||||
patch("app.api.local_auth.send_verification_email", side_effect=RuntimeError("SMTP down")),
|
||||
):
|
||||
mock_settings.allow_local_signup = True
|
||||
mock_settings.multi_user_enabled = True
|
||||
mock_settings.email_host = "smtp.example.com"
|
||||
resp = la_client.post(
|
||||
"/api/auth/signup",
|
||||
json={
|
||||
"email": "fail@example.com",
|
||||
"username": "failuser",
|
||||
"password": "password1",
|
||||
"password_confirm": "password1",
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 503
|
||||
# User should NOT exist in the DB
|
||||
user = la_session.query(LocalUser).filter(LocalUser.email == "fail@example.com").first()
|
||||
assert user is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Integration tests: email verification
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_verify_email_valid_token(la_client, pending_user):
|
||||
"""GET /verify-email with valid token activates account and redirects."""
|
||||
resp = la_client.get(
|
||||
f"/verify-email?token={pending_user.email_verification_token}",
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert resp.status_code == 302
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_verify_email_invalid_token(la_client):
|
||||
"""GET /verify-email with unknown token redirects to login with error."""
|
||||
resp = la_client.get("/verify-email?token=doesnotexist", follow_redirects=False)
|
||||
assert resp.status_code == 302
|
||||
assert "/login" in resp.headers["location"]
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_verify_email_expired_token(la_client, la_session):
|
||||
"""GET /verify-email with expired token redirects to login with error."""
|
||||
old_time = datetime.now(tz=timezone.utc) - timedelta(hours=25)
|
||||
user = LocalUser(
|
||||
email="expired@example.com",
|
||||
username="expireduser",
|
||||
hashed_password=hash_password("password123"),
|
||||
is_active=False,
|
||||
email_verification_token="expiredtoken",
|
||||
email_verification_sent_at=old_time,
|
||||
)
|
||||
la_session.add(user)
|
||||
la_session.commit()
|
||||
|
||||
resp = la_client.get("/verify-email?token=expiredtoken", follow_redirects=False)
|
||||
assert resp.status_code == 302
|
||||
assert "/login" in resp.headers["location"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Integration tests: resend verification
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_resend_verification_always_200(la_client):
|
||||
"""POST /api/auth/resend-verification returns 200 for unknown email."""
|
||||
with patch("app.api.local_auth.send_verification_email"):
|
||||
resp = la_client.post(
|
||||
"/api/auth/resend-verification",
|
||||
json={"email": "nobody@example.com"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_resend_verification_sends_email(la_client, pending_user):
|
||||
"""POST /api/auth/resend-verification sends email for pending user."""
|
||||
with patch("app.api.local_auth.send_verification_email") as mock_send:
|
||||
resp = la_client.post(
|
||||
"/api/auth/resend-verification",
|
||||
json={"email": pending_user.email},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
mock_send.assert_called_once()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Integration tests: password reset
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_request_password_reset_always_200(la_client):
|
||||
"""POST /api/auth/request-password-reset returns 200 for unknown email."""
|
||||
with patch("app.api.local_auth.send_password_reset_email"):
|
||||
resp = la_client.post(
|
||||
"/api/auth/request-password-reset",
|
||||
json={"email": "nobody@example.com"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_request_password_reset_sends_email(la_client, active_user):
|
||||
"""POST /api/auth/request-password-reset sends email for known user."""
|
||||
with patch("app.api.local_auth.send_password_reset_email") as mock_send:
|
||||
resp = la_client.post(
|
||||
"/api/auth/request-password-reset",
|
||||
json={"email": active_user.email},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
mock_send.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_reset_password_success(la_client, la_session):
|
||||
"""POST /api/auth/reset-password updates password with valid token."""
|
||||
token = "resettoken123"
|
||||
user = LocalUser(
|
||||
email="reset@example.com",
|
||||
username="resetuser",
|
||||
hashed_password=hash_password("oldpassword"),
|
||||
is_active=True,
|
||||
password_reset_token=token,
|
||||
password_reset_sent_at=datetime.now(tz=timezone.utc),
|
||||
)
|
||||
la_session.add(user)
|
||||
la_session.commit()
|
||||
|
||||
resp = la_client.post(
|
||||
"/api/auth/reset-password",
|
||||
json={
|
||||
"token": token,
|
||||
"new_password": "newpassword1",
|
||||
"new_password_confirm": "newpassword1",
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
la_session.refresh(user)
|
||||
assert verify_password("newpassword1", user.hashed_password)
|
||||
assert user.password_reset_token is None
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_reset_password_invalid_token(la_client):
|
||||
"""POST /api/auth/reset-password returns 400 for invalid token."""
|
||||
resp = la_client.post(
|
||||
"/api/auth/reset-password",
|
||||
json={
|
||||
"token": "badtoken",
|
||||
"new_password": "newpassword1",
|
||||
"new_password_confirm": "newpassword1",
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_reset_password_mismatch(la_client, la_session):
|
||||
"""POST /api/auth/reset-password returns 422 when passwords do not match."""
|
||||
token = "mismatchtoken"
|
||||
user = LocalUser(
|
||||
email="mismatch@example.com",
|
||||
username="mismatchuser",
|
||||
hashed_password=hash_password("old"),
|
||||
is_active=True,
|
||||
password_reset_token=token,
|
||||
password_reset_sent_at=datetime.now(tz=timezone.utc),
|
||||
)
|
||||
la_session.add(user)
|
||||
la_session.commit()
|
||||
|
||||
resp = la_client.post(
|
||||
"/api/auth/reset-password",
|
||||
json={
|
||||
"token": token,
|
||||
"new_password": "newpassword1",
|
||||
"new_password_confirm": "different_pw",
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Integration tests: page routes
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_signup_page_disabled_redirects(la_client):
|
||||
"""GET /signup redirects when allow_local_signup is False."""
|
||||
with patch("app.api.local_auth.settings") as mock_settings:
|
||||
mock_settings.allow_local_signup = False
|
||||
resp = la_client.get("/signup", follow_redirects=False)
|
||||
assert resp.status_code == 302
|
||||
assert "/login" in resp.headers["location"]
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_signup_page_enabled(la_client):
|
||||
"""GET /signup returns 200 when allow_local_signup is True."""
|
||||
with patch("app.api.local_auth.settings") as mock_settings:
|
||||
mock_settings.allow_local_signup = True
|
||||
mock_settings.multi_user_enabled = True
|
||||
mock_settings.version = "test"
|
||||
resp = la_client.get("/signup")
|
||||
assert resp.status_code == 200
|
||||
assert b"Create" in resp.content
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_verify_email_sent_page(la_client):
|
||||
"""GET /verify-email-sent returns 200."""
|
||||
resp = la_client.get("/verify-email-sent")
|
||||
assert resp.status_code == 200
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_reset_password_page(la_client):
|
||||
"""GET /reset-password returns 200."""
|
||||
with patch("app.api.local_auth.settings") as mock_settings:
|
||||
mock_settings.version = "test"
|
||||
resp = la_client.get("/reset-password?token=abc123")
|
||||
assert resp.status_code == 200
|
||||
assert b"password" in resp.content.lower()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Integration tests: auth() login flow with LocalUser
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.asyncio
|
||||
@patch.object(settings, "multi_user_enabled", True)
|
||||
async def test_local_login_success(la_session, active_user):
|
||||
"""auth() with valid LocalUser credentials sets session and redirects."""
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
from fastapi import Request
|
||||
|
||||
from app.auth import auth
|
||||
|
||||
mock_request = MagicMock(spec=Request)
|
||||
mock_request.form = AsyncMock(return_value={"username": "activeuser", "password": "password123"})
|
||||
mock_request.session = {}
|
||||
|
||||
result = await auth(mock_request, db=la_session)
|
||||
assert result.status_code == 302
|
||||
assert "user" in mock_request.session
|
||||
assert mock_request.session["user"]["email"] == "active@example.com"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.asyncio
|
||||
@patch.object(settings, "multi_user_enabled", True)
|
||||
async def test_local_login_by_email(la_session, active_user):
|
||||
"""auth() accepts email as username for LocalUser lookup."""
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
from fastapi import Request
|
||||
|
||||
from app.auth import auth
|
||||
|
||||
mock_request = MagicMock(spec=Request)
|
||||
mock_request.form = AsyncMock(return_value={"username": "active@example.com", "password": "password123"})
|
||||
mock_request.session = {}
|
||||
|
||||
result = await auth(mock_request, db=la_session)
|
||||
assert result.status_code == 302
|
||||
assert "user" in mock_request.session
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.asyncio
|
||||
@patch.object(settings, "multi_user_enabled", True)
|
||||
async def test_local_login_wrong_password(la_session, active_user):
|
||||
"""auth() with wrong password redirects to login with error."""
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
from fastapi import Request
|
||||
|
||||
from app.auth import auth
|
||||
|
||||
mock_request = MagicMock(spec=Request)
|
||||
mock_request.form = AsyncMock(return_value={"username": "activeuser", "password": "wrongpassword"})
|
||||
mock_request.session = {}
|
||||
|
||||
result = await auth(mock_request, db=la_session)
|
||||
assert result.status_code == 302
|
||||
assert "/login" in result.headers["location"]
|
||||
assert "user" not in mock_request.session
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.asyncio
|
||||
@patch.object(settings, "multi_user_enabled", True)
|
||||
async def test_local_login_unverified(la_session, pending_user):
|
||||
"""auth() for unverified user redirects with verification message."""
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
from fastapi import Request
|
||||
|
||||
from app.auth import auth
|
||||
|
||||
mock_request = MagicMock(spec=Request)
|
||||
mock_request.form = AsyncMock(return_value={"username": "pendinguser", "password": "password123"})
|
||||
mock_request.session = {}
|
||||
|
||||
result = await auth(mock_request, db=la_session)
|
||||
assert result.status_code == 302
|
||||
assert "verify" in result.headers["location"].lower()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Single-user backward-compatibility: LocalUser table must NOT be queried
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.asyncio
|
||||
@patch.object(settings, "multi_user_enabled", False)
|
||||
async def test_single_user_mode_skips_local_user_table(la_session, active_user):
|
||||
"""In single-user mode auth() must not query LocalUser even when a matching
|
||||
row exists. It should fall through to the admin-credential check."""
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
from unittest.mock import patch as _patch
|
||||
|
||||
from fastapi import Request
|
||||
|
||||
from app.auth import auth
|
||||
|
||||
mock_request = MagicMock(spec=Request)
|
||||
# Use the active LocalUser's credentials — they must NOT work in single-user mode
|
||||
# because the whole LocalUser block is skipped.
|
||||
mock_request.form = AsyncMock(return_value={"username": "activeuser", "password": "password123"})
|
||||
mock_request.session = {}
|
||||
|
||||
with (
|
||||
_patch.object(settings, "admin_username", "activeuser"),
|
||||
_patch.object(settings, "admin_password", "password123"),
|
||||
):
|
||||
result = await auth(mock_request, db=la_session)
|
||||
|
||||
# Should succeed via admin-credentials path (is_admin=True), not LocalUser path
|
||||
assert result.status_code == 302
|
||||
assert "user" in mock_request.session
|
||||
# Admin path sets is_admin=True and id="admin"
|
||||
assert mock_request.session["user"]["is_admin"] is True
|
||||
assert mock_request.session["user"]["id"] == "admin"
|
||||
@@ -0,0 +1,317 @@
|
||||
"""Unit tests for the onboarding wizard API (/api/onboarding).
|
||||
|
||||
Covers:
|
||||
- GET /api/onboarding/status (auth required, new user, returning user)
|
||||
- POST /api/onboarding/profile (saves display_name / contact_email)
|
||||
- POST /api/onboarding/plan (saves tier + billing cycle, rejects invalid tiers)
|
||||
- POST /api/onboarding/storage (saves preferred_destination)
|
||||
- POST /api/onboarding/complete (marks onboarding_completed=True)
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from sqlalchemy.pool import StaticPool
|
||||
|
||||
from app.database import Base, get_db
|
||||
from app.models import UserProfile
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_TEST_USER = {
|
||||
"sub": "user-onb-123",
|
||||
"name": "Test User",
|
||||
"email": "test@example.com",
|
||||
"is_admin": False,
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def ob_engine():
|
||||
"""In-memory SQLite engine for onboarding tests."""
|
||||
engine = create_engine(
|
||||
"sqlite:///:memory:",
|
||||
connect_args={"check_same_thread": False},
|
||||
poolclass=StaticPool,
|
||||
)
|
||||
Base.metadata.create_all(bind=engine)
|
||||
yield engine
|
||||
Base.metadata.drop_all(bind=engine)
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def ob_session(ob_engine):
|
||||
"""DB session scoped to one test."""
|
||||
Session = sessionmaker(bind=ob_engine)
|
||||
session = Session()
|
||||
yield session
|
||||
session.close()
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def ob_client_authed(ob_engine):
|
||||
"""TestClient that bypasses auth by monkey-patching _get_current_user_id."""
|
||||
from app.api import onboarding as ob_module
|
||||
from app.main import app
|
||||
|
||||
original = ob_module._get_current_user_id
|
||||
|
||||
def override_db():
|
||||
Session = sessionmaker(bind=ob_engine)
|
||||
session = Session()
|
||||
try:
|
||||
yield session
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
def fake_user_id(_request):
|
||||
return _TEST_USER["sub"]
|
||||
|
||||
ob_module._get_current_user_id = fake_user_id
|
||||
app.dependency_overrides[get_db] = override_db
|
||||
|
||||
with TestClient(app, base_url="http://localhost", raise_server_exceptions=False) as client:
|
||||
yield client
|
||||
|
||||
ob_module._get_current_user_id = original
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
|
||||
def _make_profile(session, user_id: str, **kwargs) -> UserProfile:
|
||||
"""Insert a UserProfile row with sensible defaults."""
|
||||
kwargs.setdefault("is_blocked", False)
|
||||
kwargs.setdefault("onboarding_completed", False)
|
||||
profile = UserProfile(user_id=user_id, **kwargs)
|
||||
session.add(profile)
|
||||
session.commit()
|
||||
session.refresh(profile)
|
||||
return profile
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# TestOnboardingAPI
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestOnboardingAPI:
|
||||
"""Unit tests for the /api/onboarding endpoints."""
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# GET /api/onboarding/status
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def test_status_requires_auth(self, ob_engine):
|
||||
"""Unauthenticated requests to /status must return 401."""
|
||||
from app.main import app
|
||||
|
||||
def override_db():
|
||||
Session = sessionmaker(bind=ob_engine)
|
||||
session = Session()
|
||||
try:
|
||||
yield session
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
app.dependency_overrides[get_db] = override_db
|
||||
with TestClient(app, base_url="http://localhost", raise_server_exceptions=False) as client:
|
||||
# No session user injected → _get_current_user_id raises 401
|
||||
resp = client.get("/api/onboarding/status")
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
assert resp.status_code == 401
|
||||
|
||||
def test_status_returns_not_completed_for_new_user(self, ob_client_authed, ob_session):
|
||||
"""A user with no profile should get completed=False and step=1."""
|
||||
resp = ob_client_authed.get("/api/onboarding/status")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["completed"] is False
|
||||
assert data["step"] == 1
|
||||
assert data["profile"] is None
|
||||
|
||||
def test_status_returns_not_completed_for_existing_incomplete_profile(self, ob_client_authed, ob_session):
|
||||
"""A user with an existing profile but onboarding_completed=False → completed=False."""
|
||||
_make_profile(ob_session, _TEST_USER["sub"], display_name="Alice", onboarding_completed=False)
|
||||
resp = ob_client_authed.get("/api/onboarding/status")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["completed"] is False
|
||||
assert data["profile"]["display_name"] == "Alice"
|
||||
|
||||
def test_status_returns_completed_when_done(self, ob_client_authed, ob_session):
|
||||
"""A user with onboarding_completed=True → completed=True and step=5."""
|
||||
_make_profile(ob_session, _TEST_USER["sub"], onboarding_completed=True)
|
||||
resp = ob_client_authed.get("/api/onboarding/status")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["completed"] is True
|
||||
assert data["step"] == 5
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# POST /api/onboarding/profile
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def test_save_profile_updates_display_name(self, ob_client_authed, ob_session):
|
||||
"""POST /profile should persist display_name to UserProfile."""
|
||||
resp = ob_client_authed.post(
|
||||
"/api/onboarding/profile",
|
||||
json={"display_name": "Jane Doe", "contact_email": "jane@example.com"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["display_name"] == "Jane Doe"
|
||||
assert data["contact_email"] == "jane@example.com"
|
||||
|
||||
# Verify DB was actually updated
|
||||
profile = ob_session.query(UserProfile).filter(UserProfile.user_id == _TEST_USER["sub"]).first()
|
||||
assert profile is not None
|
||||
assert profile.display_name == "Jane Doe"
|
||||
assert profile.contact_email == "jane@example.com"
|
||||
|
||||
def test_save_profile_requires_auth(self, ob_engine):
|
||||
"""POST /profile without auth should return 401."""
|
||||
from app.main import app
|
||||
|
||||
def override_db():
|
||||
Session = sessionmaker(bind=ob_engine)
|
||||
session = Session()
|
||||
try:
|
||||
yield session
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
app.dependency_overrides[get_db] = override_db
|
||||
with TestClient(app, base_url="http://localhost", raise_server_exceptions=False) as client:
|
||||
resp = client.post("/api/onboarding/profile", json={"display_name": "x"})
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
assert resp.status_code == 401
|
||||
|
||||
def test_save_profile_null_fields_allowed(self, ob_client_authed, ob_session):
|
||||
"""Sending null for display_name and contact_email should succeed."""
|
||||
resp = ob_client_authed.post(
|
||||
"/api/onboarding/profile",
|
||||
json={"display_name": None, "contact_email": None},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# POST /api/onboarding/plan
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def test_save_plan_updates_subscription_tier(self, ob_client_authed, ob_session):
|
||||
"""POST /plan should persist the chosen tier and billing cycle."""
|
||||
resp = ob_client_authed.post(
|
||||
"/api/onboarding/plan",
|
||||
json={"subscription_tier": "starter", "billing_cycle": "monthly"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["subscription_tier"] == "starter"
|
||||
assert data["subscription_billing_cycle"] == "monthly"
|
||||
|
||||
profile = ob_session.query(UserProfile).filter(UserProfile.user_id == _TEST_USER["sub"]).first()
|
||||
assert profile.subscription_tier == "starter"
|
||||
|
||||
def test_save_plan_yearly_billing(self, ob_client_authed, ob_session):
|
||||
"""POST /plan with billing_cycle=yearly should persist correctly."""
|
||||
resp = ob_client_authed.post(
|
||||
"/api/onboarding/plan",
|
||||
json={"subscription_tier": "professional", "billing_cycle": "yearly"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["subscription_billing_cycle"] == "yearly"
|
||||
|
||||
def test_save_plan_rejects_invalid_tier(self, ob_client_authed):
|
||||
"""POST /plan with an unknown tier should return 422."""
|
||||
resp = ob_client_authed.post(
|
||||
"/api/onboarding/plan",
|
||||
json={"subscription_tier": "unicorn", "billing_cycle": "monthly"},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
assert "Invalid subscription_tier" in resp.json()["detail"]
|
||||
|
||||
def test_save_plan_rejects_invalid_billing_cycle(self, ob_client_authed):
|
||||
"""POST /plan with an invalid billing_cycle value should return 422."""
|
||||
resp = ob_client_authed.post(
|
||||
"/api/onboarding/plan",
|
||||
json={"subscription_tier": "free", "billing_cycle": "weekly"},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# POST /api/onboarding/storage
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def test_save_storage_updates_preferred_destination(self, ob_client_authed, ob_session):
|
||||
"""POST /storage should persist the preferred_destination."""
|
||||
resp = ob_client_authed.post(
|
||||
"/api/onboarding/storage",
|
||||
json={"preferred_destination": "dropbox"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["preferred_destination"] == "dropbox"
|
||||
|
||||
profile = ob_session.query(UserProfile).filter(UserProfile.user_id == _TEST_USER["sub"]).first()
|
||||
assert profile.preferred_destination == "dropbox"
|
||||
|
||||
def test_save_storage_accepts_null_destination(self, ob_client_authed, ob_session):
|
||||
"""POST /storage with null preferred_destination should succeed (skip storage)."""
|
||||
resp = ob_client_authed.post(
|
||||
"/api/onboarding/storage",
|
||||
json={"preferred_destination": None},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["preferred_destination"] is None
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# POST /api/onboarding/complete
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def test_complete_sets_onboarding_completed(self, ob_client_authed, ob_session):
|
||||
"""POST /complete should set onboarding_completed=True on the profile."""
|
||||
_make_profile(ob_session, _TEST_USER["sub"])
|
||||
resp = ob_client_authed.post("/api/onboarding/complete")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["success"] is True
|
||||
assert "redirect_url" in data
|
||||
|
||||
# Re-query to see persisted value
|
||||
ob_session.expire_all()
|
||||
profile = ob_session.query(UserProfile).filter(UserProfile.user_id == _TEST_USER["sub"]).first()
|
||||
assert profile.onboarding_completed is True
|
||||
assert profile.onboarding_completed_at is not None
|
||||
|
||||
def test_complete_creates_profile_if_missing(self, ob_client_authed, ob_session):
|
||||
"""POST /complete should create a profile when none exists and mark it done."""
|
||||
resp = ob_client_authed.post("/api/onboarding/complete")
|
||||
assert resp.status_code == 200
|
||||
|
||||
profile = ob_session.query(UserProfile).filter(UserProfile.user_id == _TEST_USER["sub"]).first()
|
||||
assert profile is not None
|
||||
assert profile.onboarding_completed is True
|
||||
|
||||
def test_complete_requires_auth(self, ob_engine):
|
||||
"""POST /complete without auth should return 401."""
|
||||
from app.main import app
|
||||
|
||||
def override_db():
|
||||
Session = sessionmaker(bind=ob_engine)
|
||||
session = Session()
|
||||
try:
|
||||
yield session
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
app.dependency_overrides[get_db] = override_db
|
||||
with TestClient(app, base_url="http://localhost", raise_server_exceptions=False) as client:
|
||||
resp = client.post("/api/onboarding/complete")
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
assert resp.status_code == 401
|
||||
Reference in New Issue
Block a user