Coverage for tests/streamwise_app/test_streampersona.py: 97%

1#!/usr/bin/env python3 

2""" 

3Unit tests for StreamPersona. 

4""" 

5 

6import os 

7import sys 

8import asyncio 

9import pytest 

10import aiofiles 

11 

12from typing import List 

13from typing import Any 

14 

15from PIL import Image 

16 

17from http import HTTPStatus 

18 

19from pptx import Presentation 

20 

21from quart import Quart 

22 

23from unittest.mock import patch 

24from unittest.mock import MagicMock 

25from unittest.mock import AsyncMock 

26 

27# Add current path 

28sys.path.append(os.getcwd()) 

29 

30from file_utils import read_file_base64 

31from file_utils import read_file_bytes 

32from file_utils import binary_to_base64 

33 

34from tests.test_utils import temp_sys_path 

35from tests.torch_mock import TorchMock 

36from tests.streamwise_app.app_test_helpers import check_app_root 

37from tests.streamwise_app.app_test_helpers import check_health 

38from tests.streamwise_app.app_test_helpers import check_files 

39from tests.streamwise_app.app_test_helpers import check_unknown_route 

40from tests.streamwise_app.app_test_helpers import check_job_submit_page 

41from tests.streamwise_app.app_test_helpers import check_job_status_page 

42from tests.streamwise_app.app_test_helpers import check_api_job_status 

43from tests.streamwise_app.app_test_helpers import check_api_job_requests 

44 

45mock_torch = TorchMock() 

46mock_modules = {} 

47mock_modules.update(mock_torch.get_sub_modules()) 

48 

49with patch.dict(sys.modules, mock_modules): 

50 from media_utils import save_video_frames 

51 

52 with temp_sys_path("apps", "apps/streampersona"): 

53 from apps.streampersona.streampersona import StreamPersonaApp 

54 from apps.streampersona.streampersona_job import StreamPersonaJob 

55 from apps.streampersona.streampersona_job import overlay_image_on_image 

56 from apps.streamwise_job import OutputMode 

57 from apps.client import ServiceRequest 

58 from apps.client import RequestStatus 

59 

60streampersona_app = StreamPersonaApp() 

61 

62 

63@pytest.fixture(name="test_app") 

64def _test_app() -> Quart: 

65 return streampersona_app.app 

66 

67 

68@pytest.mark.asyncio 

69async def test_app(test_app: Quart) -> None: 

70 """Check that GET / returns 200.""" 

71 await check_app_root(test_app, "StreamPersona") 

72 

73 

74@pytest.mark.asyncio 

75async def test_health(test_app: Quart) -> None: 

76 """Check /health.""" 

77 await check_health(test_app) 

78 

79 

80@pytest.mark.asyncio 

81async def test_files(test_app: Quart) -> None: 

82 """Check /files endpoint.""" 

83 await check_files(test_app, "streampersona") 

84 

85 

86@pytest.mark.asyncio 

87async def test_unknown_route(test_app: Quart) -> None: 

88 """Check that an unknown route returns 404.""" 

89 await check_unknown_route(test_app) 

90 

91 

92@pytest.mark.asyncio 

93async def test_job_submit_page(test_app: Quart) -> None: 

94 """Check the web page for job submission.""" 

95 await check_job_submit_page(test_app) 

96 

97 

98@pytest.mark.asyncio 

99async def test_job_status_page(test_app: Quart) -> None: 

100 """Check the web page for job status.""" 

101 await check_job_status_page(test_app) 

102 

103 

104@pytest.mark.asyncio 

105async def test_api_job_status(test_app: Quart) -> None: 

106 """Check the API for job status (returns UNKNOWN for nonexistent jobs).""" 

107 await check_api_job_status(test_app) 

108 

109 

110@pytest.mark.asyncio 

111async def test_api_job_requests(test_app: Quart) -> None: 

112 """Check the API for job requests listing (returns empty for nonexistent jobs).""" 

113 await check_api_job_requests(test_app) 

114 

115 

116def create_example_slides( 

117 texts: List[str] = ["Hello", "World"], 

118 file_name: str = "/tmp/example.pptx" 

119) -> None: 

120 ret = Presentation() 

121 layout = ret.slide_layouts[0] 

122 

123 for text in texts: 

124 slide = ret.slides.add_slide(layout) 

125 slide.shapes.title.text = text 

126 

127 # Save the file 

128 ret.save(file_name) 

129 

130 

131@pytest.mark.asyncio 

132async def test_api_job_submit(test_app: Quart) -> None: 

133 """Check the API for job requests.""" 

134 client = test_app.test_client() 

135 

136 # Mock the service manager 

137 streampersona_app.service_manager = MagicMock() 

138 

139 response = await client.post("/api/job", json={"pdf_base64": "AAAA"}) 

140 assert response.status_code == HTTPStatus.BAD_REQUEST 

141 response_json = await response.get_json() 

142 assert "error" in response_json 

143 assert response_json["error"] == "Missing 'pptx_base64' in job config" 

144 

145 # Success case 

146 create_example_slides( 

147 texts=["Slide 1", "Slide 2"], 

148 file_name="/tmp/example.pptx" 

149 ) 

150 async with aiofiles.open("/tmp/example.pptx", "rb") as file: 

151 pptx_binary = await file.read() 

152 pptx_base64 = binary_to_base64(pptx_binary) 

153 

154 response = await client.post("/api/job", json={"pptx_base64": pptx_base64}) 

155 assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR 

156 response_json = await response.get_json() 

157 assert "error" in response_json 

158 assert "No such file or directory: 'libreoffice'" in response_json["error"] 

159 

160 """ 

161 # TODO mimic libreoffice being present 

162 assert response.status_code == HTTPStatus.OK 

163 response_json = await response.get_json() 

164 assert "job_id" in response_json 

165 assert response_json["status"] == "success" 

166 """ 

167 

168 

169async def _mock_generation( 

170 job: StreamPersonaJob 

171) -> StreamPersonaJob: 

172 """ 

173 Mock multi-modal generation components. 

174 """ 

175 audio_base64 = await read_file_base64("tests/data/audio_4675.wav") 

176 job.gen.service_manager.get_service_url = MagicMock( 

177 return_value="http://mock_service_url:1234" 

178 ) 

179 job.gen.gen_audio = AsyncMock(return_value=audio_base64) 

180 loop = asyncio.get_running_loop() 

181 image_future: asyncio.Future[Image.Image] = loop.create_future() 

182 image_future.set_result(Image.new("RGB", (640, 480), color="blue")) 

183 job.image_task = image_future # type: ignore[assignment] 

184 

185 video_frames: List[Image.Image] = [ 

186 Image.new("RGB", (640, 480), color=color) 

187 for color in ["blue", "green", "red", "yellow"] 

188 ] 

189 video_path = await save_video_frames(video_frames) 

190 video_binary = await read_file_bytes(video_path) 

191 job.gen.gen_video_audio_from_img = AsyncMock(return_value=video_binary) 

192 

193 async def gen_video_mock(*args: Any, **kwargs: Any) -> Any: 

194 if kwargs.get("wait_request", None): 

195 return video_binary 

196 req = ServiceRequest( 

197 request_id="mock_request_id", 

198 service_name="hunyuanframepackf1", 

199 payload_json={"job_id": job.job_id}, 

200 ) 

201 req.status = RequestStatus.COMPLETED 

202 req.future = asyncio.Future() 

203 req.future.set_result(("video/mp4", video_binary)) 

204 return req 

205 

206 job.gen.gen_video = AsyncMock(side_effect=gen_video_mock) 

207 return job 

208 

209 

210@pytest.mark.asyncio 

211async def test_gen_slide_video() -> None: 

212 service_manager = MagicMock() 

213 job_id = "test_gen_slide_video" 

214 

215 job = None 

216 try: 

217 job = StreamPersonaJob(job_id, service_manager) 

218 

219 with pytest.raises(Exception, match="Error generating audio"): 

220 await job.gen_slide_video( 

221 slide_number=0, 

222 slide_text="Test Slide 0", 

223 ) 

224 

225 # Mocking generation 

226 job = await _mock_generation(job) 

227 

228 await job.gen_slide_video( 

229 slide_number=1, 

230 slide_text="Test Slide 1", 

231 ) 

232 finally: 

233 if job: 

234 await job.close() 

235 

236 

237@pytest.mark.asyncio 

238async def test_gen_slide_video_unsynced() -> None: 

239 service_manager = MagicMock() 

240 job_id = "test_gen_slide_video_unsynced" 

241 

242 job = None 

243 try: 

244 job = StreamPersonaJob(job_id, service_manager) 

245 job.config["output_mode"] = OutputMode.VIDEO_AUDIO_UNSYNCED 

246 

247 with pytest.raises(Exception, match="Error generating audio"): 

248 await job.gen_slide_video( 

249 slide_number=0, 

250 slide_text="Test Slide 0", 

251 ) 

252 job = await _mock_generation(job) 

253 await job.gen_slide_video( 

254 slide_number=1, 

255 slide_text="Test Slide Unsynced", 

256 ) 

257 finally: 

258 if job: 

259 await job.close() 

260 

261 

262def test_overlay_image_on_image() -> None: 

263 """Test overlay_image_on_image().""" 

264 base_image = Image.new("RGB", (200, 200), color="blue") 

265 overlay_image = Image.new("RGBA", (100, 100), color="pink") 

266 

267 out_image = overlay_image_on_image( 

268 base_image, 

269 overlay_image, 

270 position=("bottom", "left") 

271 ) 

272 assert out_image is not None 

273 assert out_image.size == base_image.size 

274 

275 # Corner case: overlay larger than base 

276 out_image = overlay_image_on_image( 

277 overlay_image, 

278 base_image, 

279 position=("top", "center") 

280 ) 

281 assert out_image is not None 

282 assert out_image.size == overlay_image.size 

283 

284 # More positions 

285 for pos_vertical in ["top", "center", "bottom"]: 

286 for pos_horizontal in ["left", "center", "right"]: 

287 out_image = overlay_image_on_image( 

288 base_image, 

289 overlay_image, 

290 position=(pos_vertical, pos_horizontal) 

291 ) 

292 assert out_image is not None 

293 assert out_image.size == base_image.size