diff --git a/migrations/__init__.py b/migrations/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/migrations/add_communication_preferences.py b/migrations/add_communication_preferences.py new file mode 100644 index 0000000..cfe5540 --- /dev/null +++ b/migrations/add_communication_preferences.py @@ -0,0 +1,112 @@ +""" +Add communication preference columns to the user table. + +Required when upgrading from <=0.1.27 to >0.1.27. The comm_opt_in, +comm_updates, and comm_marketing columns were introduced in 0.1.28, and +SQLModel create_all() does not alter existing tables. Run this against any +local or deployed database that predates the columns. + +Usage: + uv run python -m migrations.add_communication_preferences .env + uv run python -m migrations.add_communication_preferences .env --apply +""" + +from __future__ import annotations + +import argparse +from dataclasses import dataclass + +from dotenv import load_dotenv +from sqlalchemy import text +from sqlmodel import Session, create_engine + +from utils.core.db import get_connection_url + +COLUMNS = ("comm_opt_in", "comm_updates", "comm_marketing") + + +@dataclass +class MigrationStats: + missing_columns: tuple[str, ...] = () + all_present: bool = False + + +def _column_exists(session: Session, column_name: str) -> bool: + result = session.connection().execute( + text( + """ + SELECT 1 + FROM information_schema.columns + WHERE table_schema = 'public' + AND table_name = 'user' + AND column_name = :column_name + """ + ), + {"column_name": column_name}, + ) + return result.first() is not None + + +def add_communication_preference_columns(env_file: str, apply: bool) -> MigrationStats: + load_dotenv(env_file, override=True) + engine = create_engine(get_connection_url()) + stats = MigrationStats() + + try: + with Session(engine) as session: + missing = tuple( + column for column in COLUMNS if not _column_exists(session, column) + ) + stats.missing_columns = missing + stats.all_present = not missing + + if apply and missing: + session.connection().execute( + text( + """ + ALTER TABLE "user" + ADD COLUMN IF NOT EXISTS comm_opt_in BOOLEAN NOT NULL DEFAULT FALSE, + ADD COLUMN IF NOT EXISTS comm_updates BOOLEAN NOT NULL DEFAULT FALSE, + ADD COLUMN IF NOT EXISTS comm_marketing BOOLEAN NOT NULL DEFAULT FALSE + """ + ) + ) + session.commit() + else: + session.rollback() + finally: + engine.dispose() + + return stats + + +def main() -> None: + parser = argparse.ArgumentParser( + description=( + "Add comm_opt_in, comm_updates, and comm_marketing to the user table. " + "Without --apply, runs in dry-run mode." + ) + ) + parser.add_argument("env", help="Env file to use (e.g. .env)") + parser.add_argument( + "--apply", + action="store_true", + help="Apply the schema change (default is dry-run).", + ) + args = parser.parse_args() + + stats = add_communication_preference_columns(env_file=args.env, apply=args.apply) + mode = "APPLY" if args.apply else "DRY-RUN" + if stats.all_present: + print(f"[{mode}] All communication preference columns already exist.") + return + + print(f"[{mode}] missing_columns={list(stats.missing_columns)}") + if args.apply: + print(f"[{mode}] Columns added successfully.") + else: + print("Dry-run only. Re-run with --apply to add columns.") + + +if __name__ == "__main__": + main() diff --git a/routers/core/account.py b/routers/core/account.py index 7d93e7c..bef6c09 100644 --- a/routers/core/account.py +++ b/routers/core/account.py @@ -1,4 +1,5 @@ # auth.py +import os from logging import getLogger from typing import Optional, Tuple from urllib.parse import urlparse @@ -75,6 +76,10 @@ login_email_limiter, ) from utils.core.htmx import is_htmx_request, toast_response, set_flash_cookie +from utils.core.communication_preferences import ( + parse_communication_preferences, + apply_communication_preferences, +) logger = getLogger("uvicorn.error") @@ -200,6 +205,7 @@ async def read_register( "password_pattern": HTML_PASSWORD_PATTERN, "email": email, "invitation_token": invitation_token, + "host_name": os.getenv("HOST_NAME", "our platform"), "invitation_token_warning": invitation_token_warning, }, ) @@ -289,6 +295,9 @@ async def register( title="Invitation token", description="Optional invitation token to join an organization", ), + comm_opt_in: Optional[str] = Form(None), + comm_updates: Optional[str] = Form(None), + comm_marketing: Optional[str] = Form(None), ) -> Response: """ Register a new user account, optionally processing an invitation. @@ -330,6 +339,10 @@ async def register( raise DataIntegrityError(resource="Account ID generation") new_user = User(name=name, account_id=account.id) # Use account.id + apply_communication_preferences( + new_user, + parse_communication_preferences(comm_opt_in, comm_updates, comm_marketing), + ) session.add(new_user) # Create the primary AccountEmail entry diff --git a/routers/core/user.py b/routers/core/user.py index fffebfe..cdf3b33 100644 --- a/routers/core/user.py +++ b/routers/core/user.py @@ -4,6 +4,7 @@ from typing import Optional, List from fastapi.templating import Jinja2Templates from sqlalchemy.orm import selectinload +import os from utils.core.models import ( User, UserAvatar, @@ -33,7 +34,11 @@ OrganizationNotFoundError, ) from routers.core.organization import router as organization_router -from utils.core.htmx import is_htmx_request, append_toast +from utils.core.htmx import is_htmx_request, append_toast, toast_response +from utils.core.communication_preferences import ( + parse_communication_preferences, + apply_communication_preferences, +) router = APIRouter(prefix="/user", tags=["user"]) templates = Jinja2Templates(directory="templates") @@ -68,6 +73,7 @@ async def read_profile( "user": user, "account_emails": account_emails, "max_emails": MAX_EMAILS_PER_ACCOUNT, + "host_name": os.getenv("HOST_NAME", "our platform"), }, ) @@ -172,6 +178,31 @@ async def update_profile( return RedirectResponse(url=router.url_path_for("read_profile"), status_code=303) +@router.post("/communication-preferences", response_class=RedirectResponse) +async def update_communication_preferences( + request: Request, + comm_opt_in: Optional[str] = Form(None), + comm_updates: Optional[str] = Form(None), + comm_marketing: Optional[str] = Form(None), + user: User = Depends(get_authenticated_user), + session: Session = Depends(get_session), +) -> Response: + apply_communication_preferences( + user, + parse_communication_preferences(comm_opt_in, comm_updates, comm_marketing), + ) + session.commit() + session.refresh(user) + + if is_htmx_request(request): + return toast_response( + request, + templates, + "Communication preferences updated.", + ) + return RedirectResponse(url=router.url_path_for("read_profile"), status_code=303) + + @router.get("/avatar") async def get_avatar(user: User = Depends(get_authenticated_user)): """Serve avatar image from database""" diff --git a/templates/account/register.html b/templates/account/register.html index 78d6a10..82d6314 100644 --- a/templates/account/register.html +++ b/templates/account/register.html @@ -65,6 +65,8 @@ + {% include 'partials/communication_preferences_fields.html' with context %} +
+ + diff --git a/templates/users/profile.html b/templates/users/profile.html index 2ec4521..11f82e3 100644 --- a/templates/users/profile.html +++ b/templates/users/profile.html @@ -12,6 +12,20 @@

User Profile

{% include 'users/partials/profile_display.html' %} + +
+
+ Communication Preferences +
+
+
+ {% include 'partials/communication_preferences_fields.html' with context %} + +
+
+
+
diff --git a/tests/routers/core/test_account.py b/tests/routers/core/test_account.py index 003067a..22548a9 100644 --- a/tests/routers/core/test_account.py +++ b/tests/routers/core/test_account.py @@ -85,6 +85,91 @@ def test_register_endpoint(unauth_client: TestClient, session: Session): user = session.exec(select(User).where(User.account_id == account.id)).first() assert user is not None assert user.name == "New User" + assert user.comm_opt_in is False + assert user.comm_updates is False + assert user.comm_marketing is False + + +def test_register_with_communication_preferences( + unauth_client: TestClient, session: Session +): + response = unauth_client.post( + app.url_path_for("register"), + data={ + "name": "Comm Prefs User", + "email": "commprefs@example.com", + "password": "NewPass123!@#", + "confirm_password": "NewPass123!@#", + "comm_opt_in": "on", + "comm_updates": "on", + "comm_marketing": "on", + }, + ) + assert response.status_code == 303 + + account = session.exec( + select(Account).where(Account.email == "commprefs@example.com") + ).first() + assert account is not None + user = session.exec(select(User).where(User.account_id == account.id)).first() + assert user is not None + assert user.comm_opt_in is True + assert user.comm_updates is True + assert user.comm_marketing is True + + +def test_register_communication_preferences_master_off_ignores_subs( + unauth_client: TestClient, session: Session +): + response = unauth_client.post( + app.url_path_for("register"), + data={ + "name": "Tampered Subs User", + "email": "tamperedsubs@example.com", + "password": "NewPass123!@#", + "confirm_password": "NewPass123!@#", + "comm_updates": "on", + "comm_marketing": "on", + }, + ) + assert response.status_code == 303 + + account = session.exec( + select(Account).where(Account.email == "tamperedsubs@example.com") + ).first() + assert account is not None + user = session.exec(select(User).where(User.account_id == account.id)).first() + assert user is not None + assert user.comm_opt_in is False + assert user.comm_updates is False + assert user.comm_marketing is False + + +def test_register_with_communication_preferences_updates_only( + unauth_client: TestClient, session: Session +): + response = unauth_client.post( + app.url_path_for("register"), + data={ + "name": "Updates Only User", + "email": "updatesonly@example.com", + "password": "NewPass123!@#", + "confirm_password": "NewPass123!@#", + "comm_opt_in": "on", + "comm_updates": "on", + }, + ) + assert response.status_code == 303 + + account = session.exec( + select(Account).where(Account.email == "updatesonly@example.com") + ).first() + assert account is not None + user = session.exec(select(User).where(User.account_id == account.id)).first() + assert user is not None + assert user.comm_opt_in is True + assert user.comm_updates is True + assert user.comm_marketing is False def test_register_creates_account_email_row( @@ -362,9 +447,9 @@ def test_register_page_shows_password_requirements(unauth_client: TestClient): assert "special" in html.lower(), ( "Page should mention special character requirement" ) - - -def test_register_page_confirm_password_has_autocomplete(unauth_client: TestClient): + assert 'name="comm_opt_in"' in html + assert 'name="comm_updates"' in html + assert 'name="comm_marketing"' in html """Issue #156: Both password fields must have autocomplete='new-password' for Chrome autofill.""" response = unauth_client.get(app.url_path_for("read_register")) assert response.status_code == 200 diff --git a/tests/routers/core/test_user.py b/tests/routers/core/test_user.py index b71e357..dcbc457 100644 --- a/tests/routers/core/test_user.py +++ b/tests/routers/core/test_user.py @@ -278,3 +278,86 @@ def test_profile_no_organizations( # Should show "no organizations" message assert "You are not a member of any organizations" in response.text + + +def test_read_profile_shows_communication_preferences( + auth_client: TestClient, test_user: User +): + response = auth_client.get(app.url_path_for("read_profile")) + assert response.status_code == 200 + html = response.text + assert "Communication Preferences" in html + assert 'name="comm_opt_in"' in html + assert 'name="comm_updates"' in html + assert 'name="comm_marketing"' in html + assert "Save Preferences" in html + + +def test_update_communication_preferences_unauthorized(unauth_client: TestClient): + response = unauth_client.post( + app.url_path_for("update_communication_preferences"), + data={"comm_opt_in": "on"}, + ) + assert response.status_code == 303 + assert response.headers["location"] == app.url_path_for("read_login") + + +def test_update_communication_preferences( + auth_client: TestClient, test_user: User, session: Session +): + response = auth_client.post( + app.url_path_for("update_communication_preferences"), + data={ + "comm_opt_in": "on", + "comm_updates": "on", + "comm_marketing": "on", + }, + ) + assert response.status_code == 303 + assert response.headers["location"] == app.url_path_for("read_profile") + + session.refresh(test_user) + assert test_user.comm_opt_in is True + assert test_user.comm_updates is True + assert test_user.comm_marketing is True + + +def test_update_communication_preferences_master_off_clears_subs( + auth_client: TestClient, test_user: User, session: Session +): + test_user.comm_opt_in = True + test_user.comm_updates = True + test_user.comm_marketing = True + session.add(test_user) + session.commit() + + response = auth_client.post( + app.url_path_for("update_communication_preferences"), + data={}, + ) + assert response.status_code == 303 + + session.refresh(test_user) + assert test_user.comm_opt_in is False + assert test_user.comm_updates is False + assert test_user.comm_marketing is False + + +def test_update_communication_preferences_htmx( + auth_client: TestClient, test_user: User, session: Session +): + response = auth_client.post( + app.url_path_for("update_communication_preferences"), + data={ + "comm_opt_in": "on", + "comm_updates": "on", + }, + headers={"HX-Request": "true"}, + ) + assert response.status_code == 200 + assert "Communication preferences updated." in response.text + + session.refresh(test_user) + assert test_user.comm_opt_in is True + assert test_user.comm_updates is True + assert test_user.comm_marketing is False diff --git a/tests/utils/test_communication_preferences.py b/tests/utils/test_communication_preferences.py new file mode 100644 index 0000000..6058084 --- /dev/null +++ b/tests/utils/test_communication_preferences.py @@ -0,0 +1,33 @@ +from utils.core.communication_preferences import parse_communication_preferences + + +def test_parse_communication_preferences_all_off_by_default(): + prefs = parse_communication_preferences() + assert prefs == (False, False, False) + + +def test_parse_communication_preferences_master_off_ignores_subs(): + prefs = parse_communication_preferences( + comm_opt_in=None, + comm_updates="on", + comm_marketing="on", + ) + assert prefs == (False, False, False) + + +def test_parse_communication_preferences_master_on_with_subs(): + prefs = parse_communication_preferences( + comm_opt_in="on", + comm_updates="on", + comm_marketing="on", + ) + assert prefs == (True, True, True) + + +def test_parse_communication_preferences_master_on_updates_only(): + prefs = parse_communication_preferences( + comm_opt_in="on", + comm_updates="on", + comm_marketing=None, + ) + assert prefs == (True, True, False) diff --git a/utils/core/communication_preferences.py b/utils/core/communication_preferences.py new file mode 100644 index 0000000..58024a8 --- /dev/null +++ b/utils/core/communication_preferences.py @@ -0,0 +1,32 @@ +from typing import NamedTuple, Optional + +from utils.core.models import User + + +class CommunicationPreferences(NamedTuple): + comm_opt_in: bool + comm_updates: bool + comm_marketing: bool + + +def parse_communication_preferences( + comm_opt_in: Optional[str] = None, + comm_updates: Optional[str] = None, + comm_marketing: Optional[str] = None, +) -> CommunicationPreferences: + """Parse HTML checkbox form values into communication preference booleans.""" + if comm_opt_in != "on": + return CommunicationPreferences(False, False, False) + return CommunicationPreferences( + True, + comm_updates == "on", + comm_marketing == "on", + ) + + +def apply_communication_preferences( + user: User, prefs: CommunicationPreferences +) -> None: + user.comm_opt_in = prefs.comm_opt_in + user.comm_updates = prefs.comm_updates + user.comm_marketing = prefs.comm_marketing diff --git a/utils/core/models.py b/utils/core/models.py index 970c557..a8b8f46 100644 --- a/utils/core/models.py +++ b/utils/core/models.py @@ -174,6 +174,9 @@ class RolePermissionLink(SQLModel, table=True): class UserBase(SQLModel): name: Optional[str] = None + comm_opt_in: bool = Field(default=False) + comm_updates: bool = Field(default=False) + comm_marketing: bool = Field(default=False) class UserAvatar(SQLModel, table=True):