chore: merge main into branch, resolve conflict in models.py
Both sets of UserProfile columns are retained: - is_complimentary (from main, migration 019_add_is_complimentary) - subscription_change_pending_tier / subscription_change_pending_date (our branch, renamed to migration 020_add_subscription_change_pending with down_revision updated to chain after 019_add_is_complimentary)
This commit is contained in:
copilot-swe-agent[bot]
@@ -477,3 +477,189 @@ class TestUserProfileModel:
|
||||
with pytest.raises(IntegrityError):
|
||||
au_session.commit()
|
||||
au_session.rollback()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Complimentary plan tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestComplimentaryPlan:
|
||||
"""Tests for the is_complimentary field and admin auto-creation logic."""
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_create_profile_with_complimentary_flag(self, au_client, au_session):
|
||||
"""PUT can create a profile with is_complimentary=True."""
|
||||
resp = au_client.put(
|
||||
"/api/admin/users/comp@example.com",
|
||||
json={"subscription_tier": "business", "is_complimentary": True, "is_blocked": False},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["is_complimentary"] is True
|
||||
assert data["subscription_tier"] == "business"
|
||||
|
||||
profile = au_session.query(UserProfile).filter_by(user_id="comp@example.com").first()
|
||||
assert profile is not None
|
||||
assert profile.is_complimentary is True
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_update_profile_set_complimentary(self, au_client, au_session):
|
||||
"""PUT can toggle is_complimentary on an existing profile."""
|
||||
_make_profile(au_session, "toggle@example.com", is_complimentary=False)
|
||||
|
||||
resp = au_client.put(
|
||||
"/api/admin/users/toggle@example.com",
|
||||
json={"is_blocked": False, "is_complimentary": True},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["is_complimentary"] is True
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_list_users_includes_complimentary_field(self, au_client, au_session):
|
||||
"""GET /api/admin/users/ returns is_complimentary per user."""
|
||||
_make_profile(au_session, "complist@example.com", is_complimentary=True)
|
||||
|
||||
resp = au_client.get("/api/admin/users/")
|
||||
assert resp.status_code == 200
|
||||
users = {u["user_id"]: u for u in resp.json()["users"]}
|
||||
assert "complist@example.com" in users
|
||||
assert users["complist@example.com"]["is_complimentary"] is True
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_get_user_includes_complimentary_field(self, au_client, au_session):
|
||||
"""GET /api/admin/users/<id> returns is_complimentary in profile."""
|
||||
_make_profile(au_session, "getcomp@example.com", is_complimentary=True, subscription_tier="business")
|
||||
|
||||
resp = au_client.get("/api/admin/users/getcomp%40example.com")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["is_complimentary"] is True
|
||||
assert data["profile"]["is_complimentary"] is True
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_complimentary_defaults_to_false(self, au_client, au_session):
|
||||
"""Newly created profiles have is_complimentary=False by default."""
|
||||
resp = au_client.put(
|
||||
"/api/admin/users/nocomp@example.com",
|
||||
json={"is_blocked": False},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["is_complimentary"] is False
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_profile_model_complimentary_column(self, au_session):
|
||||
"""UserProfile model stores is_complimentary correctly."""
|
||||
profile = UserProfile(user_id="modelcomp@example.com", is_complimentary=True)
|
||||
au_session.add(profile)
|
||||
au_session.commit()
|
||||
au_session.refresh(profile)
|
||||
assert profile.is_complimentary is True
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _ensure_user_profile admin auto-creation tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestEnsureUserProfileAdmin:
|
||||
"""Tests for _ensure_user_profile admin-specific behaviour."""
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_admin_login_creates_highest_tier_profile(self, au_session):
|
||||
"""Admin first login creates a profile with the highest subscription tier."""
|
||||
from app.auth import _ensure_user_profile
|
||||
from app.utils.subscription import TIER_ORDER
|
||||
|
||||
user_data = {
|
||||
"preferred_username": "admin",
|
||||
"email": "admin@local.docuelevate",
|
||||
"name": "Administrator",
|
||||
"is_admin": True,
|
||||
}
|
||||
_ensure_user_profile(au_session, user_data, is_admin=True)
|
||||
|
||||
# user_id uses preferred_username (sub not provided)
|
||||
profile = au_session.query(UserProfile).filter_by(user_id="admin").first()
|
||||
assert profile is not None
|
||||
assert profile.subscription_tier == TIER_ORDER[-1]
|
||||
assert profile.is_complimentary is True
|
||||
assert profile.onboarding_completed is True
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_regular_user_login_creates_free_profile(self, au_session):
|
||||
"""Regular user login creates a profile with the free tier."""
|
||||
from app.auth import _ensure_user_profile
|
||||
|
||||
user_data = {
|
||||
"preferred_username": "regular",
|
||||
"email": "user@example.com",
|
||||
"name": "Regular User",
|
||||
}
|
||||
_ensure_user_profile(au_session, user_data, is_admin=False)
|
||||
|
||||
# user_id uses preferred_username (sub not provided)
|
||||
profile = au_session.query(UserProfile).filter_by(user_id="regular").first()
|
||||
assert profile is not None
|
||||
assert profile.subscription_tier == "free"
|
||||
assert profile.is_complimentary is False
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_admin_login_sets_complimentary_on_existing_profile(self, au_session):
|
||||
"""Existing admin profile gets is_complimentary=True on login."""
|
||||
existing = UserProfile(user_id="existadmin", is_complimentary=False, subscription_tier="starter")
|
||||
au_session.add(existing)
|
||||
au_session.commit()
|
||||
|
||||
from app.auth import _ensure_user_profile
|
||||
|
||||
user_data = {"preferred_username": "existadmin", "email": "ea@example.com"}
|
||||
_ensure_user_profile(au_session, user_data, is_admin=True)
|
||||
|
||||
au_session.refresh(existing)
|
||||
assert existing.is_complimentary is True
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_admin_login_does_not_downgrade_existing_tier(self, au_session):
|
||||
"""Existing admin profile with the highest tier keeps that tier on re-login."""
|
||||
from app.auth import _ensure_user_profile
|
||||
from app.utils.subscription import TIER_ORDER
|
||||
|
||||
highest = TIER_ORDER[-1]
|
||||
existing = UserProfile(user_id="toptieradmin", is_complimentary=False, subscription_tier=highest)
|
||||
au_session.add(existing)
|
||||
au_session.commit()
|
||||
|
||||
user_data = {"preferred_username": "toptieradmin", "email": "tt@example.com"}
|
||||
_ensure_user_profile(au_session, user_data, is_admin=True)
|
||||
|
||||
au_session.refresh(existing)
|
||||
assert existing.subscription_tier == highest
|
||||
assert existing.is_complimentary is True
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_admin_login_upgrades_free_tier_on_existing_profile(self, au_session):
|
||||
"""Existing admin profile on free tier gets upgraded to highest tier."""
|
||||
from app.auth import _ensure_user_profile
|
||||
from app.utils.subscription import TIER_ORDER
|
||||
|
||||
existing = UserProfile(user_id="freeadmin", is_complimentary=False, subscription_tier="free")
|
||||
au_session.add(existing)
|
||||
au_session.commit()
|
||||
|
||||
user_data = {"preferred_username": "freeadmin", "email": "fa@example.com"}
|
||||
_ensure_user_profile(au_session, user_data, is_admin=True)
|
||||
|
||||
au_session.refresh(existing)
|
||||
assert existing.subscription_tier == TIER_ORDER[-1]
|
||||
assert existing.is_complimentary is True
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_ensure_user_profile_no_identifier_logs_warning(self, au_session):
|
||||
"""_ensure_user_profile logs a warning when no stable user id is present."""
|
||||
from app.auth import _ensure_user_profile
|
||||
|
||||
_ensure_user_profile(au_session, {}, is_admin=False)
|
||||
# No profile should have been created
|
||||
count = au_session.query(UserProfile).count()
|
||||
assert count == 0
|
||||
|
||||
@@ -278,8 +278,10 @@ class TestAuthEndpoint:
|
||||
mock_form_data = {"username": "admin", "password": "secret123"}
|
||||
mock_request.form = AsyncMock(return_value=mock_form_data)
|
||||
mock_request.session = {}
|
||||
mock_db = MagicMock()
|
||||
|
||||
result = await auth(mock_request)
|
||||
with patch("app.auth._ensure_user_profile"):
|
||||
result = await auth(mock_request, db=mock_db)
|
||||
|
||||
# Verify redirect to upload page
|
||||
assert isinstance(result, RedirectResponse)
|
||||
@@ -305,8 +307,9 @@ class TestAuthEndpoint:
|
||||
mock_form_data = {"username": "admin", "password": "wrong_password"}
|
||||
mock_request.form = AsyncMock(return_value=mock_form_data)
|
||||
mock_request.session = {}
|
||||
mock_db = MagicMock()
|
||||
|
||||
result = await auth(mock_request)
|
||||
result = await auth(mock_request, db=mock_db)
|
||||
|
||||
# Verify redirect to login with error
|
||||
assert isinstance(result, RedirectResponse)
|
||||
@@ -327,8 +330,10 @@ class TestAuthEndpoint:
|
||||
mock_form_data = {"username": "admin", "password": "secret123"}
|
||||
mock_request.form = AsyncMock(return_value=mock_form_data)
|
||||
mock_request.session = {"redirect_after_login": "/protected/page"}
|
||||
mock_db = MagicMock()
|
||||
|
||||
result = await auth(mock_request)
|
||||
with patch("app.auth._ensure_user_profile"):
|
||||
result = await auth(mock_request, db=mock_db)
|
||||
|
||||
# Verify redirect to saved URL
|
||||
assert isinstance(result, RedirectResponse)
|
||||
|
||||
+128
-2
@@ -208,7 +208,7 @@ def test_signup_disabled(la_client):
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_signup_smtp_not_configured(la_client):
|
||||
"""POST /api/auth/signup returns 503 when SMTP is not configured."""
|
||||
"""POST /api/auth/signup succeeds without SMTP and activates the account immediately."""
|
||||
with patch("app.api.local_auth.settings") as mock_settings:
|
||||
mock_settings.allow_local_signup = True
|
||||
mock_settings.multi_user_enabled = True
|
||||
@@ -222,7 +222,10 @@ def test_signup_smtp_not_configured(la_client):
|
||||
"password_confirm": "password1",
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 503
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert data["email_verification_required"] is False
|
||||
assert "now log in" in data["message"]
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@@ -266,6 +269,7 @@ def test_signup_success(la_client):
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
assert "Verification email sent" in resp.json()["message"]
|
||||
assert resp.json()["email_verification_required"] is True
|
||||
mock_send.assert_called_once()
|
||||
|
||||
|
||||
@@ -675,3 +679,125 @@ async def test_single_user_mode_skips_local_user_table(la_session, active_user):
|
||||
# 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"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Integration tests: admin local user management
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def admin_session_client(la_engine):
|
||||
"""TestClient with admin access via dependency override."""
|
||||
from app.api.admin_users import _require_admin
|
||||
from app.main import app
|
||||
|
||||
Session = sessionmaker(bind=la_engine)
|
||||
|
||||
def override_get_db():
|
||||
db = Session()
|
||||
try:
|
||||
yield db
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
def override_require_admin():
|
||||
return {"id": "admin@example.com", "is_admin": True, "display_name": "Admin"}
|
||||
|
||||
app.dependency_overrides[get_db] = override_get_db
|
||||
app.dependency_overrides[_require_admin] = override_require_admin
|
||||
with TestClient(app, base_url="http://localhost", raise_server_exceptions=True) as client:
|
||||
yield client
|
||||
app.dependency_overrides.pop(get_db, None)
|
||||
app.dependency_overrides.pop(_require_admin, None)
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_admin_list_local_users_empty(admin_session_client):
|
||||
"""GET /api/admin/users/local returns an empty list when no local users exist."""
|
||||
resp = admin_session_client.get("/api/admin/users/local")
|
||||
assert resp.status_code == 200
|
||||
assert resp.json() == []
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_admin_create_local_user(admin_session_client, la_session):
|
||||
"""POST /api/admin/users/local creates a new active local user."""
|
||||
resp = admin_session_client.post(
|
||||
"/api/admin/users/local",
|
||||
json={
|
||||
"email": "newuser@example.com",
|
||||
"username": "newuser",
|
||||
"password": "password1",
|
||||
"is_admin": False,
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert data["email"] == "newuser@example.com"
|
||||
assert data["username"] == "newuser"
|
||||
assert data["is_active"] is True
|
||||
assert data["is_admin"] is False
|
||||
|
||||
user = la_session.query(LocalUser).filter(LocalUser.email == "newuser@example.com").first()
|
||||
assert user is not None
|
||||
assert user.is_active is True
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_admin_create_local_user_duplicate_email(admin_session_client, active_user):
|
||||
"""POST /api/admin/users/local returns 409 when email already exists."""
|
||||
resp = admin_session_client.post(
|
||||
"/api/admin/users/local",
|
||||
json={
|
||||
"email": "active@example.com",
|
||||
"username": "differentuser",
|
||||
"password": "password1",
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 409
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_admin_create_local_user_duplicate_username(admin_session_client, active_user):
|
||||
"""POST /api/admin/users/local returns 409 when username already taken."""
|
||||
resp = admin_session_client.post(
|
||||
"/api/admin/users/local",
|
||||
json={
|
||||
"email": "different@example.com",
|
||||
"username": "activeuser",
|
||||
"password": "password1",
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 409
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_admin_delete_local_user(admin_session_client, la_session, active_user):
|
||||
"""DELETE /api/admin/users/local/{id} removes the account."""
|
||||
user_id = active_user.id
|
||||
resp = admin_session_client.delete(f"/api/admin/users/local/{user_id}")
|
||||
assert resp.status_code == 204
|
||||
|
||||
user = la_session.query(LocalUser).filter(LocalUser.id == user_id).first()
|
||||
assert user is None
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_admin_delete_local_user_not_found(admin_session_client):
|
||||
"""DELETE /api/admin/users/local/{id} returns 404 for unknown ID."""
|
||||
resp = admin_session_client.delete("/api/admin/users/local/99999")
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_admin_local_user_list_after_create(admin_session_client):
|
||||
"""GET /api/admin/users/local returns the created user."""
|
||||
admin_session_client.post(
|
||||
"/api/admin/users/local",
|
||||
json={"email": "listed@example.com", "username": "listeduser", "password": "password1"},
|
||||
)
|
||||
resp = admin_session_client.get("/api/admin/users/local")
|
||||
assert resp.status_code == 200
|
||||
users = resp.json()
|
||||
assert any(u["email"] == "listed@example.com" for u in users)
|
||||
|
||||
@@ -107,7 +107,7 @@ def test_free_tier_has_no_mailboxes():
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_business_tier_has_highest_limits():
|
||||
"""Business tier must have the highest limits of all paid tiers."""
|
||||
"""Power tier (plan_id 'business') must have the highest limits of all paid tiers."""
|
||||
t = TIERS["business"]
|
||||
# lifetime: no hard cap (0 = unlimited)
|
||||
assert t["lifetime_file_limit"] == 0
|
||||
@@ -121,9 +121,15 @@ def test_business_tier_has_highest_limits():
|
||||
assert t["max_file_size_mb"] == 0
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_business_tier_display_name_is_power():
|
||||
"""The 'business' plan_id must display as 'Power'."""
|
||||
assert TIERS["business"]["name"] == "Power"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_mailbox_limits_increase_by_tier():
|
||||
"""Mailbox limits must increase across tiers: free=0, starter=1, professional=3, business=0(inf)."""
|
||||
"""Mailbox limits must increase across tiers: free=0, starter=1, professional=3, power/business=0(inf)."""
|
||||
assert TIERS["free"]["max_mailboxes"] == 0
|
||||
assert TIERS["starter"]["max_mailboxes"] == 1
|
||||
assert TIERS["professional"]["max_mailboxes"] == 3
|
||||
@@ -144,7 +150,7 @@ def test_free_tier_has_no_trial():
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_pricing_order():
|
||||
"""Paid tier prices must increase in order: starter < professional < business."""
|
||||
"""Paid tier prices must increase in order: starter < professional < power."""
|
||||
assert TIERS["starter"]["price_monthly"] < TIERS["professional"]["price_monthly"]
|
||||
assert TIERS["professional"]["price_monthly"] < TIERS["business"]["price_monthly"]
|
||||
|
||||
|
||||
Reference in New Issue
Block a user