From 7dbeda5aafc56427777c31489a2554e546417530 Mon Sep 17 00:00:00 2001 From: Ben Sima Date: Wed, 12 Nov 2025 21:38:39 -0500 Subject: Move end to end test to own file This is a big test, and probably over-dependencied, but it's nice to have an end-to-end test as an overall bug-catcher. --- Biz/PodcastItLater/Test.py | 226 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 226 insertions(+) create mode 100644 Biz/PodcastItLater/Test.py (limited to 'Biz/PodcastItLater/Test.py') diff --git a/Biz/PodcastItLater/Test.py b/Biz/PodcastItLater/Test.py new file mode 100644 index 0000000..373467f --- /dev/null +++ b/Biz/PodcastItLater/Test.py @@ -0,0 +1,226 @@ +"""End-to-end tests for PodcastItLater.""" + +# : dep boto3 +# : dep botocore +# : dep feedgen +# : dep httpx +# : dep itsdangerous +# : dep ludic +# : dep openai +# : dep psutil +# : dep pydub +# : dep pytest +# : dep pytest-asyncio +# : dep pytest-mock +# : dep starlette +# : dep stripe +# : dep trafilatura +# : dep uvicorn +# : out podcastitlater-e2e-test +# : run ffmpeg +import Biz.PodcastItLater.Core as Core +import Biz.PodcastItLater.Web as Web +import Biz.PodcastItLater.Worker as Worker +import Omni.App as App +import Omni.Test as Test +import pathlib +import re +import sys +import unittest.mock +from starlette.testclient import TestClient + + +class BaseWebTest(Test.TestCase): + """Base test class with common setup.""" + + def setUp(self) -> None: + """Set up test environment.""" + self.app = Web.app + self.client = TestClient(self.app) + + # Initialize database for each test + Core.Database.init_db() + + @staticmethod + def tearDown() -> None: + """Clean up after each test.""" + Core.Database.teardown() + + +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.""" + # 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(" None: + """Run all end-to-end tests.""" + Test.run( + App.Area.Test, + [ + TestEndToEnd, + ], + ) + + +def main() -> None: + """Run the tests.""" + if "test" in sys.argv: + test() + else: + test() -- cgit v1.2.3