48 lines
1.8 KiB
Python
48 lines
1.8 KiB
Python
import importlib.util
|
|
import os
|
|
import pathlib
|
|
import unittest
|
|
|
|
|
|
os.environ.setdefault("WORKER_TOKEN", "x" * 48)
|
|
PATH = (pathlib.Path(__file__).parents[1]
|
|
/ "platform/docker/qwen-image-worker/qwen_image_worker.py")
|
|
SPEC = importlib.util.spec_from_file_location("qwen_image_worker", PATH)
|
|
worker = importlib.util.module_from_spec(SPEC)
|
|
assert SPEC and SPEC.loader
|
|
SPEC.loader.exec_module(worker)
|
|
|
|
|
|
class QwenImageWorkerTests(unittest.TestCase):
|
|
def test_text_to_image_uses_empty_latent(self):
|
|
workflow = worker.make_workflow("test", 7, 25, 1024, 1024, [], "job")
|
|
self.assertEqual(workflow["6"]["inputs"]["latent_image"], ["5", 0])
|
|
self.assertIn("5", workflow)
|
|
self.assertNotIn("vae", workflow["4"]["inputs"])
|
|
|
|
def test_edit_uses_reference_latent_and_qwen_dynamic_inputs(self):
|
|
workflow = worker.make_workflow(
|
|
"edit <image1>", 7, 25, 1024, 1024,
|
|
["job-ref-1.png", "job-ref-2.jpg"], "job")
|
|
encode = workflow["4"]["inputs"]
|
|
self.assertEqual(workflow["6"]["inputs"]["latent_image"], ["4", 2])
|
|
self.assertEqual(encode["vae"], ["3", 0])
|
|
self.assertEqual(encode["images.image_1"], ["9", 0])
|
|
self.assertEqual(encode["images.image_2"], ["10", 0])
|
|
self.assertNotIn("5", workflow)
|
|
|
|
def test_router_contract_is_fixed_to_production_parameters(self):
|
|
request = {
|
|
"prompt": "test", "filename": "result.png", "width": 1024,
|
|
"height": 1024, "steps": 25, "guidance": 1.0, "seed": 42,
|
|
}
|
|
self.assertEqual(worker.validate_request(request)[1:],
|
|
("result.png", 1024, 1024, 25, 1.0, 42))
|
|
request["steps"] = 4
|
|
with self.assertRaisesRegex(ValueError, "steps=25"):
|
|
worker.validate_request(request)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|