Coverage for tests/streamwise_app/test_streamanimate.py: 94%

1#!/usr/bin/env python3 

2""" 

3Unit tests for StreamAnimate. 

4""" 

5 

6import os 

7import sys 

8import pytest 

9 

10from http import HTTPStatus 

11 

12from PIL import Image # noqa: F401 - import before patch.dict to keep PIL in sys.modules 

13 

14from quart import Quart 

15 

16from unittest.mock import patch 

17from unittest.mock import MagicMock 

18 

19# Add current path 

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

21 

22from tests.test_utils import temp_sys_path 

23from tests.torch_mock import TorchMock 

24from tests.streamwise_app.app_test_helpers import check_app_root 

25from tests.streamwise_app.app_test_helpers import check_health 

26from tests.streamwise_app.app_test_helpers import check_files 

27from tests.streamwise_app.app_test_helpers import check_unknown_route 

28from tests.streamwise_app.app_test_helpers import check_job_submit_page 

29from tests.streamwise_app.app_test_helpers import check_job_status_page 

30from tests.streamwise_app.app_test_helpers import check_api_job_status 

31from tests.streamwise_app.app_test_helpers import check_api_job_requests 

32 

33mock_torch = TorchMock() 

34 

35mock_modules = {} 

36mock_modules.update(mock_torch.get_sub_modules()) 

37 

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

39 with temp_sys_path("apps", "apps/streamanimate"): 

40 from apps.streamanimate.streamanimate import StreamAnimateApp 

41 from apps.streamanimate.streamanimate_job import StreamAnimateJob 

42 from apps.streamanimate.streamanimate_job import JobStatus 

43 from apps.streamanimate.animate_prompts import IMG_PROMPT 

44 from apps.streamanimate.animate_prompts import VIDEO_PROMPT 

45 from tests.streamwise_app.lmm_generator_mock import LMMGeneratorMock 

46 

47 

48streamanimate_app = StreamAnimateApp() 

49 

50 

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

52def _test_app() -> Quart: 

53 return streamanimate_app.app 

54 

55 

56@pytest.mark.asyncio 

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

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

59 await check_app_root(test_app, "StreamAnimate") 

60 

61 

62@pytest.mark.asyncio 

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

64 """Check /health.""" 

65 await check_health(test_app) 

66 

67 

68@pytest.mark.asyncio 

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

70 """Check /files endpoint.""" 

71 await check_files(test_app, "streamanimate") 

72 

73 

74@pytest.mark.asyncio 

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

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

77 await check_unknown_route(test_app) 

78 

79 

80@pytest.mark.asyncio 

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

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

83 await check_job_submit_page(test_app) 

84 

85 

86@pytest.mark.asyncio 

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

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

89 await check_job_status_page(test_app) 

90 

91 

92@pytest.mark.asyncio 

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

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

95 await check_api_job_status(test_app) 

96 

97 

98@pytest.mark.asyncio 

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

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

101 await check_api_job_requests(test_app) 

102 

103 

104@pytest.mark.asyncio 

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

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

107 client = test_app.test_client() 

108 

109 response = await client.post("/api/job", json={"text_prompt": "A bird flying"}) 

110 assert response.status_code == HTTPStatus.BAD_REQUEST 

111 response_json = await response.get_json() 

112 assert "error" in response_json 

113 assert response_json["error"] == "Service manager not initialized" 

114 

115 # Mock the service manager 

116 streamanimate_app.service_manager = MagicMock() 

117 streamanimate_app.service_manager.get_service_url = MagicMock( 

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

119 ) 

120 

121 response = await client.post("/api/job", json={"text_prompt": "A bird flying over mountains"}) 

122 # The job is accepted; it may fail later if external services are unavailable. 

123 response_json = await response.get_json() 

124 assert response_json is not None 

125 assert "status" in response_json 

126 # If the image generation service responds before 0.1s we may get an error; 

127 # if the job is still running after 0.1s we get job_id back. 

128 if response.status_code == HTTPStatus.OK: 

129 assert "job_id" in response_json 

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

131 else: 

132 assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR 

133 assert "error" in response_json 

134 

135 

136def test_animate_prompts() -> None: 

137 """Test that animate prompts are defined and non-empty.""" 

138 assert IMG_PROMPT 

139 assert len(IMG_PROMPT) > 0 

140 assert VIDEO_PROMPT 

141 assert "animation" in VIDEO_PROMPT.lower() or "motion" in VIDEO_PROMPT.lower() 

142 

143 

144@pytest.mark.asyncio 

145async def test_gen_animate_no_video_no_audio() -> None: 

146 """StreamAnimateJob.gen_animate raises ValueError when no video/audio generated.""" 

147 service_manager = MagicMock() 

148 job = StreamAnimateJob( 

149 job_id="test_gen_animate_no_video_no_audio", 

150 service_manager=service_manager, 

151 ) 

152 job.gen = LMMGeneratorMock() 

153 job.config["output_mode"] = "audio_only" 

154 # narration_text is empty and audio_only => neither video nor audio generated 

155 with pytest.raises(ValueError, match="Neither video nor audio was generated"): 

156 await job.gen_animate( 

157 image_base64=None, 

158 text_prompt="A bird flying", 

159 narration_text="", 

160 ) 

161 job_status = await job.get_status() 

162 assert job_status == JobStatus.FAILED 

163 

164 del job 

165 del service_manager 

166 

167 

168@pytest.mark.asyncio 

169async def test_gen_animate_text_only() -> None: 

170 """StreamAnimateJob.gen_animate with text prompt generates video or fails gracefully.""" 

171 service_manager = MagicMock() 

172 job = StreamAnimateJob( 

173 job_id="test_gen_animate_text_only", 

174 service_manager=service_manager, 

175 ) 

176 job.gen = LMMGeneratorMock() 

177 job.config["output_mode"] = "video_audio_unsynced" 

178 

179 try: 

180 await job.gen_animate( 

181 image_base64=None, 

182 text_prompt="A bird flying over mountains", 

183 narration_text="", 

184 ) 

185 job_status = await job.get_status() 

186 assert job_status == JobStatus.COMPLETED 

187 except FileNotFoundError: 

188 # ffmpeg not available in this environment 

189 job_status = await job.get_status() 

190 assert job_status == JobStatus.FAILED 

191 

192 del job 

193 del service_manager 

194 

195 

196@pytest.mark.asyncio 

197async def test_gen_animate_with_narration() -> None: 

198 """StreamAnimateJob.gen_animate with narration generates video and audio or fails gracefully.""" 

199 service_manager = MagicMock() 

200 job = StreamAnimateJob( 

201 job_id="test_gen_animate_with_narration", 

202 service_manager=service_manager, 

203 ) 

204 job.gen = LMMGeneratorMock() 

205 job.config["output_mode"] = "video_audio_unsynced" 

206 

207 try: 

208 await job.gen_animate( 

209 image_base64=None, 

210 text_prompt="A sunset over the ocean", 

211 narration_text="Watch the beautiful sunset over the ocean.", 

212 ) 

213 job_status = await job.get_status() 

214 assert job_status == JobStatus.COMPLETED 

215 except FileNotFoundError: 

216 # ffmpeg not available in this environment 

217 job_status = await job.get_status() 

218 assert job_status == JobStatus.FAILED 

219 

220 del job 

221 del service_manager