Coverage for tests/test_quart_utils.py: 100%

235 statements  

« prev     ^ index     » next       coverage.py v7.15.4, created at 2026-08-09 04:47 +0000

1import pytest 

2 

3from datetime import timedelta 

4 

5from quart_utils import json_pretty_filter 

6from quart_utils import format_datetime 

7from quart_utils import format_bytes 

8from quart_utils import format_string 

9from quart_utils import format_duration 

10from quart_utils import format_gpu_model 

11from quart_utils import format_gpu_model_mig 

12from quart_utils import format_url 

13from quart_utils import get_gpu_mem 

14from quart_utils import get_aspect_ratio 

15from quart_utils import get_class_emoji 

16from quart_utils import get_file_type_emoji 

17from quart_utils import get_file_type 

18from quart_utils import get_mime_type 

19from quart_utils import get_content_type_emoji 

20from quart_utils import get_friendly_region_name 

21from quart_utils import get_friendly_container_name 

22from quart_utils import is_rtgen_container 

23from quart_utils import get_docker_image 

24from quart_utils import parse_request_id 

25 

26 

27def test_json_pretty_filter() -> None: 

28 assert json_pretty_filter("{}") == "{}" 

29 assert json_pretty_filter("{'one': 1}") == "{'one': 1}" 

30 assert json_pretty_filter("{'two': [1, 2]}") == "{'two': [1, 2]}" 

31 assert json_pretty_filter("{'three': [1, 2, 'three']}") == "{'three': [1, 2, 'three']}" 

32 

33 

34def test_format_datetime() -> None: 

35 assert format_datetime(0) == "1970-01-01 00:00:00" 

36 assert format_datetime(1672531199) == "2022-12-31 23:59:59" 

37 assert format_datetime("abc") == "Invalid date" # type: ignore[arg-type] 

38 

39 

40def test_format_bytes() -> None: 

41 assert format_bytes(0) == '<span class="text-muted">-</span>' 

42 assert format_bytes(500) == "500 B" 

43 assert format_bytes(1024) == "1.0 KiB" 

44 assert format_bytes(1536) == "1.5 KiB" 

45 assert format_bytes(1048576) == "1.0 MiB" 

46 assert format_bytes(1073741824) == "1.0 GiB" 

47 assert format_bytes(1099511627776) == "1.0 TiB" 

48 assert format_bytes(1125899906842624) == "1024.0 TiB" 

49 

50 

51def test_format_string() -> None: 

52 assert format_string("hello_world") == "Hello World" 

53 assert format_string("Test_String_Example") == "Test String Example" 

54 assert format_string("") == "" 

55 assert format_string("no_underscores") == "No Underscores" 

56 assert format_string("multiple___underscores") == "Multiple Underscores" 

57 assert format_string(None) is None 

58 assert format_string(123) == 123 # type: ignore[arg-type] 

59 

60 

61def test_format_duration() -> None: 

62 assert format_duration(None) == "0" 

63 assert format_duration(timedelta(seconds=0)) == "0" 

64 assert format_duration(timedelta(seconds=47)) == "47 seconds" 

65 assert format_duration(timedelta(minutes=2, seconds=30)) == "2 minutes 30 seconds" 

66 assert format_duration(timedelta(hours=1, minutes=15, seconds=5)) == "1 hour 15 minutes 5 seconds" 

67 assert format_duration(timedelta(hours=0, minutes=0, seconds=5)) == "5 seconds" 

68 assert format_duration(timedelta(hours=0, minutes=3, seconds=0)) == "3 minutes 0 seconds" 

69 assert format_duration(timedelta(hours=2, minutes=0, seconds=0)) == "2 hours 0 minutes 0 seconds" 

70 

71 

72def test_format_url() -> None: 

73 assert format_url(None) is None 

74 assert format_url("http://example.com") == "example.com" 

75 assert format_url("https://10.0.0.1:8080") == "10.0.0.1:8080" 

76 

77 

78def test_format_gpu_model() -> None: 

79 # Raw NVIDIA GPU model strings 

80 assert format_gpu_model("NVIDIA A100-SXM4-80GB") == "A100 80GB" 

81 assert format_gpu_model("NVIDIA-A100-SXM4-80GB") == "A100 80GB" 

82 assert format_gpu_model("NVIDIA-A100-80GB-PCIe") == "A100 80GB" 

83 assert format_gpu_model("NVIDIA A100 80GB PCIe") == "A100 80GB" 

84 assert format_gpu_model("NVIDIA-H200") == "H200" 

85 assert format_gpu_model("NVIDIA H200") == "H200" 

86 assert format_gpu_model("NVIDIA-H100") == "H100" 

87 assert format_gpu_model("NVIDIA H100") == "H100" 

88 assert format_gpu_model("NVIDIA-H100-NVL") == "H100 NVL" 

89 assert format_gpu_model("NVIDIA H100 NVL") == "H100 NVL" 

90 assert format_gpu_model("NVIDIA-H100-80GB-HBM3") == "H100" 

91 assert format_gpu_model("NVIDIA H100 80GB HBM3") == "H100" 

92 assert format_gpu_model("Tesla-V100-PCIE-16GB") == "V100 16GB" 

93 assert format_gpu_model("Tesla V100-PCIE-16GB") == "V100 16GB" 

94 assert format_gpu_model("Tesla-V100-SXM2-32GB") == "V100 32GB" 

95 assert format_gpu_model("Tesla V100-SXM2-32GB") == "V100 32GB" 

96 # Azure VM SKU names — ND series (ND_A100_v4) 

97 assert format_gpu_model("Standard_ND96ams_A100_v4") == "A100 80GB" 

98 assert format_gpu_model("Standard_ND96amsr_A100_v4") == "A100 80GB" 

99 # Azure VM SKU names — ND series (ND_H100_v5) 

100 assert format_gpu_model("Standard_ND96isrf_H100_v5") == "H100" 

101 assert format_gpu_model("Standard_ND96isr_H100_v5") == "H100" 

102 # Azure VM SKU names — ND series (ND_MI300X_v5) 

103 assert format_gpu_model("Standard_ND96isr_MI300X_v5") == "MI300X" 

104 # Azure VM SKU names — ND series (ND_H200_v5) 

105 assert format_gpu_model("Standard_ND96isr_H200_v5") == "H200" 

106 # Azure VM SKU names — ND series (ND_GB200_v6 / ND_GB300_v6) 

107 assert format_gpu_model("Standard_ND128isr_GB300_v6") == "GB300" 

108 # Azure VM SKU names — NC series (NC_A100_v4) 

109 assert format_gpu_model("Standard_NC96ads_A100_v4") == "A100 80GB" 

110 # Azure VM SKU names — NC series (NCasT4_v3) 

111 assert format_gpu_model("Standard_NC4as_T4_v3") == "T4" 

112 assert format_gpu_model("Standard_NC8as_T4_v3") == "T4" 

113 assert format_gpu_model("Standard_NC64as_T4_v3") == "T4" 

114 # Azure VM SKU names — NC series (NCads_H100_v5) 

115 assert format_gpu_model("Standard_NC40ads_H100_v5") == "H100" 

116 # Azure VM SKU names — NV series (NVads_A10_v5) 

117 assert format_gpu_model("Standard_NV18ads_A10_v5") == "A10" 

118 assert format_gpu_model("Standard_NV36ads_A10_v5") == "A10" 

119 assert format_gpu_model("Standard_NV72ads_A10_v5") == "A10" 

120 # GB200 and GB300 raw strings 

121 assert format_gpu_model("NVIDIA GB200") == "GB200" 

122 assert format_gpu_model("NVIDIA-GB200") == "GB200" 

123 assert format_gpu_model("NVIDIA GB300") == "GB300" 

124 assert format_gpu_model("NVIDIA-GB300") == "GB300" 

125 # Azure VM SKU — unknown GPU identifier falls back to the GPU part (uppercased) 

126 assert format_gpu_model("Standard_NC6s_UnknownGPU_v3") == "UNKNOWNGPU" 

127 # Edge cases 

128 assert format_gpu_model("Some Other GPU") == "Some Other GPU" 

129 assert format_gpu_model("") == "" 

130 assert format_gpu_model(None) is None 

131 assert format_gpu_model(123) == 123 # type: ignore[arg-type] 

132 # MIG instance names: preserve the MIG profile alongside the formatted GPU name 

133 # assert format_gpu_model("NVIDIA A100-SXM4-80GB MIG 1g.10gb") == "A100 80GB MIG 1g.10gb" 

134 # assert format_gpu_model("NVIDIA A100-SXM4-40GB MIG 1g.5gb") == "A100 40GB MIG 1g.5gb" 

135 # assert format_gpu_model("NVIDIA A100-SXM4-80GB MIG 2g.20gb") == "A100 80GB MIG 2g.20gb" 

136 # assert format_gpu_model("NVIDIA H100-SXM5-80GB MIG 3g.40gb") == "H100 MIG 3g.40gb" 

137 assert format_gpu_model("NVIDIA A100-SXM4-80GB MIG 1g.10gb") == "A100 80GB" 

138 assert format_gpu_model("NVIDIA A100-SXM4-40GB MIG 1g.5gb") == "A100 40GB" 

139 assert format_gpu_model("NVIDIA A100-SXM4-80GB MIG 2g.20gb") == "A100 80GB" 

140 assert format_gpu_model("NVIDIA H100-SXM5-80GB MIG 3g.40gb") == "H100" 

141 

142 

143def test_format_gpu_model_mig() -> None: 

144 assert format_gpu_model_mig("A100 80GB", "1g.10gb") == "⅛ A100 80GB" 

145 assert format_gpu_model_mig("A100 80GB", None) == "A100 80GB" 

146 assert format_gpu_model_mig("A100 40GB", "2g.10gb") == "¼ A100 40GB" 

147 

148 

149def test_get_gpu_mem() -> None: 

150 assert get_gpu_mem("NVIDIA A100-SXM4-80GB") == 80 

151 assert get_gpu_mem("NVIDIA A100-SXM4-40GB") == 40 

152 assert get_gpu_mem("NVIDIA-GB200") == 196 

153 assert get_gpu_mem("Standard_ND96isr_H200_v5") == 141 

154 

155 

156def test_get_aspect_ratio() -> None: 

157 assert get_aspect_ratio(1) == "1:1" 

158 assert get_aspect_ratio(16 / 9) == "16:9" 

159 assert get_aspect_ratio(1280 / 720) == "16:9" 

160 assert get_aspect_ratio(1.77) == "16:9" 

161 assert get_aspect_ratio(16 / 10) == "16:10" 

162 assert get_aspect_ratio(1280 / 800) == "16:10" 

163 assert get_aspect_ratio(4 / 3) == "4:3" 

164 assert get_aspect_ratio(5 / 4) == "5:4" 

165 assert get_aspect_ratio(1280 / 1024) == "5:4" 

166 assert get_aspect_ratio(3 / 2) == "3:2" 

167 assert get_aspect_ratio(2 / 1) == "2:1" 

168 assert get_aspect_ratio(0) == "0.00:1" 

169 assert get_aspect_ratio(-1) == "-1.00:1" 

170 

171 

172@pytest.mark.asyncio 

173async def test_get_class_emoji() -> None: 

174 assert await get_class_emoji(None) == "" 

175 assert await get_class_emoji("flux") == "📄→🖼️" 

176 assert await get_class_emoji("hunyuanframepackf1") == "📄🖼️→🎬" 

177 assert await get_class_emoji("fantasytalking") == "📄🖼️🔉→🎬" 

178 assert await get_class_emoji("hunyuanframepackvae") == "🔢→🎬" 

179 assert await get_class_emoji("wan") == "📄🖼️→🎬" 

180 assert await get_class_emoji("kokoro") == "📄→🔉" 

181 assert await get_class_emoji("thinksound") == "🎬→🔉" 

182 assert await get_class_emoji("yolo") == "🖼️→🖼️" 

183 assert await get_class_emoji("mola") == "<span class='text-muted' title='mola'>❓</span>" 

184 assert await get_class_emoji("podcasttranscript") == "📄→📄" 

185 assert await get_class_emoji("gemma") == "📄→📄" 

186 

187 

188def test_get_file_type_emoji() -> None: 

189 assert get_file_type_emoji("image") == "🖼️" 

190 assert get_file_type_emoji("kernel") == "📊" 

191 assert get_file_type_emoji("tensor") == "📊" 

192 assert get_file_type_emoji("directory") == "📁" 

193 assert get_file_type_emoji("text") == "📄" 

194 assert get_file_type_emoji("archive") == "📦" 

195 assert get_file_type_emoji("audio") == "🎵" 

196 assert get_file_type_emoji("video") == "🎥" 

197 assert get_file_type_emoji("mola") == "❓ mola" 

198 assert get_file_type_emoji(None) == "❓" 

199 

200 

201def test_get_file_type() -> None: 

202 """Check the file type detection.""" 

203 assert get_file_type("example.png") == "image" 

204 assert get_file_type("example.pdf") == "pdf" 

205 assert get_file_type("example.jsonl") == "jsonl" 

206 assert get_file_type("image.jpg") == "image" 

207 assert get_file_type("image.png") == "image" 

208 assert get_file_type("video.mp4") == "video" 

209 assert get_file_type("audio.mp3") == "audio" 

210 assert get_file_type("audio.wav") == "audio" 

211 assert get_file_type("document.pdf") == "pdf" 

212 assert get_file_type("document.pt") == "tensor" 

213 assert get_file_type("document.ptx") == "kernel" 

214 assert get_file_type("document.json") == "json" 

215 assert get_file_type("document.jsonl") == "jsonl" 

216 assert get_file_type("document.txt") == "text" 

217 assert get_file_type("unknown.xyz") == "unknown" 

218 

219 

220def test_get_mime_type() -> None: 

221 """Check the MIME type detection.""" 

222 assert get_mime_type("example.png") == "image/png" 

223 assert get_mime_type("example.mp4") == "video/mp4" 

224 assert get_mime_type("example.wav") == "audio/x-wav" 

225 assert get_mime_type("image.jpg") == "image/jpeg" 

226 assert get_mime_type("image.png") == "image/png" 

227 assert get_mime_type("video.mp4") == "video/mp4" 

228 assert get_mime_type("audio.mp3") == "audio/mpeg" 

229 assert get_mime_type("audio.wav") == "audio/x-wav" 

230 assert get_mime_type("log.log") == "text/plain" 

231 assert get_mime_type("document.pdf") == "application/pdf" 

232 # assert get_mime_type("document.pt") == "application/octet-stream" 

233 # assert get_mime_type("document.ptx") == "application/octet-stream" 

234 assert get_mime_type("document.json") == "application/json" 

235 assert get_mime_type("document.jsonl") == "application/x-ndjson" 

236 assert get_mime_type("document.txt") == "text/plain" 

237 

238 

239def test_get_content_type_emoji() -> None: 

240 assert get_content_type_emoji("image/png") == "🖼️" 

241 assert get_content_type_emoji("video/mp4") == "🎥" 

242 assert get_content_type_emoji("text/plain") == "📄" 

243 assert get_content_type_emoji("application/json") == "📄" 

244 assert get_content_type_emoji("application/octet-stream") == "📦" 

245 assert get_content_type_emoji("application/mola") == "❓ application/mola" 

246 

247 

248def test_get_friendly_region_name() -> None: 

249 assert get_friendly_region_name("eastus2") == "East US 2" 

250 assert get_friendly_region_name("southeastasia") == "Southeast Asia" 

251 assert get_friendly_region_name("eastasia") == "East Asia" 

252 assert get_friendly_region_name("centralus") == "Central US" 

253 assert get_friendly_region_name("mola") == "Mola" 

254 assert get_friendly_region_name(None) == "N/A" 

255 

256 

257@pytest.mark.asyncio 

258async def test_get_friendly_container_name() -> None: 

259 assert await get_friendly_container_name("flux") == "FLUX" 

260 assert await get_friendly_container_name("fantasytalking") == "Fantasy Talking" 

261 assert await get_friendly_container_name("qwenimage") == "Qwen Image" 

262 assert await get_friendly_container_name("mola") == "mola" 

263 

264 

265@pytest.mark.asyncio 

266async def test_is_rtgen_container() -> None: 

267 assert await is_rtgen_container("flux") is True 

268 assert await is_rtgen_container("qwenimage") is True 

269 assert await is_rtgen_container("nonexistent") is False 

270 

271 

272@pytest.mark.asyncio 

273async def test_get_docker_image() -> None: 

274 docker_image = await get_docker_image("flux") 

275 assert docker_image is not None 

276 assert "/flux:v" in docker_image 

277 

278 docker_image = await get_docker_image("llamagen") 

279 assert docker_image is not None 

280 assert "/llamagen:v" in docker_image 

281 

282 docker_image = await get_docker_image("nonexistent") 

283 assert docker_image is None 

284 

285 

286@pytest.mark.asyncio 

287async def test_get_docker_image_custom_tag() -> None: 

288 docker_image = await get_docker_image("flux", tag="custom-tag") 

289 assert docker_image is not None 

290 assert docker_image.endswith(":custom-tag") 

291 assert "/flux:custom-tag" in docker_image 

292 

293 docker_image = await get_docker_image("flux", tag="v9.9.9") 

294 assert docker_image is not None 

295 assert docker_image.endswith(":v9.9.9") 

296 

297 # Custom tag on nonexistent container should still return None 

298 docker_image = await get_docker_image("nonexistent", tag="custom-tag") 

299 assert docker_image is None 

300 

301 

302def test_parse_request_id() -> None: 

303 assert parse_request_id("20250904T010335296_006_001_fantasytalking") == { 

304 'job_id': '20250904T010335296', 

305 'scene_id': '006', 

306 'service_name': 'fantasytalking', 

307 'sub_scene_id': '001', 

308 "task_id": "", 

309 } 

310 

311 assert parse_request_id("20250904T010335296_flux") == { 

312 'job_id': '20250904T010335296', 

313 'scene_id': "", 

314 'service_name': 'flux', 

315 'sub_scene_id': "", 

316 "task_id": "", 

317 } 

318 

319 assert parse_request_id("20260105T194652416_main_image_flux") == { 

320 'job_id': '20260105T194652416', 

321 'scene_id': "", 

322 'service_name': "flux", 

323 'sub_scene_id': "", 

324 "task_id": "main_image" 

325 } 

326 

327 assert parse_request_id("") == {}