22 lines
1.0 KiB
Python
22 lines
1.0 KiB
Python
import providers.flux as fx
|
|
|
|
def test_blocked_during_trading_hours(monkeypatch):
|
|
monkeypatch.setenv("FLUX_BLOCK_TRADING_HOURS", "1")
|
|
monkeypatch.setattr(fx, "_is_trading_hours", lambda: True)
|
|
calls = []
|
|
monkeypatch.setattr(fx, "webhook_update_task", lambda *a, **k: calls.append((a, k)))
|
|
fx.run_flux_generation("t1", {"prompt": "a cat"})
|
|
assert calls[-1][0][1] == "failed"
|
|
assert "장중" in calls[-1][1]["error"]
|
|
|
|
def test_success_polls_history_and_saves(monkeypatch):
|
|
monkeypatch.setattr(fx, "_is_trading_hours", lambda: False)
|
|
calls = []
|
|
monkeypatch.setattr(fx, "webhook_update_task", lambda *a, **k: calls.append((a, k)))
|
|
monkeypatch.setattr(fx, "_load_workflow", lambda prompt, size: {"3": {}})
|
|
monkeypatch.setattr(fx, "_submit_prompt", lambda wf: "pid-1")
|
|
monkeypatch.setattr(fx, "_poll_image_b64", lambda pid: "ZmFrZQ==")
|
|
monkeypatch.setattr(fx, "save_b64_png", lambda tid, b64: "/media/image/t1.png")
|
|
fx.run_flux_generation("t1", {"prompt": "a cat"})
|
|
assert [c for c in calls if c[0][1] == "succeeded"]
|