diff options
Diffstat (limited to 'Biz/PodcastItLater/Web.py')
| -rw-r--r-- | Biz/PodcastItLater/Web.py | 239 |
1 files changed, 239 insertions, 0 deletions
diff --git a/Biz/PodcastItLater/Web.py b/Biz/PodcastItLater/Web.py index a706eb5..b41f31d 100644 --- a/Biz/PodcastItLater/Web.py +++ b/Biz/PodcastItLater/Web.py @@ -2363,6 +2363,242 @@ class TestEpisodeDetailPage(BaseWebTest): ) +class TestPublicFeed(BaseWebTest): + """Test public feed functionality.""" + + def setUp(self) -> None: + """Set up test database, client, and create sample episodes.""" + super().setUp() + + # Create admin user + self.admin_id, _ = Core.Database.create_user( + "ben@bensima.com", + status="active", + ) + + # Create some episodes, some public, some private + self.public_episode_id = Core.Database.create_episode( + title="Public Episode", + audio_url="https://example.com/public.mp3", + duration=300, + content_length=1000, + user_id=self.admin_id, + author="Test Author", + original_url="https://example.com/public", + original_url_hash=Core.hash_url("https://example.com/public"), + ) + Core.Database.mark_episode_public(self.public_episode_id) + + self.private_episode_id = Core.Database.create_episode( + title="Private Episode", + audio_url="https://example.com/private.mp3", + duration=200, + content_length=800, + user_id=self.admin_id, + author="Test Author", + original_url="https://example.com/private", + original_url_hash=Core.hash_url("https://example.com/private"), + ) + + def test_public_feed_page(self) -> None: + """Public feed page should show only public episodes.""" + response = self.client.get("/public") + + self.assertEqual(response.status_code, 200) + self.assertIn("Public Episode", response.text) + self.assertNotIn("Private Episode", response.text) + + def test_home_page_shows_public_feed_when_logged_out(self) -> None: + """Home page should show public episodes when user is not logged in.""" + response = self.client.get("/") + + self.assertEqual(response.status_code, 200) + self.assertIn("Public Episode", response.text) + self.assertNotIn("Private Episode", response.text) + + def test_admin_can_toggle_episode_public(self) -> None: + """Admin should be able to toggle episode public/private status.""" + # Login as admin + self.client.post("/login", data={"email": "ben@bensima.com"}) + + # Toggle private episode to public + response = self.client.post( + f"/admin/episode/{self.private_episode_id}/toggle-public", + ) + + self.assertEqual(response.status_code, 200) + + # Verify it's now public + episode = Core.Database.get_episode_by_id(self.private_episode_id) + self.assertEqual(episode["is_public"], 1) # type: ignore[index] + + def test_non_admin_cannot_toggle_public(self) -> None: + """Non-admin users should not be able to toggle public status.""" + # Create and login as regular user + _user_id, _ = Core.Database.create_user("user@example.com") + self.client.post("/login", data={"email": "user@example.com"}) + + # Try to toggle + response = self.client.post( + f"/admin/episode/{self.private_episode_id}/toggle-public", + ) + + self.assertEqual(response.status_code, 403) + + +class TestEpisodeDeduplication(BaseWebTest): + """Test episode deduplication functionality.""" + + def setUp(self) -> None: + """Set up test database, client, and create test user.""" + super().setUp() + + self.user_id, self.token = Core.Database.create_user( + "test@example.com", + status="active", + ) + + # Create an existing episode + self.existing_url = "https://example.com/article" + self.url_hash = Core.hash_url(self.existing_url) + + self.episode_id = Core.Database.create_episode( + title="Existing Article", + audio_url="https://example.com/audio.mp3", + duration=300, + content_length=1000, + user_id=self.user_id, + author="Test Author", + original_url=self.existing_url, + original_url_hash=self.url_hash, + ) + + def test_url_normalization(self) -> None: + """URLs should be normalized for deduplication.""" + # Different URL variations that should be normalized to same hash + urls = [ + "http://example.com/article", + "https://example.com/article", + "https://www.example.com/article", + "https://EXAMPLE.COM/article", + "https://example.com/article/", + ] + + hashes = [Core.hash_url(url) for url in urls] + + # All should produce the same hash + self.assertEqual(len(set(hashes)), 1) + + def test_find_existing_episode_by_hash(self) -> None: + """Should find existing episode by normalized URL hash.""" + # Try different URL variations + similar_urls = [ + "http://example.com/article", + "https://www.example.com/article", + ] + + for url in similar_urls: + url_hash = Core.hash_url(url) + episode = Core.Database.find_episode_by_url_hash(url_hash) + + self.assertIsNotNone(episode) + self.assertEqual(episode["id"], self.episode_id) # type: ignore[index] + + def test_add_existing_episode_to_user_feed(self) -> None: + """Should add existing episode to new user's feed.""" + # Create second user + user2_id, _ = Core.Database.create_user("user2@example.com") + + # Add existing episode to their feed + Core.Database.add_episode_to_user(user2_id, self.episode_id) + + # Verify it appears in their feed + episodes = Core.Database.get_user_episodes(user2_id) + episode_ids = [e["id"] for e in episodes] + + self.assertIn(self.episode_id, episode_ids) + + +class TestMetricsTracking(BaseWebTest): + """Test episode metrics tracking.""" + + def setUp(self) -> None: + """Set up test database, client, and create test episode.""" + super().setUp() + + self.user_id, _ = Core.Database.create_user( + "test@example.com", + status="active", + ) + + self.episode_id = Core.Database.create_episode( + title="Test Episode", + audio_url="https://example.com/audio.mp3", + duration=300, + content_length=1000, + user_id=self.user_id, + author="Test Author", + original_url="https://example.com/article", + original_url_hash=Core.hash_url("https://example.com/article"), + ) + + def test_track_episode_added(self) -> None: + """Should track when episode is added to feed.""" + Core.Database.track_episode_event( + self.episode_id, + "added", + self.user_id, + ) + + # Verify metric was recorded + metrics = Core.Database.get_episode_metrics(self.episode_id) + self.assertEqual(len(metrics), 1) + self.assertEqual(metrics[0]["event_type"], "added") + self.assertEqual(metrics[0]["user_id"], self.user_id) + + def test_track_episode_played(self) -> None: + """Should track when episode is played.""" + Core.Database.track_episode_event( + self.episode_id, + "played", + self.user_id, + ) + + metrics = Core.Database.get_episode_metrics(self.episode_id) + self.assertEqual(len(metrics), 1) + self.assertEqual(metrics[0]["event_type"], "played") + + def test_track_anonymous_play(self) -> None: + """Should track plays from anonymous users.""" + Core.Database.track_episode_event( + self.episode_id, + "played", + user_id=None, + ) + + metrics = Core.Database.get_episode_metrics(self.episode_id) + self.assertEqual(len(metrics), 1) + self.assertEqual(metrics[0]["event_type"], "played") + self.assertIsNone(metrics[0]["user_id"]) + + def test_track_endpoint(self) -> None: + """POST /episode/{id}/track should record metrics.""" + # Login as user + self.client.post("/login", data={"email": "test@example.com"}) + + response = self.client.post( + f"/episode/{self.episode_id}/track", + data={"event_type": "played"}, + ) + + self.assertEqual(response.status_code, 200) + + # Verify metric was recorded + metrics = Core.Database.get_episode_metrics(self.episode_id) + played_metrics = [m for m in metrics if m["event_type"] == "played"] + self.assertGreater(len(played_metrics), 0) + + def test() -> None: """Run all tests for the web module.""" Test.run( @@ -2375,6 +2611,9 @@ def test() -> None: TestAdminInterface, TestJobCancellation, TestEpisodeDetailPage, + TestPublicFeed, + TestEpisodeDeduplication, + TestMetricsTracking, ], ) |