diff options
Diffstat (limited to 'Biz/PodcastItLater/Web.py')
| -rw-r--r-- | Biz/PodcastItLater/Web.py | 168 |
1 files changed, 134 insertions, 34 deletions
diff --git a/Biz/PodcastItLater/Web.py b/Biz/PodcastItLater/Web.py index 044cdca..0f095d3 100644 --- a/Biz/PodcastItLater/Web.py +++ b/Biz/PodcastItLater/Web.py @@ -54,6 +54,7 @@ from starlette.middleware.sessions import SessionMiddleware from starlette.responses import RedirectResponse from starlette.testclient import TestClient from typing import override +from unittest.mock import patch logger = logging.getLogger(__name__) Log.setup(logger) @@ -384,6 +385,11 @@ class QueueStatus(Component[AnyChildren, QueueStatusAttrs]): badge_class = status_classes.get(item["status"], "bg-secondary") icon_class = status_icons.get(item["status"], "bi-question-circle") + # Get queue position for pending items + queue_pos = None + if item["status"] == "pending": + queue_pos = Core.Database.get_queue_position(item["id"]) + queue_items.append( html.div( html.div( @@ -435,6 +441,16 @@ class QueueStatus(Component[AnyChildren, QueueStatusAttrs]): f"Created: {item['created_at']}", classes=["text-muted", "d-block", "mt-1"], ), + # Display queue position if available + html.small( + html.i( + classes=["bi", "bi-hourglass-split", "me-1"], + ), + f"Position in queue: #{queue_pos}", + classes=["text-info", "d-block", "mt-1"], + ) + if queue_pos + else html.span(), *( [ html.div( @@ -462,6 +478,33 @@ class QueueStatus(Component[AnyChildren, QueueStatusAttrs]): ), # Add cancel button for pending jobs, remove for others html.div( + # Retry button for error items + html.button( + html.i( + classes=[ + "bi", + "bi-arrow-clockwise", + "me-1", + ], + ), + "Retry", + hx_post=f"/queue/{item['id']}/retry", + hx_trigger="click", + hx_on=( + "htmx:afterRequest: " + "if(event.detail.successful) " + "htmx.trigger('body', 'queue-updated')" + ), + classes=[ + "btn", + "btn-sm", + "btn-outline-primary", + "mt-2", + "me-2", + ], + ) + if item["status"] == "error" + else html.span(), html.button( html.i(classes=["bi", "bi-x-lg", "me-1"]), "Cancel", @@ -1009,6 +1052,57 @@ def upgrade(request: Request) -> RedirectResponse: return RedirectResponse(url="/pricing?error=checkout_failed") +@app.get("/account") +def account(request: Request) -> UI.AccountPage | RedirectResponse: + """Display account management page.""" + user_id = request.session.get("user_id") + if not user_id: + return RedirectResponse(url="/?error=login_required") + + user = Core.Database.get_user_by_id(user_id) + if not user: + request.session.clear() + return RedirectResponse(url="/?error=user_not_found") + + # Get usage stats + period_start, period_end = Billing.get_period_boundaries(user) + usage = Billing.get_usage(user_id, period_start, period_end) + + # Get limits + tier = user.get("plan_tier", "free") + limits = Billing.TIER_LIMITS.get(tier, Billing.TIER_LIMITS["free"]) + + return UI.AccountPage( + user=user, + usage=usage, + limits=limits, + portal_url="/billing/portal" if tier == "paid" else None, + ) + + +@app.post("/logout") +def logout(request: Request) -> RedirectResponse: + """Log out user.""" + request.session.clear() + return RedirectResponse(url="/", status_code=303) + + +@app.post("/billing/portal") +def billing_portal(request: Request) -> RedirectResponse: + """Redirect to Stripe billing portal.""" + user_id = request.session.get("user_id") + if not user_id: + return RedirectResponse(url="/?error=login_required") + + try: + portal_url = Billing.create_portal_session(user_id, BASE_URL) + return RedirectResponse(url=portal_url, status_code=303) + except ValueError as e: + logger.warning("Failed to create portal session: %s", e) + # If user has no customer ID (e.g. free tier), redirect to pricing + return RedirectResponse(url="/pricing") + + def _handle_test_login(email: str, request: Request) -> Response: """Handle login in test mode.""" # Special handling for demo account @@ -1167,11 +1261,11 @@ def account_page(request: Request) -> UI.AccountPage | RedirectResponse: # Get usage stats period_start, period_end = Billing.get_period_boundaries(user) usage = Billing.get_usage(user["id"], period_start, period_end) - + # Get limits tier = user.get("plan_tier", "free") limits = Billing.TIER_LIMITS.get(tier, Billing.TIER_LIMITS["free"]) - + return UI.AccountPage( user=user, usage=usage, @@ -1180,17 +1274,6 @@ def account_page(request: Request) -> UI.AccountPage | RedirectResponse: ) -@app.get("/logout") -def logout(request: Request) -> Response: - """Handle logout.""" - request.session.clear() - return Response( - "", - status_code=302, - headers={"Location": "/"}, - ) - - @app.post("/submit") def submit_article( # noqa: PLR0911, PLR0914 request: Request, @@ -1565,21 +1648,6 @@ def billing_checkout(request: Request, data: FormData) -> Response: return Response(f"Error: {e!s}", status_code=400) -@app.post("/billing/portal") -def billing_portal(request: Request) -> Response | RedirectResponse: - """Create Stripe Billing Portal session.""" - user_id = request.session.get("user_id") - if not user_id: - return Response("Unauthorized", status_code=401) - - try: - portal_url = Billing.create_portal_session(user_id, BASE_URL) - return RedirectResponse(url=portal_url, status_code=303) - except Exception: - logger.exception("Portal error - ensure Stripe portal is configured") - return Response("Portal not configured", status_code=500) - - @app.post("/stripe/webhook") async def stripe_webhook(request: Request) -> Response: """Handle Stripe webhook events.""" @@ -3036,9 +3104,9 @@ class TestAccountPage(BaseWebTest): ) self.client.post("/login", data={"email": "test@example.com"}) - def test_account_page_shows_usage(self) -> None: - """Account page should show usage stats.""" - # Create some usage + def test_account_page_logged_in(self) -> None: + """Account page should render for logged-in users.""" + # Create some usage to verify stats are shown ep_id = Core.Database.create_episode( title="Test Episode", audio_url="https://example.com/audio.mp3", @@ -3052,19 +3120,51 @@ class TestAccountPage(BaseWebTest): Core.Database.add_episode_to_user(self.user_id, ep_id) response = self.client.get("/account") - + self.assertEqual(response.status_code, 200) - self.assertIn("1 / 10", response.text) # Usage / Limit for free tier self.assertIn("My Account", response.text) self.assertIn("test@example.com", response.text) + self.assertIn("1 / 10", response.text) # Usage / Limit for free tier def test_account_page_login_required(self) -> None: """Should redirect to login if not logged in.""" - self.client.get("/logout") + self.client.post("/logout") response = self.client.get("/account", follow_redirects=False) self.assertEqual(response.status_code, 307) self.assertEqual(response.headers["location"], "/?error=login_required") + def test_logout(self) -> None: + """Logout should clear session.""" + response = self.client.post("/logout", follow_redirects=False) + self.assertEqual(response.status_code, 303) + self.assertEqual(response.headers["location"], "/") + + # Verify session cleared + response = self.client.get("/account", follow_redirects=False) + self.assertEqual(response.status_code, 307) + + def test_billing_portal_redirect(self) -> None: + """Billing portal should redirect to Stripe.""" + # First set a customer ID + Core.Database.set_user_stripe_customer(self.user_id, "cus_test") + + # Mock the create_portal_session method + with patch( + "Biz.PodcastItLater.Billing.create_portal_session", + ) as mock_portal: + mock_portal.return_value = "https://billing.stripe.com/test" + + response = self.client.post( + "/billing/portal", + follow_redirects=False, + ) + + self.assertEqual(response.status_code, 303) + self.assertEqual( + response.headers["location"], + "https://billing.stripe.com/test", + ) + def test() -> None: """Run all tests for the web module.""" |