1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
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("<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)
def test() -> 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()
|
