Tests for personal and work email addresses (#2110)

This commit adds backend support for testing if an account has work and personal email addresses associated with it. UI changes are not included.

funnel/models/email_address.py

@@ -5,11 +5,13 @@
 import hashlib
 import unicodedata
 import warnings
+from collections.abc import MutableMapping
 from datetime import datetime
 from typing import TYPE_CHECKING, Any, ClassVar, Literal, Self, cast, overload
 
 import base58
 import idna
+from mxsniff import mxsniff
 from pyisemail import is_email
 from pyisemail.diagnosis import BaseDiagnosis
 from sqlalchemy import event, inspect
@@ -408,6 +410,19 @@ def is_available_for(self, owner: Account | None) -> bool:
                     return False
         return True
 
+    def is_public_provider(self, cache: MutableMapping | None = None) -> bool:
+        """
+        Check if this email address is from a known public provider (Gmail, etc).
+
+        This performs a DNS lookup for unknown domains. Providing a cache is highly
+        recommended. The cache object must implement the
+        :class:`~collections.abc.MutableMapping` API.
+        """
+        if not self.domain:
+            raise ValueError("Can't lookup a removed email address.")
+        lookup = mxsniff(self.domain, cache=cache)
+        return lookup['public']
+
     @delivery_state.transition(None, delivery_state.SENT)
     def mark_sent(self) -> None:
         """Record fact of an email message being sent to this address."""

funnel/utils/__init__.py

@@ -4,7 +4,8 @@
 
 # MARK: Everything below this line is auto-generated using `make initpy` ---------------
 
-from . import jinja_template, markdown, misc, mustache
+from . import cache, jinja_template, markdown, misc, mustache
+from .cache import DictCache
 from .jinja_template import JinjaTemplateBase, jinja_global, jinja_undefined
 from .markdown import (
     MarkdownConfig,
@@ -32,13 +33,15 @@
 from .mustache import mustache_html, mustache_md
 
 __all__ = [
+    "DictCache",
     "JinjaTemplateBase",
     "MarkdownConfig",
     "MarkdownPlugin",
     "MarkdownString",
     "TIMEDELTA_1DAY",
     "abort_null",
     "blake2b160_hex",
+    "cache",
     "extract_twitter_handle",
     "format_twitter_handle",
     "jinja_global",

funnel/utils/cache.py

@@ -0,0 +1,70 @@
+"""Cache wrapper."""
+
+from collections.abc import Iterator, MutableMapping
+from typing import Any
+
+from cachelib import BaseCache
+from flask_caching import Cache as CacheExtension
+
+__all__ = ['DictCache']
+
+_marker = object()
+
+
+class DictCache(MutableMapping):
+    """
+    Provide a dict-like interface to a Cachelib cache.
+
+    This object has three significant differences from regular dicts:
+
+    1. Since a cache can't be enumerated, this object will behave like an empty dict.
+    2. However, this object is always truthy despite appearing to be empty.
+    3. `None` is a special value indicating a cache miss and can't be used as a value.
+
+    :param cache: Flask-Caching cache backend to wrap
+    :param prefix: Prefix string to apply to all keys
+    :param timeout: Timeout when setting a value
+    """
+
+    def __init__(
+        self,
+        cache: CacheExtension | BaseCache,
+        prefix: str = '',
+        timeout: int | None = None,
+    ) -> None:
+        self.cache = cache
+        self.prefix = prefix
+        self.timeout = timeout
+
+    def __getitem__(self, key: str) -> Any:
+        result = self.cache.get(self.prefix + key)
+        if result is not None:
+            return result
+        raise KeyError(key)
+
+    def __setitem__(self, key: str, value: Any) -> None:
+        success = self.cache.set(self.prefix + key, value, timeout=self.timeout)
+        if not success:
+            raise KeyError(key)
+
+    def __delitem__(self, key: str) -> None:
+        success = self.cache.delete(self.prefix + key)
+        if not success:
+            raise KeyError(key)
+
+    def __contains__(self, key: Any) -> bool:
+        return self.cache.has(key)
+
+    # Dummy implementations for compatibility with MutableMapping:
+
+    def __iter__(self) -> Iterator:
+        """Return an empty iterable since the cache's contents can't be enumerated."""
+        return iter(())
+
+    def __len__(self) -> int:
+        """Return 0 since the cache's size is not queryable."""
+        return 0
+
+    def __bool__(self) -> bool:
+        """Return True since the cache can't be iterated and is assumed non-empty."""
+        return True

funnel/views/account.py

@@ -20,7 +20,7 @@
 )
 from markupsafe import Markup, escape
 
-from baseframe import _, forms
+from baseframe import _, cache, forms
 from baseframe.forms import render_delete_sqla, render_form, render_message
 from coaster.sqlalchemy import RoleAccessProxy
 from coaster.views import ClassView, get_next_url, render_with, route
@@ -61,6 +61,7 @@
 from ..registry import login_registry
 from ..signals import user_data_changed
 from ..typing import ReturnRenderWith, ReturnResponse, ReturnView
+from ..utils import DictCache
 from .decorators import etag_cache_for_user, xhr_only
 from .email import send_email_verify_link
 from .helpers import (
@@ -81,6 +82,8 @@
 from .notification import dispatch_notification
 from .otp import OtpSession, OtpTimeoutError
 
+email_cache = DictCache(cache, 'email/', 86400)
+
 # MARK: View registry ------------------------------------------------------------------
 
 
@@ -211,6 +214,52 @@ def user_not_likely_throwaway(obj: Account) -> bool:
     return obj.is_verified or bool(obj.phone)
 
 
+@Account.features('has_work_email', cached_property=True)
+def account_has_work_email(obj: Account) -> bool:
+    """Confirm the user has a work email address associated with their account."""
+    if not obj.emails:
+        return False
+    return any(
+        not ae.email_address.is_public_provider(email_cache)
+        for ae in obj.emails
+        if ae.email_address.domain is not None
+    )
+
+
+@Account.features('has_personal_email', cached_property=True)
+def account_has_personal_email(obj: Account) -> bool:
+    """Confirm the user has a personal email address associated with their account."""
+    if not obj.emails:
+        return False
+    return any(
+        ae.email_address.is_public_provider(email_cache)
+        for ae in obj.emails
+        if ae.email_address.domain is not None
+    )
+
+
+@Account.features('may_need_to_add_email', cached_property=True)
+def account_may_need_to_add_email(obj: Account) -> bool:
+    """Check if the user missing work or personal email addresses."""
+    if not obj.emails:
+        return True
+    has_work_email = False
+    has_personal_email = False
+    for ae in obj.emails:
+        if ae.email_address.domain is None:
+            continue
+        is_public = ae.email_address.is_public_provider(email_cache)
+        if is_public:
+            has_personal_email = True
+            if has_work_email:
+                return False
+        else:
+            has_work_email = True
+            if has_personal_email:
+                return False
+    return True
+
+
 _quoted_str_re = re.compile('"(.*?)"')
 _quoted_ua_re = re.compile(r'"(.*?)"\s*;\s*v\s*=\s*"(.*?)",?\s*')
 _fake_ua_re = re.compile(

requirements/base.in

@@ -7,6 +7,7 @@ base58
 bcrypt
 better_profanity!=0.7.0  # https://github.com/snguyenthanh/better_profanity/issues/19
 Brotli
+cachelib
 chevron
 click
 cryptography
@@ -37,6 +38,7 @@ lazy-loader
 linkify-it-py
 markdown-it-py
 mdit-py-plugins
+mxsniff
 oauth2client
 passlib
 phonenumbers

requirements/base.txt

@@ -1,4 +1,4 @@
-# SHA1:5f90855f3a570ce53e6e6a8535075ec897910b30
+# SHA1:6b1a6817110b5a911f195a7fd4a1b4b5ea8a8b51
 #
 # This file is autogenerated by pip-compile-multi
 # To update, run:
@@ -65,7 +65,9 @@ blinker==1.8.2
 brotli==1.1.0
     # via -r requirements/base.in
 cachelib==0.9.0
-    # via flask-caching
+    # via
+    #   -r requirements/base.in
+    #   flask-caching
 cachetools==5.5.0
     # via premailer
 certifi==2024.7.4
@@ -271,7 +273,9 @@ multidict==6.0.5
     #   aiohttp
     #   yarl
 mxsniff==0.3.5
-    # via baseframe
+    # via
+    #   -r requirements/base.in
+    #   baseframe
 mypy-extensions==1.0.0
     # via typing-inspect
 oauth2client==4.1.3

tests/unit/utils/cache_test.py

@@ -0,0 +1,49 @@
+"""Tests for cache wrapper."""
+# pylint: disable=redefined-outer-name,pointless-statement
+
+import pytest
+from cachelib import SimpleCache
+
+from funnel.utils import DictCache
+
+
+@pytest.fixture
+def cache() -> SimpleCache:
+    return SimpleCache()
+
+
+@pytest.fixture
+def dict_cache(cache) -> DictCache:
+    return DictCache(cache)
+
+
+def test_cache_interface(cache: SimpleCache, dict_cache: DictCache) -> None:
+    """Test dict interface to cachelib cache."""
+    assert not cache.has('test-key1')
+    assert 'test-key1' not in dict_cache
+    cache.set('test-key1', 'value1')
+    assert cache.has('test-key1')
+    assert 'test-key1' in dict_cache
+    assert dict_cache['test-key1'] == 'value1'
+
+    assert not cache.has('test-key2')
+    assert 'test-key2' not in dict_cache
+    dict_cache['test-key2'] = 'value2'
+    assert 'test-key2' in dict_cache
+    assert cache.has('test-key2')
+    assert dict_cache['test-key2'] == 'value2'
+
+    del dict_cache['test-key2']
+    assert 'test_key2' not in dict_cache
+    assert not cache.has('test-key2')
+
+    with pytest.raises(KeyError):
+        del dict_cache['test-key2']
+
+    with pytest.raises(KeyError):
+        dict_cache['test-key2']
+
+    # Cache is not enumerable
+    assert len(dict_cache) == 0
+    assert not list(dict_cache)
+    assert bool(dict_cache) is True

tests/unit/views/account_test.py

@@ -72,6 +72,9 @@ def test_username_available(
     }
 
 
+# MARK: User agent details
+
+
 @pytest.mark.parametrize(
     ('user_agent', 'client_hints', 'output'),
     [
@@ -327,3 +330,68 @@ def test_user_agent_details(
         )
         == output
     )
+
+
+# MARK: Work and personal email
+
+
+def test_has_work_email_empty(user_twoflower: models.User) -> None:
+    """Test for work email addresses on an account (when user has no emails)."""
+    assert user_twoflower.features.has_work_email is False
+
+
+def test_has_personal_email_empty(user_twoflower: models.User) -> None:
+    """Test for personal email addresses on an account (when user has no emails)."""
+    assert user_twoflower.features.has_personal_email is False
+
+
+def test_may_need_email_empty(user_twoflower: models.User) -> None:
+    """Test if user needs to add any email addresses (when having none)."""
+    assert user_twoflower.features.may_need_to_add_email is True
+
+
+@pytest.fixture
+def work_email(user_twoflower: models.User) -> models.AccountEmail:
+    """Provide a (test) work email for the user."""
+    return user_twoflower.add_email('[email protected]')
+
+
+@pytest.fixture
+def personal_email(user_twoflower: models.User) -> models.AccountEmail:
+    """Provide a (test) personal email for the user."""
+    return user_twoflower.add_email('[email protected]')
+
+
+@pytest.mark.usefixtures('personal_email')
+@pytest.mark.enable_socket
+def test_has_work_email_no(user_twoflower) -> None:
+    """Confirm user is missing a work email."""
+    assert user_twoflower.features.has_work_email is False
+
+
+@pytest.mark.usefixtures('work_email')
+@pytest.mark.enable_socket
+def test_has_work_email_yes(user_twoflower: models.User) -> None:
+    """Confirm user has a work email."""
+    assert user_twoflower.features.has_work_email is True
+
+
+@pytest.mark.usefixtures('personal_email')
+@pytest.mark.enable_socket
+def test_has_personal_not_work_email(user_twoflower) -> None:
+    """Confirm user has both work and personal email addresses."""
+    assert user_twoflower.features.may_need_to_add_email is True
+
+
+@pytest.mark.usefixtures('work_email')
+@pytest.mark.enable_socket
+def test_has_work_not_personal_email(user_twoflower) -> None:
+    """Confirm user has both work and personal email addresses."""
+    assert user_twoflower.features.may_need_to_add_email is True
+
+
+@pytest.mark.usefixtures('work_email', 'personal_email')
+@pytest.mark.enable_socket
+def test_has_both_work_and_personal_email(user_twoflower) -> None:
+    """Confirm user has both work and personal email addresses."""
+    assert user_twoflower.features.may_need_to_add_email is False