371 lines
23 KiB
Python
371 lines
23 KiB
Python
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)
|