Spaces:
Running on Zero
Running on Zero
| """Lifecycle contracts only; no model, scheduler or simulated inference claim.""" | |
| from contextlib import contextmanager | |
| from contextvars import ContextVar | |
| from pathlib import Path | |
| import threading | |
| from types import SimpleNamespace | |
| import pytest | |
| from gradio.context import LocalContext | |
| import workflow_artifacts as artifacts | |
| import workflow_execution as execution | |
| def arguments(path): | |
| return (["A", "A"], ["segment1", "segment2"], [90, 90], [0, 5], [90, 85], | |
| 512, 512, [None, None], 42, ["array1", "array2"], ["anchor1", "anchor2"], str(path)) | |
| def queued_context(): | |
| event = SimpleNamespace(alive=True) | |
| queue = SimpleNamespace(event_ids_to_events={"event": event}) | |
| values = {"request": object(), "event_id": "event", | |
| "blocks": SimpleNamespace(_queue=queue)} | |
| tokens = [(getattr(LocalContext, k), getattr(LocalContext, k).set(v)) for k, v in values.items()] | |
| try: | |
| yield values, event | |
| finally: | |
| for var, token in reversed(tokens): | |
| var.reset(token) | |
| def test_context_and_all_windows_preserved_in_one_call(tmp_path): | |
| calls = [] | |
| marker = ContextVar("marker", default="missing") | |
| marker.set("inherited") | |
| with queued_context() as (values, event): | |
| def render(*args): | |
| assert marker.get() == "inherited" | |
| for key, value in values.items(): | |
| assert getattr(LocalContext, key).get(None) is value | |
| calls.append(args) | |
| return "unaltered-result" | |
| bridge = execution.GPUBridge(render, lambda *args: 1) | |
| args = arguments(tmp_path / "progress.json") | |
| assert bridge(*args, None) == "unaltered-result" | |
| assert calls == [args] | |
| bridge.ensure_idle() | |
| def test_error_propagation_and_no_false_success(tmp_path): | |
| error = RuntimeError("original scheduler/model failure") | |
| def fail(*args): | |
| raise error | |
| progress = [] | |
| bridge = execution.GPUBridge(fail, lambda *args: 1) | |
| with pytest.raises(RuntimeError) as raised: | |
| bridge(*arguments(tmp_path / "p.json"), lambda *a, **kw: progress.append(kw)) | |
| assert raised.value is error | |
| assert not any(p.get("desc") == "Viggle — done" for p in progress) | |
| bridge.ensure_idle() | |
| def test_timeout_blocks_new_jobs_until_real_worker_returns(tmp_path, monkeypatch): | |
| release = threading.Event() | |
| def held(*args): | |
| assert release.wait(2), "test worker release missing" | |
| return "late result" | |
| bridge = execution.GPUBridge(held, lambda *a: .03, grace_seconds=0, poll_seconds=.005) | |
| monkeypatch.setattr(execution, "_bridge", bridge) | |
| try: | |
| with pytest.raises(TimeoutError, match="waiting_worker"): | |
| artifacts.cached("timeout_test", [], lambda: ((bridge(*arguments(tmp_path / "p.json"), None),), [])) | |
| with pytest.raises(execution.ExecutionBusy): | |
| artifacts.cached("other_model_stage", [], lambda: pytest.fail("model overlap")) | |
| assert not (artifacts.STATE / "cache" / (artifacts.digest(["timeout_test", []]) + ".json")).exists() | |
| finally: | |
| done = bridge.pending | |
| release.set() | |
| if done: | |
| assert done.wait(2) | |
| assert artifacts.cached("timeout_test", [], lambda: (("retry after worker exit",), [])) == ("retry after worker exit",) | |
| def test_cancelled_result_never_commits_after_cpu_stage(tmp_path): | |
| with queued_context() as (values, event): | |
| def cancelled_compute(): | |
| values["blocks"]._queue.event_ids_to_events.clear() | |
| return ("cancelled-result",), [] | |
| with pytest.raises(execution.ExecutionCancelled): | |
| artifacts.cached("cancelled_cpu_test", [], cancelled_compute) | |
| assert artifacts.cached("cancelled_cpu_test", [], lambda: (("retry",), [])) == ("retry",) | |
| def test_waiting_progress_does_not_claim_denoising(tmp_path): | |
| release = threading.Event() | |
| progress = [] | |
| def worker(*args): | |
| assert release.wait(2) | |
| return "ok" | |
| def update(*args, **kw): | |
| progress.append(kw["desc"]) | |
| release.set() | |
| bridge = execution.GPUBridge(worker, lambda *a: 1) | |
| try: | |
| assert bridge(*arguments(tmp_path / "absent.json"), update) == "ok" | |
| finally: | |
| release.set() | |
| assert "model entry not confirmed" in progress[0] | |
| assert not any("denoise" in p for p in progress) | |
| def test_native_stage_lock_rejects_instead_of_waiting(): | |
| held, release = threading.Event(), threading.Event() | |
| def occupy(): | |
| with artifacts.EXECUTION_LOCK: | |
| held.set() | |
| release.wait(2) | |
| thread = threading.Thread(target=occupy) | |
| thread.start() | |
| try: | |
| assert held.wait(2) | |
| with pytest.raises(execution.ExecutionBusy, match="Another V22 stage"): | |
| artifacts.cached("lock_test", [], lambda: pytest.fail("must not run")) | |
| finally: | |
| release.set() | |
| thread.join(2) | |