diff --git a/app/admin/base.py b/app/admin/base.py index 594e563c2..cb2ef2691 100644 --- a/app/admin/base.py +++ b/app/admin/base.py @@ -6,8 +6,11 @@ from flask import redirect, url_for, request, flash, session, Response from flask_admin import expose, AdminIndexView, BaseView from flask_admin.contrib import sqla +from flask_admin.contrib.sqla import tools from flask_login import current_user from markupsafe import Markup +from sqlalchemy import Unicode, or_ +from sqlalchemy.sql.expression import cast from time import time from app import config @@ -30,6 +33,24 @@ def _has_valid_admin_time() -> bool: return True +_MAX_BIGINT = 2**63 - 1 + + +def _term_as_int(term: str) -> Optional[int]: + """Return the term as an int if it is a plain positive integer, else None. + + flask-admin's `=` (exact match) prefix is accepted, as an exact match on a + number is exactly what the integer lookup does. + """ + digits = term[1:] if term.startswith("=") else term + if not (digits.isascii() and digits.isdigit()): + return None + value = int(digits) + if value > _MAX_BIGINT: + return None + return value + + def _admin_action_formatter(view, context, model, name): action_name = AuditLogActionEnum.get_name(model.action) return "{} ({})".format(action_name, model.action) @@ -81,6 +102,73 @@ class SLModelView(sqla.ModelView, BaseAdminView): can_delete = False edit_modal = True + def _apply_search(self, query, count_query, joins, count_joins, search): + """Numeric-aware version of flask-admin's search. + + flask-admin casts *every* searchable column to text and matches it with + ILIKE '%term%'. On the user list that means searching an id runs + `CAST(users.id AS VARCHAR) ILIKE '%123%' OR CAST(users.email AS VARCHAR) + ILIKE '%123%'`, so both the list and the count query sequentially scan + the whole table. + + Instead, when a term is a plain integer we compare the integer columns + directly (indexed lookup) and skip the text columns for that term; when + it is not, we skip the integer columns, which can never match by + equality. So a term is either an id lookup or a text search, never + both. To search text columns for a number, use flask-admin's `^` + (starts with) prefix. + """ + for term in search.split(" "): + if not term: + continue + + term_as_int = _term_as_int(term) + stmt = tools.parse_like_term(term) + + int_filter, int_count_filter = [], [] + text_filter, text_count_filter = [], [] + + for field, path in self._search_fields: + query, joins, alias = self._apply_path_joins( + query, joins, path, inner_join=False + ) + + count_alias = None + if count_query is not None: + count_query, count_joins, count_alias = self._apply_path_joins( + count_query, count_joins, path, inner_join=False + ) + + column = field if alias is None else getattr(alias, field.key) + count_column = ( + field if count_alias is None else getattr(count_alias, field.key) + ) + + # hybrid properties have no type, treat them as text + if isinstance(getattr(field, "type", None), sqlalchemy.Integer): + if term_as_int is not None: + int_filter.append(column == term_as_int) + int_count_filter.append(count_column == term_as_int) + else: + text_filter.append(cast(column, Unicode).ilike(stmt)) + text_count_filter.append(cast(count_column, Unicode).ilike(stmt)) + + # Prefer the integer columns, and only fall back to the text ones + # if the term produced no integer clause. If the term produced no + # clause at all - a non-numeric term on a view that only has + # integer searchable columns - nothing can match, and the filter + # must say so rather than be dropped. + filter_stmt = int_filter or text_filter or [sqlalchemy.false()] + count_filter_stmt = ( + int_count_filter or text_count_filter or [sqlalchemy.false()] + ) + + query = query.filter(or_(*filter_stmt)) + if count_query is not None: + count_query = count_query.filter(or_(*count_filter_stmt)) + + return query, count_query, joins, count_joins + def on_model_change(self, form, model, is_created): changes = {} for attr in sqlalchemy.inspect(model).attrs: diff --git a/app/admin/email_search.py b/app/admin/email_search.py index 9e88c66b6..f6beddc4e 100644 --- a/app/admin/email_search.py +++ b/app/admin/email_search.py @@ -30,6 +30,7 @@ Contact, EmailLog, Fido, + Subscription, ) from app.alias_audit_log_utils import emit_alias_audit_log, AliasAuditLogAction from app.alias_delete import delete_alias as perform_alias_delete @@ -331,6 +332,9 @@ class EmailSearchHelpers: PAGE_SIZE = 25 ALIAS_DISPLAY_LIMIT = 5000 UNPAGINATED_QUERY_LIMIT = 100 + PADDLE_SUBSCRIPTION_URL = ( + "https://vendors.paddle.com/subscriptions/customers/manage/{}" + ) @staticmethod def mailbox_list( @@ -439,6 +443,21 @@ def alias_total_pages(user: User) -> int: def partner_user(user: User) -> Optional[PartnerUser]: return PartnerUser.get_by(user_id=user.id) + @staticmethod + def paddle_subscription(user: User) -> Optional[Subscription]: + """Return the user's Paddle subscription, even if cancelled or expired. + + Unlike User.get_paddle_subscription(), past subscriptions are returned + too, so admins can always reach the subscription in Paddle. + """ + return Subscription.get_by(user_id=user.id) + + @staticmethod + def paddle_subscription_url(subscription: Subscription) -> str: + return EmailSearchHelpers.PADDLE_SUBSCRIPTION_URL.format( + subscription.subscription_id + ) + @staticmethod def user_audit_log(user: User) -> list[UserAuditLog]: return ( diff --git a/templates/admin/email_search.html b/templates/admin/email_search.html index d44d14952..d6e134034 100644 --- a/templates/admin/email_search.html +++ b/templates/admin/email_search.html @@ -28,6 +28,7 @@ {% set pu = helper.partner_user(user) %} {% set active_sub = user.get_active_subscription() %} {% set sub_end = user.get_active_subscription_end() %} + {% set paddle_sub = helper.paddle_subscription(user) %}
{# Account Information - 3/4 width #}
@@ -150,6 +151,30 @@
Subscription
{% endif %} + {% if paddle_sub %} + + + Paddle: + + {{ paddle_sub.subscription_id }} + {{ copy_btn(paddle_sub.subscription_id) }} + {% if paddle_sub.cancelled %}Cancelled{% endif %} +
+ + {{ paddle_sub.plan_name() }}, + {% if paddle_sub.cancelled or paddle_sub.next_bill_date < now.date() %} + + last bill + {% else %} + next bill + {% endif %} + {{ paddle_sub.next_bill_date }} + + + + {% endif %} Paid: {{ badge_status(user.is_paid() , 'Yes', 'No', 'info', 'secondary') }} diff --git a/tests/admin/test_email_search.py b/tests/admin/test_email_search.py index fc0c2a634..5ae295cfe 100644 --- a/tests/admin/test_email_search.py +++ b/tests/admin/test_email_search.py @@ -15,7 +15,10 @@ Fido, AdminAuditLog, AuditLogActionEnum, + Subscription, + PlanEnum, ) +from app.admin.email_search import EmailSearchHelpers from app.proton.proton_partner import get_proton_partner from tests.utils import create_new_user, random_token @@ -213,6 +216,82 @@ def test_email_search_user_with_subscription(flask_client): assert b"Subscription" in r.data # Subscription section header +def test_email_search_user_with_paddle_subscription_shows_link(flask_client): + """Test that a Paddle subscription links to the Paddle dashboard.""" + login_admin(flask_client) + + test_user = create_new_user(email=f"paddle_{random_token(8)}@example.com") + subscription_id = random_token(10) + Subscription.create( + user_id=test_user.id, + cancel_url="https://checkout.paddle.com/cancel", + update_url="https://checkout.paddle.com/update", + subscription_id=subscription_id, + event_time=arrow.now(), + next_bill_date=arrow.now().shift(months=1).date(), + plan=PlanEnum.monthly, + flush=True, + ) + Session.commit() + + r = flask_client.get( + url_for("admin.email_search.index"), + query_string={"query": test_user.email, "search_type": "email"}, + ) + assert r.status_code == 200 + expected_url = EmailSearchHelpers.PADDLE_SUBSCRIPTION_URL.format(subscription_id) + assert expected_url.encode() in r.data + assert subscription_id.encode() in r.data + + +def test_email_search_user_without_paddle_subscription_has_no_link(flask_client): + """Test that users without a Paddle subscription get no Paddle link.""" + login_admin(flask_client) + + test_user = create_new_user(email=f"nopaddle_{random_token(8)}@example.com") + Session.commit() + + r = flask_client.get( + url_for("admin.email_search.index"), + query_string={"query": test_user.email, "search_type": "email"}, + ) + assert r.status_code == 200 + assert b"vendors.paddle.com" not in r.data + + +def test_email_search_expired_paddle_subscription_still_shows_link(flask_client): + """An expired Paddle subscription is still reachable from the admin panel.""" + login_admin(flask_client) + + test_user = create_new_user(email=f"oldpaddle_{random_token(8)}@example.com") + subscription_id = random_token(10) + Subscription.create( + user_id=test_user.id, + cancel_url="https://checkout.paddle.com/cancel", + update_url="https://checkout.paddle.com/update", + subscription_id=subscription_id, + event_time=arrow.now().shift(years=-1), + next_bill_date=arrow.now().shift(years=-1).date(), + plan=PlanEnum.yearly, + cancelled=True, + flush=True, + ) + Session.commit() + # the subscription is past its billing date, so it is not the active one + assert test_user.get_paddle_subscription() is None + + r = flask_client.get( + url_for("admin.email_search.index"), + query_string={"query": test_user.email, "search_type": "email"}, + ) + assert r.status_code == 200 + expected_url = EmailSearchHelpers.PADDLE_SUBSCRIPTION_URL.format(subscription_id) + assert expected_url.encode() in r.data + assert b"Cancelled" in r.data + # a subscription that will never bill again isn't labelled "next bill" + assert b"last bill" in r.data + + def test_email_search_user_with_audit_logs(flask_client): """Test that user audit logs are displayed.""" login_admin(flask_client) diff --git a/tests/admin/test_model_view_search.py b/tests/admin/test_model_view_search.py new file mode 100644 index 000000000..c1df71bda --- /dev/null +++ b/tests/admin/test_model_view_search.py @@ -0,0 +1,91 @@ +"""Tests for the numeric-aware search of the admin model views.""" + +from app.admin.alias import AliasAdmin +from app.admin.email_log import EmailLogAdmin +from app.admin.user import UserAdmin +from app.db import Session +from app.models import Alias, EmailLog, User +from tests.utils import create_new_user, random_token + + +def _search_sql(view_cls, model, term: str) -> str: + view = view_cls(model, Session) + query, count_query, _, _ = view._apply_search( + Session.query(model), Session.query(model), {}, {}, term + ) + return str(query) + + +def test_numeric_search_looks_up_the_id_directly(): + sql = _search_sql(UserAdmin, User, "12345") + assert "users.id = " in sql + # no full table scan on the text columns + assert "ILIKE" not in sql.upper() + + +def test_non_numeric_search_only_hits_the_text_columns(): + sql = _search_sql(UserAdmin, User, "bob@example.com") + assert "CAST(users.email AS VARCHAR) ILIKE" in sql + assert "users.id = " not in sql + + +def test_numeric_search_on_joined_id_looks_up_the_id_directly(): + sql = _search_sql(AliasAdmin, Alias, "12345") + assert "alias.id = " in sql + assert "ILIKE" not in sql.upper() + + +def test_exact_match_prefix_still_looks_up_the_id_directly(): + sql = _search_sql(UserAdmin, User, "=12345") + assert "users.id = " in sql + assert "ILIKE" not in sql.upper() + + +def test_search_terms_that_are_not_plain_integers_keep_the_text_search(): + # flask-admin's `^` (starts with) prefix, negative numbers and separators + # are not plain integers + for term in ("^12345", "-12345", "1_2345"): + sql = _search_sql(UserAdmin, User, term) + assert "CAST(users.email AS VARCHAR) ILIKE" in sql, term + assert "users.id = " not in sql, term + + +def test_non_numeric_term_on_an_id_only_view_matches_nothing(): + """A non-numeric term must not be silently dropped, which would list the + whole table. EmailLogAdmin only has an integer searchable column.""" + sql = _search_sql(EmailLogAdmin, EmailLog, "foo@example.com") + assert "WHERE" in sql + view = EmailLogAdmin(EmailLog, Session) + query, count_query, _, _ = view._apply_search( + Session.query(EmailLog), Session.query(EmailLog), {}, {}, "foo@example.com" + ) + assert query.all() == [] + assert count_query.count() == 0 + + +def test_search_ignores_out_of_range_numbers(): + sql = _search_sql(UserAdmin, User, str(2**64)) + assert "users.id = " not in sql + + +def test_numeric_search_finds_the_user(flask_client): + user = create_new_user(email=f"search_{random_token(8)}@example.com") + Session.commit() + + view = UserAdmin(User, Session) + query, count_query, _, _ = view._apply_search( + Session.query(User), Session.query(User), {}, {}, str(user.id) + ) + assert [found.id for found in query.all()] == [user.id] + assert count_query.count() == 1 + + +def test_numeric_search_does_not_match_ids_by_substring(flask_client): + user = create_new_user(email=f"substring_{random_token(8)}@example.com") + Session.commit() + + view = UserAdmin(User, Session) + query, _, _, _ = view._apply_search( + Session.query(User), Session.query(User), {}, {}, str(user.id * 10) + ) + assert user.id not in [found.id for found in query.all()]