Merge task t-1fbElKv: Implement change email address functionality

2 files changed, 234 insertions, 0 deletions

diff --git a/Biz/PodcastItLater/Core.py b/Biz/PodcastItLater/Core.py
index cc94659..0bb7251 100644
--- a/Biz/PodcastItLater/Core.py
+++ b/Biz/PodcastItLater/Core.py
@@ -1000,6 +1000,30 @@ class Database:  # noqa: PLR0904
             logger.info("Deleted user %s and all associated data", user_id)
 
     @staticmethod
+    def update_user_email(user_id: int, new_email: str) -> None:
+        """Update user's email address.
+
+        Args:
+            user_id: ID of the user to update
+            new_email: New email address
+
+        Raises:
+            ValueError: If email is already taken by another user
+        """
+        with Database.get_connection() as conn:
+            cursor = conn.cursor()
+            try:
+                cursor.execute(
+                    "UPDATE users SET email = ? WHERE id = ?",
+                    (new_email, user_id),
+                )
+                conn.commit()
+                logger.info("Updated user %s email to %s", user_id, new_email)
+            except sqlite3.IntegrityError:
+                msg = f"Email {new_email} is already taken"
+                raise ValueError(msg) from None
+
+    @staticmethod
     def mark_episode_public(episode_id: int) -> None:
         """Mark an episode as public."""
         with Database.get_connection() as conn:
@@ -1660,6 +1684,32 @@ class TestUserManagement(Test.TestCase):
         # Verify episodes are gone (direct lookup)
         self.assertIsNone(Database.get_episode_by_id(ep_id))
 
+    def test_update_user_email(self) -> None:
+        """Update user email address."""
+        user_id, _ = Database.create_user("old@example.com")
+
+        # Update email
+        Database.update_user_email(user_id, "new@example.com")
+
+        # Verify update
+        user = Database.get_user_by_id(user_id)
+        self.assertIsNotNone(user)
+        if user:
+            self.assertEqual(user["email"], "new@example.com")
+
+        # Old email should not exist
+        self.assertIsNone(Database.get_user_by_email("old@example.com"))
+
+    @staticmethod
+    def test_update_user_email_duplicate() -> None:
+        """Cannot update to an existing email."""
+        user_id1, _ = Database.create_user("user1@example.com")
+        Database.create_user("user2@example.com")
+
+        # Try to update user1 to user2's email
+        with pytest.raises(ValueError, match="already taken"):
+            Database.update_user_email(user_id1, "user2@example.com")
+
 
 class TestQueueOperations(Test.TestCase):
     """Test queue operations."""
diff --git a/Biz/PodcastItLater/Web.py b/Biz/PodcastItLater/Web.py
index 348c847..0bd3552 100644
--- a/Biz/PodcastItLater/Web.py
+++ b/Biz/PodcastItLater/Web.py
@@ -1219,6 +1219,147 @@ def verify_magic_link(request: Request) -> Response:
         return RedirectResponse("/?error=expired_link")
 
 
+@app.get("/settings/email/edit")
+def edit_email_form(request: Request) -> Response:
+    """Return form to edit email."""
+    user_id = request.session.get("user_id")
+    if not user_id:
+        return Response("Unauthorized", status_code=401)
+
+    user = Core.Database.get_user_by_id(user_id)
+    if not user:
+        return Response("User not found", status_code=404)
+
+    return html.div(
+        html.form(
+            html.strong("Email: ", classes=["me-2"]),
+            html.input(
+                type="email",
+                name="email",
+                value=user["email"],
+                required=True,
+                classes=[
+                    "form-control",
+                    "form-control-sm",
+                    "d-inline-block",
+                    "w-auto",
+                    "me-2",
+                ],
+            ),
+            html.button(
+                "Save",
+                type="submit",
+                classes=["btn", "btn-sm", "btn-primary", "me-1"],
+            ),
+            html.button(
+                "Cancel",
+                hx_get="/settings/email/cancel",
+                hx_target="closest div",
+                hx_swap="outerHTML",
+                classes=["btn", "btn-sm", "btn-secondary"],
+            ),
+            hx_post="/settings/email",
+            hx_target="closest div",
+            hx_swap="outerHTML",
+            classes=["d-flex", "align-items-center"],
+        ),
+        classes=["mb-2"],
+    )
+
+
+@app.get("/settings/email/cancel")
+def cancel_edit_email(request: Request) -> Response:
+    """Cancel email editing and show original view."""
+    user_id = request.session.get("user_id")
+    if not user_id:
+        return Response("Unauthorized", status_code=401)
+
+    user = Core.Database.get_user_by_id(user_id)
+    if not user:
+        return Response("User not found", status_code=404)
+
+    return html.div(
+        html.strong("Email: "),
+        html.span(user["email"]),
+        html.button(
+            "Change",
+            classes=[
+                "btn",
+                "btn-sm",
+                "btn-outline-secondary",
+                "ms-2",
+                "py-0",
+            ],
+            hx_get="/settings/email/edit",
+            hx_target="closest div",
+            hx_swap="outerHTML",
+        ),
+        classes=["mb-2", "d-flex", "align-items-center"],
+    )
+
+
+@app.post("/settings/email")
+def update_email(request: Request, data: FormData) -> Response:
+    """Update user email."""
+    user_id = request.session.get("user_id")
+    if not user_id:
+        return Response("Unauthorized", status_code=401)
+
+    new_email_raw = data.get("email", "")
+    new_email = (
+        new_email_raw.strip().lower() if isinstance(new_email_raw, str) else ""
+    )
+
+    if not new_email:
+        return Response("Email required", status_code=400)
+
+    try:
+        Core.Database.update_user_email(user_id, new_email)
+        return cancel_edit_email(request)
+    except ValueError as e:
+        # Return form with error
+        return html.div(
+            html.form(
+                html.strong("Email: ", classes=["me-2"]),
+                html.input(
+                    type="email",
+                    name="email",
+                    value=new_email,
+                    required=True,
+                    classes=[
+                        "form-control",
+                        "form-control-sm",
+                        "d-inline-block",
+                        "w-auto",
+                        "me-2",
+                        "is-invalid",
+                    ],
+                ),
+                html.button(
+                    "Save",
+                    type="submit",
+                    classes=["btn", "btn-sm", "btn-primary", "me-1"],
+                ),
+                html.button(
+                    "Cancel",
+                    hx_get="/settings/email/cancel",
+                    hx_target="closest div",
+                    hx_swap="outerHTML",
+                    classes=["btn", "btn-sm", "btn-secondary"],
+                ),
+                html.div(
+                    str(e),
+                    classes=["invalid-feedback", "d-block", "ms-2"],
+                ),
+                hx_post="/settings/email",
+                hx_target="closest div",
+                hx_swap="outerHTML",
+                classes=["d-flex", "align-items-center", "flex-wrap"],
+            ),
+            classes=["mb-2"],
+        )
+
+
 @app.get("/account")
 def account_page(request: Request) -> UI.AccountPage | RedirectResponse:
     """Account management page."""
@@ -3153,6 +3294,49 @@ class TestAccountPage(BaseWebTest):
                 "https://billing.stripe.com/test",
             )
 
+    def test_update_email_success(self) -> None:
+        """Should allow updating email."""
+        # POST new email
+        response = self.client.post(
+            "/settings/email",
+            data={"email": "new@example.com"},
+        )
+        self.assertEqual(response.status_code, 200)
+
+        # Verify update in DB
+        user = Core.Database.get_user_by_id(self.user_id)
+        self.assertEqual(user["email"], "new@example.com")  # type: ignore[index]
+
+    def test_update_email_duplicate(self) -> None:
+        """Should prevent updating to existing email."""
+        # Create another user
+        Core.Database.create_user("other@example.com")
+
+        # Try to update to their email
+        response = self.client.post(
+            "/settings/email",
+            data={"email": "other@example.com"},
+        )
+
+        # Should show error (return 200 with error message in form)
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("already taken", response.text.lower())
+
+    def test_delete_account(self) -> None:
+        """Should allow user to delete their account."""
+        # Delete account
+        response = self.client.delete("/account")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("HX-Redirect", response.headers)
+
+        # Verify user gone
+        user = Core.Database.get_user_by_id(self.user_id)
+        self.assertIsNone(user)
+
+        # Verify session cleared
+        response = self.client.get("/account", follow_redirects=False)
+        self.assertEqual(response.status_code, 307)
+
 
 def test() -> None:
     """Run all tests for the web module."""