"""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()


@pytest.fixture(scope="module")
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 &amp; 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