import base64 import json import struct import tempfile import unittest from unittest.mock import patch import httpx from fastapi.testclient import TestClient from bridge import VIDEO_MODELS, create_app, h3_request, video_workflows class BridgeTests(unittest.TestCase): def test_cancel_image_is_authenticated_owned_and_exact(self): self.assertEqual(self.client.post('/v1/images/task/unknown/cancel').status_code,401) self.assertEqual(self.client.post('/v1/images/task/unknown/cancel',headers=self.headers).status_code,404) video=self.client.post('/v2/video_generation',json=self.payload(),headers=self.headers).json()['task_id'] self.assertEqual(self.client.post('/v1/images/task/'+video+'/cancel',headers=self.headers).status_code,404) jid=self.client.post('/v1/images/generations',json=self.image_payload(),headers=self.headers).json()['task_id'] self.image_status='cancelled' result=self.client.post('/v1/images/task/'+jid+'/cancel',headers=self.headers) self.assertEqual(result.status_code,200) self.assertEqual(result.json()['task'],{'id':jid,'status':'cancelled','cancelRequested':False}) calls=[r for r in self.calls if r.url.path.endswith('/cancel')] self.assertEqual([r.url.path for r in calls],['/v1/jobs/'+jid+'/cancel']) polled=self.client.get('/v1/images/task/'+jid,headers=self.headers).json() self.assertTrue(polled['cancelled']);self.assertEqual(polled['task_id'],jid) def test_completed_image_is_not_reported_cancelled(self): jid=self.client.post('/v1/images/generations',json=self.image_payload(),headers=self.headers).json()['task_id'] self.image_status='succeeded' result=self.client.post('/v1/images/task/'+jid+'/cancel',headers=self.headers).json() self.assertEqual(result['task']['status'],'succeeded') polled=self.client.get('/v1/images/task/'+jid,headers=self.headers).json() self.assertEqual(polled['status'],'completed');self.assertEqual(polled['task_id'],jid) def test_local_worker_state_is_forwarded_without_claiming_failure(self): video=self.client.post('/v2/video_generation',json=self.payload(),headers=self.headers).json()['task_id'] self.client.post('/v1/images/generations',json=self.image_payload(),headers=self.headers) self.job_status=self.image_status='running' for state in ['queued','running','checking','missing']: self.worker_state=state task=self.client.get('/v2/query/video_generation/'+video,headers=self.headers).json()['task'] self.assertEqual(task['worker_state'],state) self.assertEqual(task['status'],'running') self.assertEqual(task['progress'],{'phase':'sampling','step':4,'totalSteps':20,'observedAt':123456}) image=self.client.get('/v1/images/task/image-job',headers=self.headers).json() self.assertEqual(image['worker_state'],state) self.assertEqual(image['status'],'processing') def test_cancel_requires_auth_and_bridge_owned_video(self): self.assertEqual(self.client.post('/v2/video_generation/unknown/cancel').status_code, 401) self.assertEqual(self.client.post('/v2/video_generation/unknown/cancel', headers=self.headers).status_code, 404) self.assertFalse(any('/cancel' in r.url.path for r in self.calls)) def test_cancel_targets_exact_owned_id(self): result = self.client.post('/v2/video_generation', json=self.payload(), headers=self.headers) jid = result.json()['task_id'] self.job_status = 'cancelled' result = self.client.post('/v2/video_generation/' + jid + '/cancel', headers=self.headers) self.assertEqual(result.status_code, 200, result.text) self.assertEqual(result.json()['task']['status'], 'cancelled') calls = [r for r in self.calls if '/cancel' in r.url.path] self.assertEqual(len(calls), 1) self.assertEqual(calls[0].url.path, '/v1/jobs/' + jid + '/cancel') def test_vdn_exact_hd_dimensions_and_seed(self): for resolution,landscape in [('2K',(2560,1440)),('4K',(3840,2160))]: for ratio in ('16:9','9:16'): payload={**self.payload(),'model':'MiniMax-H3-VDN-H3','resolution':resolution,'ratio':ratio,'seed':42} payload['content'].append({'type':'image_url','role':'reference_image','image_url':{'url':'data:image/png;base64,AA=='}}) request=h3_request(payload) self.assertEqual((request['width'],request['height']),landscape if ratio=='16:9' else landscape[::-1]) self.assertEqual(request['seed'],42) payload['resolution']='720P' with self.assertRaises(Exception): h3_request(payload) def test_explicit_gpu_routing_for_video_and_images(self): for gpu,worker in [('gpu1','gpu1_8189'),('gpu2','default'),('gpu3','gpu3')]: result=self.client.post('/v2/video_generation',json={**self.payload(),'gpu':gpu},headers=self.headers) self.assertEqual(result.status_code,200,result.text) sent=json.loads(next(r for r in reversed(self.calls) if r.url.path=='/v1/jobs').content) self.assertEqual(sent['provider_id'],worker) self.assertIs(sent['strict_provider'],True) self.assertIs(sent['huobao_panels'],True) self.assertEqual(result.json()['routing']['port'],{'gpu1':8189,'gpu2':8188,'gpu3':8191}[gpu]) image={'model':'qwen_image_2_1','prompt':'A teacup.','gpu':gpu,'size':'512x512'} result=self.client.post('/v1/images/generations',json=image,headers=self.headers) self.assertEqual(result.status_code,200,result.text) sent=json.loads(next(r for r in reversed(self.calls) if r.url.path=='/v1/jobs').content) self.assertEqual(sent['provider_id'],worker) self.assertIs(sent['strict_provider'],True) def test_invalid_gpu_never_submits(self): for gpu in ('gpu0','gpu4',None,False,[]): result=self.client.post('/v2/video_generation',json={**self.payload(),'gpu':gpu},headers=self.headers) self.assertEqual(result.status_code,400) self.assertFalse(any(r.url.path=='/v1/jobs' for r in self.calls)) def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.config = {"token": "test-secret", "state_dir": self.tmp.name, "public_url": "http://bridge.test", "h3": {"url": "http://h3.test", "token": "upstream-secret"}, "dify": {"qwen-dify": {"url": "http://dify.test/v1", "key": "dify-secret"}}} self.calls = [] self.job_status = "succeeded" self.dify_error = False self.qwen_available = True self.video_available = True self.image_status = "succeeded" self.image_outputs = [{"index": 0, "kind": "image", "mimeType": "image/png"}] original = httpx.AsyncClient self.patch = patch("bridge.httpx.AsyncClient", side_effect=lambda **kw: original(**kw, transport=httpx.MockTransport(self.handle))) self.patch.start() self.client = TestClient(create_app(self.config)) self.headers = {"Authorization": "Bearer test-secret"} def tearDown(self): self.patch.stop() self.tmp.cleanup() def handle(self, request): self.calls.append(request) if request.url.path == '/v1/huobao/route': return httpx.Response(200,json={'providerId':json.loads(request.content)['provider_id'],'ready':True}) if request.url.host == "dify.test": if self.dify_error: events = [{"event": "error", "message": "Quota exceeded"}] else: events = [{"event": "message", "answer": "Hello"}, {"event": "message", "answer": " world"}, {"event": "message_end", "metadata": {"usage": {"total_tokens": 8}}}] return httpx.Response(200, text="".join("data: " + json.dumps(e) + "\n\n" for e in events)) self.assertEqual(request.headers["Authorization"], "Bearer upstream-secret") if request.url.path == '/health': return httpx.Response(200,json={'autoBalance':True,'providers':2}) if request.url.path == '/v1/providers': return httpx.Response(200,json={'providers':[{'id':'default','name':'GPU2'},{'id':'gpu3','name':'GPU3'}]}) if request.url.path == "/v1/models": data = [{"id": k} for k in set().union(*(video_workflows(s) for s in VIDEO_MODELS.values()))] if self.video_available else [] if self.qwen_available: data.append({"id": "qwen_image_2_1", "model_type": "image"}) return httpx.Response(200, json={"data": data}) if request.url.path == "/v1/audio-plan": return httpx.Response(200, json={"status": "ready"}) if request.url.path == "/v1/jobs": data=json.loads(request.content) worker=data['provider_id'] if data.get('strict_provider') else 'gpu3' suffix='-'+worker if data.get('strict_provider') else '' if data["workflow_key"] == "qwen_image_2_1": return httpx.Response(200, json={"id": "image-job"+suffix, "status": "submitted", "providerId": worker}) return httpx.Response(200, json={"id": "test-job"+suffix, "status": "submitted", "providerId": worker}) if request.url.path in ("/v1/huobao/jobs/image-job", "/v1/jobs/image-job/cancel"): return httpx.Response(200, json={"status": self.image_status, "error": "image sampler failed", "outputs": self.image_outputs, "providerId": "gpu3", "worker_state":getattr(self,'worker_state','running')}) if request.url.path == "/v1/jobs/image-job/outputs/0": return httpx.Response(200, content=b"test-image-bytes", headers={"content-type": "image/png"}) if request.url.path in ("/v1/huobao/jobs/test-job", "/v1/jobs/test-job/cancel"): return httpx.Response(200, json={"status": self.job_status, "error": "quality gate failed", "providerId": "gpu3", "worker_state":getattr(self,'worker_state','running'), "progress":{'phase':'sampling','step':4,'totalSteps':20,'observedAt':123456}, "outputs": [{"index": 0, "kind": "image"}, {"index": 1, "kind": "video"}]}) if request.url.path == "/v1/jobs/test-job/outputs/1": return httpx.Response(200, content=b"test-video-bytes", headers={"content-type": "video/mp4"}) raise AssertionError(str(request.url)) def payload(self): return {"model": "MiniMax-H3", "content": [{"type": "text", "text": "A still mountain."}], "duration": 5, "resolution": "768P", "ratio": "9:16"} def test_auth_is_required(self): self.assertEqual(self.client.get("/v1/models").status_code, 401) self.assertEqual(self.calls, []) def test_empty_probe_cannot_submit(self): self.assertEqual(self.client.post("/v2/video_generation", json={}, headers=self.headers).status_code, 200) self.assertTrue(self.calls) self.assertTrue(all(r.url.path == "/v1/models" for r in self.calls)) def test_validation_does_not_submit_gpu_job(self): result = self.client.post("/v2/video_generation/validate", json=self.payload(), headers=self.headers) self.assertTrue(result.json()["valid"]) self.assertEqual(result.json()["width"], 768) self.assertGreater(result.json()["height"], 768) self.assertFalse(any(r.url.path == "/v1/jobs" for r in self.calls)) def test_video_submission_poll_and_download(self): result = self.client.post("/v2/video_generation", json=self.payload(), headers=self.headers) self.assertEqual(result.json()["task_id"], "test-job") self.assertEqual(result.json()["provider_id"], "gpu3") sent = json.loads(self.calls[-1].content) self.assertEqual(sent["workflow_key"], "minimax_h3_t2v_official") self.assertEqual(sent["duration"], 5) self.assertEqual(sent["provider_id"], "default") poll = self.client.get("/v2/query/video_generation/test-job", headers=self.headers).json() url = poll["task"]["content"]["url"] self.assertEqual(poll["task"]["provider_id"], "gpu3") self.assertIn("/video.mp4?", url) self.assertNotIn("test-secret", url) self.assertEqual(self.client.get(url).content, b"test-video-bytes") self.assertEqual(self.client.get(url + "x").status_code, 403) def test_every_video_mode_uses_automatic_pool(self): image = {"type": "image_url", "image_url": {"url": "data:image/png;base64,AA=="}} for model, spec in VIDEO_MODELS.items(): for mode in ("text", "first", "reference"): if mode not in spec: continue with self.subTest(model=model, mode=mode): payload = {**self.payload(), "model": model} if model == "MiniMax-H3-VDN-H3": payload["resolution"] = "2K" if mode != "text": payload["content"].append({**image, "role": "first_frame" if mode == "first" else "reference_image"}) request = h3_request(payload) self.assertEqual(request["workflow_key"], spec[mode]) self.assertEqual(request["provider_id"], "default") result = self.client.post("/v2/video_generation/validate", json=payload, headers=self.headers) self.assertEqual(result.status_code, 200) self.assertTrue(result.json()["valid"]) self.assertFalse(any(r.url.path == "/v1/jobs" for r in self.calls)) def test_invalid_video_modes_fail_before_submission(self): image = {"type": "image_url", "image_url": {"url": "data:image/png;base64,AA=="}, "role": "reference_image"} for model, refs in [("Wan2.2-I2V", []), ("Wan2.2-I2V", [image, image]), ("LTX-2.3", [image, image]), ("MiniMax-H3-R2V-4Step", []), ("MiniMax-H3-R2V-4Step", [{**image, "role": "first_frame"}]), ("unknown", [])]: payload = {**self.payload(), "model": model} payload["content"] += refs response = self.client.post("/v2/video_generation", json=payload, headers=self.headers) self.assertEqual(response.status_code, 400, response.text) self.assertFalse(self.calls) def test_unavailable_video_models_hidden_and_blocked(self): self.video_available = False models = self.client.get("/v1/models", headers=self.headers).json()["data"] self.assertFalse(set(m["id"] for m in models) & set(VIDEO_MODELS)) response = self.client.post("/v2/video_generation", json=self.payload(), headers=self.headers) self.assertEqual(response.status_code, 503) self.assertFalse(any(r.url.path == "/v1/jobs" for r in self.calls)) def test_unknown_jobs_are_rejected(self): self.assertEqual(self.client.get("/v2/query/video_generation/other", headers=self.headers).status_code, 404) self.assertFalse(self.calls) def test_failed_job_never_returns_partial_output(self): self.client.post("/v2/video_generation", json=self.payload(), headers=self.headers) self.job_status = "failed" task = self.client.get("/v2/query/video_generation/test-job", headers=self.headers).json()["task"] self.assertEqual(task["status"], "failed") self.assertNotIn("content", task) def test_image_modes_and_unsupported_reference(self): payload = self.payload() item = {"type": "image_url", "image_url": {"url": "data:image/png;base64,AA=="}, "role": "first_frame"} payload["content"].append(item) self.assertEqual(h3_request(payload)["workflow_key"], "minimax_h3_i2v") item["role"] = "reference_image" self.assertEqual(h3_request(payload)["workflow_key"], "minimax_h3_r2v") item["role"] = "last_frame" self.assertEqual(self.client.post("/v2/video_generation", json=payload, headers=self.headers).status_code, 400) self.assertFalse(self.calls) def test_first_frame_preserves_portrait_ratio(self): payload = self.payload() payload.pop("ratio") header = b"\x89PNG\r\n\x1a\n" + b"\x00" * 8 + struct.pack(">II", 720, 1280) url = "data:image/png;base64," + base64.b64encode(header).decode() item = {"type": "image_url", "image_url": {"url": url}, "role": "first_frame"} payload["content"].append(item) parsed = h3_request(payload) self.assertEqual(parsed["width"], 768) self.assertGreater(parsed["height"], parsed["width"]) payload["content"].append({**item, "role": "reference_image"}) self.assertEqual(self.client.post("/v2/video_generation", json=payload, headers=self.headers).status_code, 400) def test_dify_nonstream_preserves_instructions(self): body = {"model": "qwen-dify", "messages": [{"role": "system", "content": "Write a story"}, {"role": "user", "content": "About a cat"}]} result = self.client.post("/v1/chat/completions", json=body, headers=self.headers) self.assertEqual(result.json()["choices"][0]["message"]["content"], "Hello world") query = json.loads(self.calls[-1].content)["query"] self.assertIn("Write a story", query) self.assertIn("About a cat", query) self.assertEqual(result.json()["usage"]["total_tokens"], 8) def test_dify_stream(self): body = {"model": "qwen-dify", "messages": [{"role": "user", "content": "hi"}], "stream": True} result = self.client.post("/v1/chat/completions", json=body, headers=self.headers) self.assertIn('"finish_reason": "stop"', result.text) self.assertIn("data: [DONE]", result.text) def test_dify_errors_are_not_success(self): self.dify_error = True body = {"model": "qwen-dify", "messages": [{"role": "user", "content": "hi"}]} self.assertEqual(self.client.post("/v1/chat/completions", json=body, headers=self.headers).status_code, 502) body["stream"] = True stream = self.client.post("/v1/chat/completions", json=body, headers=self.headers).text self.assertIn("Quota exceeded", stream) self.assertNotIn('"finish_reason": "stop"', stream) def image_payload(self): return {"model": "qwen_image_2_1", "prompt": "A red ceramic cup", "size": "512x768", "n": 1, "response_format": "url"} def test_image_generation_poll_and_download(self): response = self.client.post("/v1/images/generations", json=self.image_payload(), headers=self.headers) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["task_id"], "image-job") submitted = json.loads(self.calls[-1].content) self.assertEqual(submitted["workflow_key"], "qwen_image_2_1") self.assertEqual((submitted["width"], submitted["height"]), (512, 768)) self.assertEqual(submitted["input_images"], []) result = self.client.get("/v1/images/task/image-job", headers=self.headers).json() self.assertEqual(result["status"], "completed") self.assertIn("/image.png?", result["image_url"]) image = self.client.get(result["image_url"]) self.assertEqual(image.headers["content-type"], "image/png") self.assertEqual(image.content, b"test-image-bytes") def test_image_edits_forward_all_references(self): image = b"\x89PNG\r\n\x1a\n" + b"\x00" * 8 + struct.pack(">II", 400, 600) files = [("image[]", ("ref1.png", image, "image/png")), ("image[]", ("ref2.png", image, "image/png"))] response = self.client.post("/v1/images/edits", data=self.image_payload(), files=files, headers=self.headers) self.assertEqual(response.status_code, 200, response.text) submitted = json.loads(self.calls[-1].content) self.assertEqual(len(submitted["input_images"]), 2) self.assertEqual(base64.b64decode(submitted["input_images"][0].split(",")[1]), image) self.assertEqual(submitted["prompt"], self.image_payload()["prompt"]) self.assertEqual(submitted['provider_id'],'default') self.assertEqual(response.json()['provider_id'],'gpu3') def test_image_pool_is_live_and_authenticated(self): self.assertEqual(self.client.get('/v1/images/pool').status_code,401) data=self.client.get('/v1/images/pool',headers=self.headers).json() self.assertTrue(data['auto_balance']) self.assertEqual([w['id'] for w in data['configured_workers']],['default','gpu3']) self.assertFalse(any(r.url.path=='/v1/jobs' for r in self.calls)) def test_image_edits_reject_invalid_or_excess_references(self): response = self.client.post("/v1/images/edits", data=self.image_payload(), files={"image[]": ("fake.png", b"not an image", "image/png")}, headers=self.headers) self.assertEqual(response.status_code, 400) image = b"\x89PNG\r\n\x1a\n" + b"\x00" * 8 + struct.pack(">II", 400, 600) files = [("image[]", (f"ref{i}.png", image, "image/png")) for i in range(7)] response = self.client.post("/v1/images/edits", data=self.image_payload(), files=files, headers=self.headers) self.assertEqual(response.status_code, 400) self.assertFalse(self.calls) def test_image_unavailable_does_not_submit(self): self.qwen_available = False response = self.client.post("/v1/images/generations", json=self.image_payload(), headers=self.headers) self.assertEqual(response.status_code, 503) self.assertTrue(all(r.url.path == "/v1/models" for r in self.calls)) def test_image_failed_or_pending_never_returns_partial_output(self): self.client.post("/v1/images/generations", json=self.image_payload(), headers=self.headers) for upstream, expected in [("failed", "failed"), ("cancelled", "failed"), ("running", "processing")]: self.image_status = upstream result = self.client.get("/v1/images/task/image-job", headers=self.headers).json() self.assertEqual(result["status"], expected) self.assertNotIn("image_url", result) def test_image_validation_and_invalid_requests_do_not_generate(self): self.assertEqual(self.client.post("/v1/images/generations", json=self.image_payload()).status_code, 401) self.assertEqual(self.client.post("/v1/images/validate", json=self.image_payload(), headers=self.headers).status_code, 200) for update in [{"n": 2}, {"size": "0x1024"}, {"size": "10000x10000"}, {"model": "MiniMax-H3"}, {"prompt": ""}]: result = self.client.post("/v1/images/generations", json={**self.image_payload(), **update}, headers=self.headers) self.assertEqual(result.status_code, 400) self.assertTrue(all(r.method == 'GET' and r.url.path in {'/v1/models', '/health', '/v1/providers'} for r in self.calls)) def test_image_job_cannot_be_polled_as_video(self): self.client.post("/v1/images/generations", json=self.image_payload(), headers=self.headers) self.assertEqual(self.client.get("/v2/query/video_generation/image-job", headers=self.headers).status_code, 404) self.assertEqual(self.client.get("/v1/images/task/unknown", headers=self.headers).status_code, 404) if __name__ == "__main__": unittest.main(verbosity=2)