diff options
Diffstat (limited to 'Biz/PodcastItLater/Web.py')
| -rw-r--r-- | Biz/PodcastItLater/Web.py | 165 |
1 files changed, 165 insertions, 0 deletions
diff --git a/Biz/PodcastItLater/Web.py b/Biz/PodcastItLater/Web.py index 1c628d5..697b92c 100644 --- a/Biz/PodcastItLater/Web.py +++ b/Biz/PodcastItLater/Web.py @@ -2161,6 +2161,170 @@ class TestAdminInterface(BaseWebTest): self.assertIn("PROCESSING: 1", response.text) +class TestEndToEnd(BaseWebTest): + """Test complete end-to-end flows.""" + + def setUp(self) -> None: + """Set up test client with logged-in user.""" + super().setUp() + + # Create and login user + self.user_id, self.token = Core.Database.create_user( + "test@example.com", + ) + Core.Database.update_user_status( + self.user_id, + "active", + ) + self.client.post("/login", data={"email": "test@example.com"}) + + def test_full_article_to_rss_flow(self) -> None: # noqa: PLR0915 + """Test complete flow: submit URL → process → appears in RSS feed.""" + import Biz.PodcastItLater.Worker as Worker # noqa: PLC0415 + import unittest.mock # noqa: PLC0415 + + # Step 1: Submit article URL + response = self.client.post( + "/submit", + data={"url": "https://example.com/great-article"}, + ) + + self.assertEqual(response.status_code, 200) + self.assertIn("Article submitted successfully", response.text) + + # Extract job ID from response + match = re.search(r"Job ID: (\d+)", response.text) + self.assertIsNotNone(match) + if match is None: + self.fail("Job ID not found in response") + job_id = int(match.group(1)) + + # Verify job was created + job = Core.Database.get_job_by_id(job_id) + self.assertIsNotNone(job) + if job is None: + self.fail("Job should not be None") + self.assertEqual(job["status"], "pending") + self.assertEqual(job["user_id"], self.user_id) + + # Step 2: Process the job with mocked external services + shutdown_handler = Worker.ShutdownHandler() + processor = Worker.ArticleProcessor(shutdown_handler) + + # Mock external dependencies + mock_audio_data = b"fake-mp3-audio-content-12345" + + with ( + unittest.mock.patch.object( + Worker.ArticleProcessor, + "extract_article_content", + return_value=( + "Great Article Title", + "This is the article content.", + ), + ), + unittest.mock.patch( + "Biz.PodcastItLater.Worker.prepare_text_for_tts", + return_value=["This is the article content."], + ), + unittest.mock.patch( + "Biz.PodcastItLater.Worker.check_memory_usage", + return_value=50.0, + ), + unittest.mock.patch.object( + processor.openai_client.audio.speech, + "create", + ) as mock_tts, + unittest.mock.patch.object( + processor, + "upload_to_s3", + return_value="https://cdn.example.com/episode_123_Great_Article.mp3", + ), + unittest.mock.patch( + "pydub.AudioSegment.from_mp3", + ) as mock_audio_segment, + unittest.mock.patch( + "pathlib.Path.read_bytes", + return_value=mock_audio_data, + ), + ): + # Mock TTS response + mock_tts_response = unittest.mock.MagicMock() + mock_tts_response.content = mock_audio_data + mock_tts.return_value = mock_tts_response + + # Mock audio segment + mock_segment = unittest.mock.MagicMock() + mock_segment.export = lambda path, **_kwargs: pathlib.Path( + path, + ).write_bytes( + mock_audio_data, + ) + mock_audio_segment.return_value = mock_segment + + # Process the pending job + Worker.process_pending_jobs(processor) + + # Step 3: Verify job was marked completed + job = Core.Database.get_job_by_id(job_id) + self.assertIsNotNone(job) + if job is None: + self.fail("Job should not be None") + self.assertEqual(job["status"], "completed") + + # Step 4: Verify episode was created + episodes = Core.Database.get_user_all_episodes(self.user_id) + self.assertEqual(len(episodes), 1) + + episode = episodes[0] + self.assertEqual(episode["title"], "Great Article Title") + self.assertEqual( + episode["audio_url"], + "https://cdn.example.com/episode_123_Great_Article.mp3", + ) + self.assertGreater(episode["duration"], 0) + self.assertEqual(episode["user_id"], self.user_id) + + # Step 5: Verify episode appears in RSS feed + response = self.client.get(f"/feed/{self.token}.xml") + + self.assertEqual(response.status_code, 200) + self.assertEqual( + response.headers["content-type"], + "application/rss+xml; charset=utf-8", + ) + + # Check RSS contains the episode + self.assertIn("Great Article Title", response.text) + self.assertIn( + "https://cdn.example.com/episode_123_Great_Article.mp3", + response.text, + ) + self.assertIn("<enclosure", response.text) + self.assertIn('type="audio/mpeg"', response.text) + + # Step 6: Verify only this user's episode is in their feed + # Create another user with their own episode + user2_id, token2 = Core.Database.create_user("other@example.com") + Core.Database.create_episode( + "Other User's Article", + "https://cdn.example.com/other.mp3", + 200, + 3000, + user2_id, + ) + + # Original user's feed should not contain other user's episode + response = self.client.get(f"/feed/{self.token}.xml") + self.assertIn("Great Article Title", response.text) + self.assertNotIn("Other User's Article", response.text) + + # Other user's feed should only contain their episode + response = self.client.get(f"/feed/{token2}.xml") + self.assertNotIn("Great Article Title", response.text) + self.assertIn("Other User's Article", response.text) + + class TestJobCancellation(BaseWebTest): """Test job cancellation functionality.""" @@ -2288,6 +2452,7 @@ def test() -> None: TestArticleSubmission, TestRSSFeed, TestAdminInterface, + TestEndToEnd, TestJobCancellation, ], ) |