From f005f68a3b9f7fd8cf019ec28b52b2cfaac508f6 Mon Sep 17 00:00:00 2001 From: Ben Sima Date: Tue, 18 Nov 2025 18:57:13 -0500 Subject: Fix homepage auto-refresh and add test coverage for admin workflows - Fix dashboard-updates endpoint to return Response with both components concatenated as HTML strings, preventing episodes from disappearing after HTMX innerHTML swap - Add viewing_own_feed flag to EpisodeList to hide 'In your feed' button when users are viewing their own feed on homepage - Add test coverage for admin adding user episodes to own feed - Add test coverage for admin adding user episodes to public feed Amp-Thread-ID: https://ampcode.com/threads/T-6d73d458-3d80-44e5-865f-358a69e5b2bf Co-authored-by: Amp --- Biz/PodcastItLater/Web.py | 122 ++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 113 insertions(+), 9 deletions(-) (limited to 'Biz') diff --git a/Biz/PodcastItLater/Web.py b/Biz/PodcastItLater/Web.py index 6d00649..f996dbd 100644 --- a/Biz/PodcastItLater/Web.py +++ b/Biz/PodcastItLater/Web.py @@ -516,6 +516,7 @@ class EpisodeListAttrs(Attrs): episodes: list[dict[str, typing.Any]] rss_url: str | None user: dict[str, typing.Any] | None + viewing_own_feed: bool class EpisodeList(Component[AnyChildren, EpisodeListAttrs]): @@ -526,6 +527,7 @@ class EpisodeList(Component[AnyChildren, EpisodeListAttrs]): episodes = self.attrs["episodes"] rss_url = self.attrs.get("rss_url") user = self.attrs.get("user") + viewing_own_feed = self.attrs.get("viewing_own_feed", False) if not episodes: return html.div( @@ -571,8 +573,9 @@ class EpisodeList(Component[AnyChildren, EpisodeListAttrs]): ) # "Add to my feed" button for logged-in users + # (only when NOT viewing own feed) user_button: html.div | html.button = html.div() - if user: + if user and not viewing_own_feed: # Check if user already has this episode user_has_episode = Core.Database.user_has_episode( user["id"], @@ -756,6 +759,7 @@ class PublicFeedPage(Component[AnyChildren, PublicFeedPageAttrs]): episodes=episodes, rss_url=f"{BASE_URL}/public.rss", user=user, + viewing_own_feed=False, ), ), user=user, @@ -865,6 +869,7 @@ class HomePage(Component[AnyChildren, HomePageAttrs]): episodes=episodes, rss_url=None, user=None, + viewing_own_feed=False, ), ), user=None, @@ -881,6 +886,7 @@ class HomePage(Component[AnyChildren, HomePageAttrs]): episodes=episodes, rss_url=f"{BASE_URL}/feed/{user['token']}.xml", user=user, + viewing_own_feed=True, ), id="dashboard-content", hx_get="/dashboard-updates", @@ -1589,14 +1595,22 @@ def queue_status(request: Request) -> QueueStatus: @app.get("/dashboard-updates") -def dashboard_updates(request: Request) -> html.div: +def dashboard_updates(request: Request) -> Response: """Return both queue status and recent episodes for dashboard updates.""" # Check if user is logged in user_id = request.session.get("user_id") if not user_id: - return html.div( - QueueStatus(items=[]), - EpisodeList(episodes=[], rss_url=None, user=None), + queue_status = QueueStatus(items=[]) + episode_list = EpisodeList( + episodes=[], + rss_url=None, + user=None, + viewing_own_feed=False, + ) + # Return HTML as string with both components + return Response( + str(queue_status) + str(episode_list), + media_type="text/html", ) # Get user info for RSS URL @@ -1607,10 +1621,18 @@ def dashboard_updates(request: Request) -> html.div: queue_items = Core.Database.get_user_queue_status(user_id) episodes = Core.Database.get_user_recent_episodes(user_id, 10) - return html.div( - QueueStatus(items=queue_items), - EpisodeList(episodes=episodes, rss_url=rss_url, user=user), - id="dashboard-content", + # Return just the content components, not the wrapper div + # The wrapper div with HTMX attributes is in HomePage + queue_status = QueueStatus(items=queue_items) + episode_list = EpisodeList( + episodes=episodes, + rss_url=rss_url, + user=user, + viewing_own_feed=True, + ) + return Response( + str(queue_status) + str(episode_list), + media_type="text/html", ) @@ -2661,6 +2683,88 @@ class TestPublicFeed(BaseWebTest): self.assertEqual(response.status_code, 403) + def test_admin_can_add_user_episode_to_own_feed(self) -> None: + """Admin can add another user's episode to their own feed.""" + # Create regular user and their episode + user_id, _ = Core.Database.create_user( + "user@example.com", + status="active", + ) + user_episode_id = Core.Database.create_episode( + title="User Episode", + audio_url="https://example.com/user.mp3", + duration=400, + content_length=1200, + user_id=user_id, + author="User Author", + original_url="https://example.com/user-article", + original_url_hash=Core.hash_url("https://example.com/user-article"), + ) + Core.Database.add_episode_to_user(user_id, user_episode_id) + + # Login as admin + self.client.post("/login", data={"email": "ben@bensima.com"}) + + # Admin adds user's episode to their own feed + response = self.client.post(f"/episode/{user_episode_id}/add-to-feed") + + self.assertEqual(response.status_code, 200) + + # Verify episode is now in admin's feed + admin_episodes = Core.Database.get_user_episodes(self.admin_id) + episode_ids = [e["id"] for e in admin_episodes] + self.assertIn(user_episode_id, episode_ids) + + # Verify "added" event was tracked + metrics = Core.Database.get_episode_metric_events(user_episode_id) + added_events = [m for m in metrics if m["event_type"] == "added"] + self.assertEqual(len(added_events), 1) + self.assertEqual(added_events[0]["user_id"], self.admin_id) + + def test_admin_can_add_user_episode_to_public_feed(self) -> None: + """Admin should be able to add another user's episode to public feed.""" + # Create regular user and their episode + user_id, _ = Core.Database.create_user( + "user@example.com", + status="active", + ) + user_episode_id = Core.Database.create_episode( + title="User Episode for Public", + audio_url="https://example.com/user-public.mp3", + duration=500, + content_length=1500, + user_id=user_id, + author="User Author", + original_url="https://example.com/user-public-article", + original_url_hash=Core.hash_url( + "https://example.com/user-public-article", + ), + ) + Core.Database.add_episode_to_user(user_id, user_episode_id) + + # Verify episode is private initially + episode = Core.Database.get_episode_by_id(user_episode_id) + self.assertEqual(episode["is_public"], 0) # type: ignore[index] + + # Login as admin + self.client.post("/login", data={"email": "ben@bensima.com"}) + + # Admin toggles episode to public + response = self.client.post( + f"/admin/episode/{user_episode_id}/toggle-public", + ) + + self.assertEqual(response.status_code, 200) + + # Verify episode is now public + episode = Core.Database.get_episode_by_id(user_episode_id) + self.assertEqual(episode["is_public"], 1) # type: ignore[index] + + # Verify episode appears in public feed + public_episodes = Core.Database.get_public_episodes() + episode_ids = [e["id"] for e in public_episodes] + self.assertIn(user_episode_id, episode_ids) + class TestEpisodeDeduplication(BaseWebTest): """Test episode deduplication functionality.""" -- cgit v1.2.3