Spaces:
Running on Zero
Running on Zero
| """Actual FastAPI/Gradio routes; CPU media only, no simulated inference.""" | |
| import copy | |
| import json | |
| from pathlib import Path | |
| import pytest | |
| from fastapi.testclient import TestClient | |
| import workflow_artifacts as a | |
| from cpu_preview import cpu_core | |
| from workflow_ui import create_workflow | |
| from workflow_nodes import Nodes | |
| from workflow_media import output_file | |
| def test_gpu_registration_does_not_duplicate_canvas_load_events_or_app_routes(): | |
| # Construction probe only: never substitutes GPU inference in production. | |
| base, base_app = create_workflow(cpu_core()) | |
| core = cpu_core() | |
| def registration_probe(value: str): | |
| return value | |
| core.render_all_segments = registration_probe | |
| registered, app = create_workflow(core, register_gpu=True) | |
| def load_count(demo): | |
| return sum(any(event == "load" for _, event in dep["targets"]) | |
| for dep in demo.get_config_file()["dependencies"]) | |
| try: | |
| assert load_count(registered) == load_count(base) | |
| assert len(registered.fns) == len(base.fns) + 1 | |
| # Launch will add these exactly once; custom routes are already present. | |
| assert not any(getattr(r, "path", None) == "/config" for r in app.routes) | |
| app = type(app).create_app(registered, app=app) | |
| assert sum(m.cls.__name__ == "BrotliMiddleware" for m in app.user_middleware) == 1 | |
| assert sum(m.cls.__name__ == "CustomCORSMiddleware" for m in app.user_middleware) == 1 | |
| with TestClient(app) as client: | |
| config = client.get("/config").json() | |
| ids = {c["id"] for c in config["components"]} | |
| assert ids <= set(app.get_blocks().blocks) | |
| canvas = next(c for c in config["components"] if c["type"] == "workflowcanvas") | |
| result = client.post("/gradio_api/component_server", json={ | |
| "component_id": canvas["id"], "fn_name": "get_workflow_key", | |
| "data": {}, "session_hash": "startup-regression"}) | |
| assert result.status_code == 200 | |
| finally: | |
| registered.close() | |
| base.close() | |
| def client(): | |
| demo, app = create_workflow(cpu_core()) | |
| app = type(app).create_app(demo, app=app) | |
| # Production launch adds this path. TestClient does not call launch(). | |
| demo.allowed_paths = [str(a.PUBLIC)] | |
| with TestClient(app) as c: | |
| yield c | |
| demo.close() | |
| def test_native_config_routes_and_canvas_default(client): | |
| response = client.get("/", follow_redirects=False) | |
| assert response.status_code == 307 and "ui=canvas" in response.headers["location"] | |
| assert a.ID.fullmatch(client.cookies["vm_workspace"]) | |
| config = client.get("/config").json() | |
| assert config["version"] == "6.30.0" | |
| assert any(c["type"] == "workflowcanvas" for c in config["components"]) | |
| # Bound-function endpoints are intentionally undocumented; /info lists | |
| # public generated-pipeline endpoints, while config contains both. | |
| assert client.get("/gradio_api/info").status_code == 200 | |
| names = {d["api_name"] for d in config["dependencies"]} | |
| for name in Nodes(cpu_core()).bindings(): | |
| assert "predict_fn_" + name.replace(" ", "_") in names | |
| def test_canvas_redirect_keeps_browser_https_behind_hf_proxy(client): | |
| # Reproduce HF's proxy boundary: Python receives HTTP but the embedding | |
| # browser is on HTTPS. Never derive the redirect scheme from this scope. | |
| from urllib.parse import urljoin, parse_qs, urlsplit | |
| response = client.get("http://fffiloni-viggle-mine-workflow.hf.space/?__theme=light", | |
| follow_redirects=False) | |
| assert response.status_code == 307 | |
| location = response.headers["location"] | |
| assert location.startswith("/?") and not location.startswith("//") | |
| browser_target = urljoin("https://fffiloni-viggle-mine-workflow.hf.space/", location) | |
| assert urlsplit(browser_target).scheme == "https" | |
| assert parse_qs(urlsplit(browser_target).query) == {"__theme":["light"],"ui":["canvas"]} | |
| assert response.headers["cache-control"] == "no-store" | |
| # HTML rendering requires launch()'s theme setup; the real-server smoke | |
| # follows this redirect and checks the resulting page separately. | |
| def test_source_picker_bridge_and_legacy_retention(client, prepared, point_payload): | |
| ident = prepared["id"] | |
| page = client.get("/picker/", params={"session": ident}) | |
| assert page.status_code == 200 and "Save points & continue" not in page.text | |
| assert "Save points & continue" in page.text | |
| model = client.get(f"/picker/api/session/{ident}").json() | |
| assert model["coordinate_space"] == "normalized_source" | |
| assert len(model["anchors"]) == 3 | |
| assert client.get(model["anchors"][0]["url"]).headers["content-type"] == "image/png" | |
| saved = client.post(f"/picker/api/session/{ident}/points", json=point_payload) | |
| assert saved.status_code == 200 and saved.json()["saved"] | |
| assert a.points(prepared)["anchors"] == saved.json()["anchors"] | |
| # No queue/SSE is involved in saving or reloading the picker. | |
| again = client.get(f"/picker/api/session/{ident}").json() | |
| assert again["anchors"][0]["points"] == point_payload["anchors"][0]["points"] | |
| assert client.get("/picker/").status_code == 200 | |
| fallback = client.get("/picker/legacy/", params={"session":ident}) | |
| assert fallback.status_code == 200 and "Save points to Workflow" in fallback.text | |
| assert prepared["manifest_path"] in fallback.text and ident in fallback.text | |
| legacy = client.get("/picker/api/manifest", params={"path": prepared["manifest_path"]}) | |
| assert legacy.status_code == 200 | |
| assert client.get(legacy.json()["anchors"][0]["url"]).status_code == 200 | |
| wrong = copy.deepcopy(point_payload); wrong["coordinate_space"] = "stabilized" | |
| assert client.post(f"/picker/api/session/{ident}/points", json=wrong).status_code == 400 | |
| assert client.post(f"/picker/api/session/{ident}/points", json=point_payload, | |
| headers={"origin":"https://another.example"}).status_code == 403 | |
| assert client.get(f"/artifact/{ident}").json()["kind"] == "prepared" | |
| def test_recovery_browser_extracts_exact_stabilized_input_frame(client, camera, prepared): | |
| # A manifest fixture stands in for a completed generation. Only the frame | |
| # extraction below is real; no assertion here implies GPU generation. | |
| passed = a.create("pass_a", {"camera_id":camera["id"], "preparation_id":prepared["id"], | |
| "stabilized_generated":camera["stable_video"], "final_video":prepared["meta"]["normalized"]}, | |
| files=[camera["stable_video"]], parents=[camera["id"]]) | |
| ident = passed["id"] | |
| data = client.get(f"/recovery/api/{ident}").json() | |
| assert data["coordinate_space"] == "stabilized" | |
| assert data["url"] == output_file(camera["stable_video"])["url"] | |
| assert data["url"] != output_file(prepared["meta"]["normalized"])["url"] | |
| assert client.get("/recovery/", params={"checkpoint":ident}).status_code == 200 | |
| response = client.post(f"/recovery/api/{ident}/select", json={"time":1.501}) | |
| assert response.status_code == 200 | |
| selected = response.json() | |
| assert selected["time"] == 1.5 and selected["frame"] == 30 | |
| artifact = a.load(Nodes(cpu_core()).prepare_recovery(a.card(passed,"Saved A")), "recovery") | |
| assert artifact["reacq_time"] == selected["time"] and artifact["pass_a_id"] == ident | |
| from PIL import Image | |
| import media | |
| expected = a.PUBLIC / "recovery-exact-frame.png" | |
| media.extract_frame(camera["stable_video"], expected, 30) | |
| assert Image.open(expected).tobytes() == Image.open(artifact["candidate"]).tobytes() | |
| assert client.post(f"/recovery/api/{ident}/select", json={"time":999}).status_code == 400 | |
| assert client.post(f"/recovery/api/{ident}/select", json={"time":data["start"]}).status_code == 400 | |
| assert client.post(f"/recovery/api/{ident}/select", json={"time":data["end"]}).status_code == 400 | |