Spaces:

HayalabCore
/

gemma-endpoint-test

Sleeping

Update waiting state messaging in app.py to provide clearer feedback when the endpoint is inactive. Adjust tests to validate the new placeholder behavior and ensure proper spinner functionality during waiting periods.
b3b0bdf about 1 month ago
Raw History Blame Contribute Delete
28.6 kB
"""Tests for the endpoint chat app.
Run: pytest -q (the required environment is filled in below)
"""

import json
import os

import pytest
import requests

# Assigned outright so real values exported in the shell cannot leak in.
os.environ.update(
    ENDPOINT_URL="http://endpoint.test/",
    QWEN_ENDPOINT_URL="http://qwen.endpoint.test/",
    DEEPSEEK_ENDPOINT_URL="http://deepseek.endpoint.test/",
    HF_TOKEN="test-token",
    QWEN_HF_TOKEN="qwen-org-token",
)
os.environ.pop("DEEPSEEK_HF_TOKEN", None)
os.environ.setdefault("LOGIN_PASSWORD", "s3cret")

import app


def drain(generator):
    """Consume a streaming handler, returning (all yields, final yield)."""
    frames = list(generator)
    return frames, frames[-1]


def histories(frames):
    return [frame[0] for frame in frames]


def textbox(frame):
    """The value the textbox update carries."""
    update = frame[1]
    return update["value"] if isinstance(update, dict) else update


def locked(frame):
    """True when the frame locks the input controls (a request is in flight)."""
    msg_update, send_update, stop_update = frame[1:]
    return (
        msg_update["interactive"] is False
        and send_update["interactive"] is False
        and stop_update["interactive"] is True
    )


def drain_failing(generator, expected):
    """Consume a streaming handler that fails partway; returns frames seen."""
    frames = []

    with pytest.raises(expected):
        for frame in generator:
            frames.append(frame)  # noqa: PERF402 - loop can raise mid-iteration

    return frames


def fake_stream(*chunks):
    def stream(messages, model_name, turn="-"):
        yield from chunks

    return stream


def test_build_api_messages_keeps_order_and_appends_new_turn():
    history = [
        {"role": "user", "content": "hi"},
        {"role": "assistant", "content": "hello"},
    ]

    assert app.build_api_messages(history, "how are you?") == [
        {"role": "user", "content": "hi"},
        {"role": "assistant", "content": "hello"},
        {"role": "user", "content": "how are you?"},
    ]


def test_build_api_messages_flattens_gradio_typed_content():
    """Gradio hands handlers typed content parts, not plain strings."""
    import gradio as gr

    chatbot = gr.Chatbot()
    history = chatbot.preprocess(
        chatbot.postprocess(
            [
                {"role": "user", "content": "hi"},
                {"role": "assistant", "content": "hello"},
            ]
        )
    )

    assert history[0]["content"] != "hi", "test no longer exercises typed parts"
    assert app.build_api_messages(history, "next") == [
        {"role": "user", "content": "hi"},
        {"role": "assistant", "content": "hello"},
        {"role": "user", "content": "next"},
    ]


def test_build_api_messages_drops_non_text_and_malformed_entries():
    history = [
        {"role": "user", "content": [{"type": "file", "file": {"path": "a.png"}}]},
        {"role": "system", "content": "ignored"},
        {"role": "user"},
        None,
        {"role": "assistant", "content": "kept"},
    ]

    assert app.build_api_messages(history, "next") == [
        {"role": "assistant", "content": "kept"},
        {"role": "user", "content": "next"},
    ]


def test_message_text_handles_mixed_and_unknown_content():
    assert app.message_text("plain") == "plain"
    assert app.message_text([{"type": "text", "text": "a"}, {"type": "file"}]) == "a"
    assert (
        app.message_text([{"type": "text", "text": "a"}, {"type": "text", "text": "b"}])
        == "ab"
    )
    assert app.message_text([{"type": "text", "text": None}]) == ""
    assert app.message_text(None) == ""
    assert app.message_text(42) == ""


def test_respond_streams_partial_replies_then_final(monkeypatch):
    monkeypatch.setattr(app, "stream_completion", fake_stream("Hel", "lo ", "there"))

    frames, final = drain(app.respond("ping", [], app.DEFAULT_MODEL))
    history = final[0]

    # Assistant text grows monotonically across frames.
    assert [h[-1]["content"] for h in histories(frames)] == [
        app.WAITING_TEXT,
        "Hel",
        "Hello ",
        "Hello there",
        "Hello there",
    ]
    assert history == [
        {"role": "user", "content": "ping"},
        {"role": "assistant", "content": "Hello there"},
    ]
    assert textbox(final) == ""


def test_respond_shows_pending_spinner_until_first_token(monkeypatch):
    monkeypatch.setattr(app, "stream_completion", fake_stream("first token"))

    frames, final = drain(app.respond("ping", [], app.DEFAULT_MODEL))

    # Frame 1: user turn plus a pending placeholder that renders as a spinner.
    placeholder = frames[0][0][-1]
    assert placeholder["metadata"]["status"] == "pending"
    assert placeholder["metadata"]["title"]

    # Regression: the Chatbot with reasoning_tags silently dropped the empty
    # placeholder, so a sleeping endpoint showed no feedback at all.
    import gradio as gr

    chatbot = next(b for b in app.demo.blocks.values() if isinstance(b, gr.Chatbot))
    rendered = chatbot.postprocess(frames[0][0]).root
    assert [m.metadata.get("status") if m.metadata else None for m in rendered] == [
        None,
        "pending",
    ]

    # The spinner is gone the moment real text exists, and never comes back.
    for history in histories(frames[1:]):
        assert "metadata" not in history[-1], "spinner outlived the first token"

    assert "metadata" not in final[0][-1]


def test_respond_locks_controls_only_while_in_flight(monkeypatch):
    monkeypatch.setattr(app, "stream_completion", fake_stream("a", "b"))

    frames, final = drain(app.respond("ping", [], app.DEFAULT_MODEL))

    assert [locked(f) for f in frames] == [True, True, True, False]
    assert locked(final) is False


def test_respond_unlocks_controls_when_message_is_blank(monkeypatch):
    monkeypatch.setattr(app, "stream_completion", fake_stream("unused"))

    _frames, final = drain(app.respond("", [], app.DEFAULT_MODEL))

    assert locked(final) is False


def test_respond_rejects_an_empty_reply(monkeypatch):
    import gradio as gr

    monkeypatch.setattr(app, "stream_completion", fake_stream())

    frames = drain_failing(app.respond("ping", [], app.DEFAULT_MODEL), gr.Error)

    # No stuck spinner, no empty bubble, and the controls are released.
    assert frames[-1][0] == []
    assert locked(frames[-1]) is False


def test_respond_history_survives_gradio_format_check(monkeypatch):
    """Regression: tuple history raised 'Data incompatible with messages format'."""
    import gradio as gr

    monkeypatch.setattr(app, "stream_completion", fake_stream("pong"))

    frames, _ = drain(app.respond("ping", [], app.DEFAULT_MODEL))

    for history in histories(frames):
        gr.Chatbot()._check_format(history)


def test_respond_sends_history_without_the_placeholder_turn(monkeypatch):
    sent = {}

    def stream(messages, model_name, turn="-"):
        sent["messages"] = messages
        yield "ok"

    monkeypatch.setattr(app, "stream_completion", stream)

    drain(app.respond("second", [{"role": "user", "content": "first"}], None))

    assert sent["messages"] == [
        {"role": "user", "content": "first"},
        {"role": "user", "content": "second"},
    ]


def test_respond_ignores_blank_message(monkeypatch):
    def fail(*args, **kwargs):
        raise AssertionError("endpoint should not be called for a blank message")

    monkeypatch.setattr(app, "stream_completion", fail)

    _frames, final = drain(
        app.respond("   ", [{"role": "user", "content": "hi"}], None)
    )

    assert final[0] == [{"role": "user", "content": "hi"}]


def test_respond_does_not_mutate_caller_history(monkeypatch):
    monkeypatch.setattr(app, "stream_completion", fake_stream("pong"))
    history = []

    drain(app.respond("ping", history, app.DEFAULT_MODEL))

    assert history == []


@pytest.mark.parametrize(
    "error",
    [
        requests.exceptions.Timeout(),
        requests.exceptions.ConnectionError("refused"),
        ValueError("not json"),
    ],
)
def test_respond_surfaces_immediate_failures_and_restores_input(monkeypatch, error):
    import gradio as gr

    def stream(messages, model_name, turn="-"):
        raise error
        yield  # pragma: no cover - makes this a generator

    monkeypatch.setattr(app, "stream_completion", stream)

    frames = drain_failing(app.respond("ping", [], app.DEFAULT_MODEL), gr.Error)

    # Nothing streamed, so the whole failed turn is rolled back and the user's
    # text is handed back to the input box rather than duplicated in the chat.
    assert frames[-1][0] == []
    assert textbox(frames[-1]) == "ping"
    assert locked(frames[-1]) is False


def test_respond_keeps_partial_reply_when_stream_breaks(monkeypatch):
    import gradio as gr

    def stream(messages, model_name, turn="-"):
        yield "half a "
        raise requests.exceptions.ChunkedEncodingError("dropped")

    monkeypatch.setattr(app, "stream_completion", stream)

    frames = drain_failing(app.respond("ping", [], app.DEFAULT_MODEL), gr.Error)

    assert frames[-1][0] == [
        {"role": "user", "content": "ping"},
        {"role": "assistant", "content": "half a "},
    ]
    # A kept partial reply means the box is not refilled with the same text.
    assert textbox(frames[-1]) == ""
    assert locked(frames[-1]) is False


def test_respond_reraises_unexpected_errors(monkeypatch):
    def stream(messages, model_name, turn="-"):
        raise RuntimeError("boom")
        yield  # pragma: no cover

    monkeypatch.setattr(app, "stream_completion", stream)

    with pytest.raises(RuntimeError, match="boom"):
        drain(app.respond("ping", [], app.DEFAULT_MODEL))


def test_verify_login_rejects_wrong_and_empty_password():
    assert app.verify_login("user", "s3cret") is True
    assert app.verify_login("user", "wrong") is False
    assert app.verify_login("user", "") is False
    assert app.verify_login("user", None) is False


def test_missing_required_env_fails_with_named_message(monkeypatch):
    monkeypatch.delenv("ENDPOINT_URL", raising=False)

    with pytest.raises(RuntimeError, match="ENDPOINT_URL"):
        app._required_env("ENDPOINT_URL")


class FakeStreamResponse:
    """Stand-in for the streaming `requests` response."""

    def __init__(self, lines, status_code=200):
        self._lines = lines
        self.status_code = status_code
        self.closed = False

    def __enter__(self):
        return self

    def __exit__(self, *exc_info):
        self.closed = True

    def raise_for_status(self):
        pass

    def iter_lines(self, chunk_size=512, decode_unicode=False):
        self.iter_kwargs = {"chunk_size": chunk_size, "decode_unicode": decode_unicode}
        yield from self._lines


def sse(*payloads):
    return [f"data: {p}" for p in payloads]


def chunk(content):
    return json.dumps({"choices": [{"delta": {"content": content}}]})


def test_each_model_posts_to_its_own_endpoint(monkeypatch):
    captured = {}

    def fake_post(url, **kwargs):
        captured.update(url=url, **kwargs)
        return FakeStreamResponse(sse(chunk("hi"), "[DONE]"))

    monkeypatch.setattr(app.requests, "post", fake_post)

    expected = {
        app.GEMMA_MODEL: "http://endpoint.test",
        app.QWEN_MODEL: "http://qwen.endpoint.test",
        app.DEEPSEEK_MODEL: "http://deepseek.endpoint.test",
    }

    for model, url in expected.items():
        list(app.stream_completion([{"role": "user", "content": "hi"}], model))
        assert captured["url"] == f"{url}/v1/chat/completions"
        assert captured["json"]["model"] == model

    assert app.AVAILABLE_MODELS == [
        app.GEMMA_MODEL,
        app.QWEN_MODEL,
        app.DEEPSEEK_MODEL,
    ]


def test_each_model_sends_its_own_token(monkeypatch):
    """Endpoints in different orgs need tokens with access to that org."""
    sent = {}

    def fake_post(url, **kwargs):
        sent[url] = kwargs["headers"]["Authorization"]
        return FakeStreamResponse(sse(chunk("hi"), "[DONE]"))

    monkeypatch.setattr(app.requests, "post", fake_post)

    for model in app.AVAILABLE_MODELS:
        list(app.stream_completion([], model))

    assert sent == {
        "http://endpoint.test/v1/chat/completions": "Bearer test-token",
        "http://qwen.endpoint.test/v1/chat/completions": "Bearer qwen-org-token",
        # No DEEPSEEK_HF_TOKEN, so it falls back to HF_TOKEN.
        "http://deepseek.endpoint.test/v1/chat/completions": "Bearer test-token",
    }
    assert app.MODELS[app.QWEN_MODEL].token_env == "QWEN_HF_TOKEN"
    assert app.MODELS[app.DEEPSEEK_MODEL].token_env == "HF_TOKEN"


@pytest.mark.parametrize("status", [401, 403])
def test_rejected_token_names_the_secret_to_fix(monkeypatch, status):
    import gradio as gr

    monkeypatch.setattr(
        app.requests, "post", lambda url, **kw: FakeStreamResponse([], status)
    )

    with pytest.raises(gr.Error, match="QWEN_HF_TOKEN") as error:
        list(app.stream_completion([], app.QWEN_MODEL))

    assert "qwen-org-token" not in str(error.value)


def test_stream_completion_parses_sse_chunks(monkeypatch):
    lines = [
        ": keep-alive",
        "",
        *sse(chunk("Hel"), chunk("lo")),
        "",
        # Final chunk carries a finish_reason and no content.
        json.dumps({"choices": [{"delta": {}, "finish_reason": "stop"}]}).join(
            ["data: ", ""]
        ),
        "data: [DONE]",
        "data: {bad json that must never be reached}",
    ]
    response = FakeStreamResponse(lines)
    captured = {}

    def fake_post(url, **kwargs):
        captured.update(url=url, **kwargs)
        return response

    monkeypatch.setattr(app.requests, "post", fake_post)

    assert list(app.stream_completion([{"role": "user", "content": "hi"}], "m")) == [
        "Hel",
        "lo",
    ]
    assert captured["url"] == "http://endpoint.test/v1/chat/completions"
    assert captured["json"]["stream"] is True
    assert captured["stream"] is True
    assert captured["headers"]["Authorization"] == "Bearer test-token"
    assert response.closed


def reasoning_chunk(text):
    return json.dumps({"choices": [{"delta": {"reasoning_content": text}}]})


def test_stream_completion_wraps_reasoning_in_think_tags(monkeypatch):
    """llama.cpp streams thinking as reasoning_content; the Chatbot needs tags."""
    lines = sse(reasoning_chunk("7 x "), reasoning_chunk("143"), chunk("No."), "[DONE]")
    monkeypatch.setattr(
        app.requests, "post", lambda url, **kwargs: FakeStreamResponse(lines)
    )

    assert "".join(app.stream_completion([], "m")) == "<think>7 x 143</think>No."


def test_stream_completion_closes_reasoning_cut_off_mid_thought(monkeypatch):
    lines = sse(reasoning_chunk("still thinking"), "[DONE]")
    monkeypatch.setattr(
        app.requests, "post", lambda url, **kwargs: FakeStreamResponse(lines)
    )

    assert "".join(app.stream_completion([], "m")) == "<think>still thinking</think>"


def test_reasoning_is_not_sent_back_to_the_model():
    """The Chatbot splits reasoning into a titled thought, which is dropped."""
    import gradio as gr

    chatbot = gr.Chatbot(reasoning_tags=[(app.THINK_OPEN, app.THINK_CLOSE)])
    history = chatbot.preprocess(
        chatbot.postprocess(
            [
                {"role": "user", "content": "hi"},
                {"role": "assistant", "content": "<think>hmm</think>hello"},
            ]
        )
    )

    assert app.build_api_messages(history, "next") == [
        {"role": "user", "content": "hi"},
        {"role": "assistant", "content": "hello"},
        {"role": "user", "content": "next"},
    ]


def test_stop_replaces_send_while_a_reply_streams():
    _msg, send, stop = app.busy_controls()
    assert send["visible"] is False and stop["visible"] is True

    _msg, send, stop = app.idle_controls()
    assert send["visible"] is True and stop["visible"] is False


def test_new_chat_empties_history_and_unlocks():
    history, *controls = app.new_chat()

    assert history == []
    assert controls[0]["interactive"] is True


def test_every_model_has_a_display_label():
    assert set(app.MODEL_LABELS) == set(app.AVAILABLE_MODELS)


def test_stream_completion_tolerates_chunks_without_choices(monkeypatch):
    lines = sse(json.dumps({"choices": []}), chunk("ok"), "[DONE]")
    monkeypatch.setattr(
        app.requests, "post", lambda url, **kwargs: FakeStreamResponse(lines)
    )

    assert list(app.stream_completion([], "m")) == ["ok"]


def test_stream_completion_raises_on_malformed_chunk(monkeypatch):
    lines = sse(chunk("ok"), "{not json")
    monkeypatch.setattr(
        app.requests, "post", lambda url, **kwargs: FakeStreamResponse(lines)
    )

    with pytest.raises(ValueError):
        list(app.stream_completion([], "m"))


def test_module_exposes_demo_for_spaces_hot_reload():
    """Spaces' reloader does getattr(__main__, "demo") after every reload."""
    import gradio as gr

    assert isinstance(app.demo, gr.Blocks)


def test_cancel_stream_clears_spinner_and_unlocks(monkeypatch):
    """Stop cancels the generator mid-flight, so cleanup runs in its own event."""
    monkeypatch.setattr(app, "stream_completion", fake_stream("half"))

    frames, _final = drain(app.respond("ping", [], app.DEFAULT_MODEL))
    mid_flight = frames[0][0]  # still showing the pending spinner

    history, *controls = app.cancel_stream(mid_flight)

    assert history == [{"role": "user", "content": "ping"}]
    assert all(update["interactive"] is not False for update in controls[:2])


def test_normalize_history_drops_thought_placeholders():
    history = [
        {"role": "user", "content": "hi"},
        {"role": "assistant", "content": "", "metadata": app.WAITING_METADATA},
        {"role": "assistant", "content": "kept", "metadata": {}},
    ]

    assert app.normalize_history(history) == [
        {"role": "user", "content": "hi"},
        {"role": "assistant", "content": "kept"},
    ]


def test_stream_completion_reads_incrementally(monkeypatch):
    """Regression: the default 512-byte reads deliver the whole reply at once."""
    response = FakeStreamResponse(sse(chunk("hi"), "[DONE]"))
    monkeypatch.setattr(app.requests, "post", lambda url, **kwargs: response)

    list(app.stream_completion([], "m"))

    assert response.iter_kwargs["chunk_size"] == 1
    assert response.iter_kwargs["decode_unicode"] is True


def test_successful_turn_is_logged(monkeypatch, caplog):
    monkeypatch.setattr(app, "stream_completion", fake_stream("pong"))

    with caplog.at_level("INFO", logger="gemma-endpoint"):
        drain(app.respond("ping", [], app.DEFAULT_MODEL))

    messages = [r.getMessage() for r in caplog.records]
    assert any("start model=" in m for m in messages)
    assert any("complete reply_chars=4" in m for m in messages)


def test_failed_turn_is_logged_at_error(monkeypatch, caplog):
    import gradio as gr

    def stream(messages, model_name, turn="-"):
        raise requests.exceptions.ConnectionError("refused")
        yield  # pragma: no cover

    monkeypatch.setattr(app, "stream_completion", stream)

    with caplog.at_level("INFO", logger="gemma-endpoint"):
        drain_failing(app.respond("ping", [], app.DEFAULT_MODEL), gr.Error)

    errors = [r.getMessage() for r in caplog.records if r.levelname == "ERROR"]
    assert any("failed after" in m and "refused" in m for m in errors)


def test_unexpected_error_logs_a_traceback(monkeypatch, caplog):
    def stream(messages, model_name, turn="-"):
        raise RuntimeError("boom")
        yield  # pragma: no cover

    monkeypatch.setattr(app, "stream_completion", stream)

    with caplog.at_level("INFO", logger="gemma-endpoint"), pytest.raises(RuntimeError):
        drain(app.respond("ping", [], app.DEFAULT_MODEL))

    assert any(r.exc_info for r in caplog.records), "no traceback was logged"


def test_turn_ids_correlate_lines_of_one_turn(monkeypatch, caplog):
    def stream(messages, model_name, turn="-"):
        app.logger.info("turn=%s inner", turn)
        yield "ok"

    monkeypatch.setattr(app, "stream_completion", stream)

    with caplog.at_level("INFO", logger="gemma-endpoint"):
        drain(app.respond("one", [], app.DEFAULT_MODEL))
        first = {
            m.split("turn=")[1].split()[0]
            for m in (r.getMessage() for r in caplog.records)
            if "turn=" in m
        }
        caplog.clear()
        drain(app.respond("two", [], app.DEFAULT_MODEL))
        second = {
            m.split("turn=")[1].split()[0]
            for m in (r.getMessage() for r in caplog.records)
            if "turn=" in m
        }

    assert len(first) == 1 and len(second) == 1, "one turn must use one id"
    assert first != second, "consecutive turns must not share an id"


def test_rejected_login_is_logged_without_the_password(caplog):
    with caplog.at_level("INFO", logger="gemma-endpoint"):
        assert app.verify_login("intruder", "hunter2") is False

    record = caplog.records[-1]
    assert record.levelname == "WARNING"
    assert "REJECTED" in record.getMessage()
    assert "hunter2" not in record.getMessage()
    assert "s3cret" not in record.getMessage()


def test_logs_never_contain_the_hf_token(monkeypatch, caplog):
    """The token is only ever an Authorization header, never a log field."""
    response = FakeStreamResponse(sse(chunk("hi"), "[DONE]"))
    monkeypatch.setattr(app.requests, "post", lambda url, **kwargs: response)

    with caplog.at_level("DEBUG", logger="gemma-endpoint"):
        for model in app.AVAILABLE_MODELS:
            list(
                app.stream_completion(
                    [{"role": "user", "content": "hi"}], model, turn=1
                )
            )

    assert caplog.records, "expected the request to be logged"
    tokens = {endpoint.token for endpoint in app.MODELS.values()}
    for record in caplog.records:
        assert not any(token in record.getMessage() for token in tokens)
        assert "Authorization" not in record.getMessage()


def test_log_level_applies_to_this_app_only():
    """LOG_LEVEL must not switch on httpx/urllib3 chatter along with ours."""
    import logging

    assert app.logger.level == logging.INFO
    assert logging.getLogger().level == logging.WARNING
    assert app.logger.isEnabledFor(logging.INFO)
    assert not logging.getLogger("httpx").isEnabledFor(logging.INFO)


@pytest.fixture
def no_sleep(monkeypatch):
    """Run retry loops at full speed, with a clock that still advances."""
    clock = {"now": 0.0}

    monkeypatch.setattr(app.time, "monotonic", lambda: clock["now"])
    monkeypatch.setattr(
        app.time, "sleep", lambda seconds: clock.update(now=clock["now"] + seconds)
    )

    return clock


@pytest.mark.parametrize("status", [502, 503, 504])
def test_stream_completion_flags_a_cold_endpoint(monkeypatch, status):
    monkeypatch.setattr(
        app.requests, "post", lambda url, **kw: FakeStreamResponse([], status)
    )

    with pytest.raises(app.EndpointWaking):
        list(app.stream_completion([], "m"))


def test_stream_completion_still_raises_for_real_errors(monkeypatch):
    response = FakeStreamResponse([], 404)

    def raise_for_status():
        raise requests.exceptions.HTTPError("404", response=None)

    response.raise_for_status = raise_for_status
    monkeypatch.setattr(app.requests, "post", lambda url, **kw: response)

    with pytest.raises(requests.exceptions.HTTPError):
        list(app.stream_completion([], "m"))


def test_stream_turn_waits_out_a_cold_endpoint(monkeypatch, no_sleep):
    attempts = {"n": 0}

    def stream(messages, model_name, turn="-"):
        attempts["n"] += 1

        if attempts["n"] < 3:
            raise app.EndpointWaking(503)
            yield  # pragma: no cover

        yield "awake"

    monkeypatch.setattr(app, "stream_completion", stream)

    events = list(app.stream_turn([], "m", turn=1))

    assert attempts["n"] == 3, "should retry until the endpoint answers"
    assert ("token", "awake") in events
    waking = [value for kind, value in events if kind == "waking"]
    assert waking, "no waking updates were emitted"
    assert waking == sorted(waking), "the waited counter must not go backwards"


def test_stream_turn_gives_up_after_the_wake_timeout(monkeypatch, no_sleep):
    import gradio as gr

    def always_cold(messages, model_name, turn="-"):
        raise app.EndpointWaking(503)
        yield  # pragma: no cover

    monkeypatch.setattr(app, "stream_completion", always_cold)

    with pytest.raises(gr.Error, match="did not come up"):
        for _ in app.stream_turn([], "m", turn=1):
            pass

    assert no_sleep["now"] >= app.WAKE_TIMEOUT_SECONDS


def test_respond_shows_a_waking_message_then_the_reply(monkeypatch, no_sleep):
    attempts = {"n": 0}

    def stream(messages, model_name, turn="-"):
        attempts["n"] += 1

        if attempts["n"] == 1:
            raise app.EndpointWaking(503)
            yield  # pragma: no cover

        yield "hello"

    monkeypatch.setattr(app, "stream_completion", stream)

    frames, final = drain(app.respond("ping", [], app.DEFAULT_MODEL))
    titles = [
        (h[-1].get("metadata") or {}).get("title") for h in histories(frames) if h
    ]

    assert any(t and "Waking up the endpoint" in t for t in titles)
    assert all(locked(f) for f in frames[:-1]), "controls stay locked while waking"
    assert final[0][-1] == {"role": "assistant", "content": "hello"}


def test_respond_reports_a_cold_endpoint_that_never_wakes(monkeypatch, no_sleep):
    import gradio as gr

    def always_cold(messages, model_name, turn="-"):
        raise app.EndpointWaking(503)
        yield  # pragma: no cover

    monkeypatch.setattr(app, "stream_completion", always_cold)

    frames = drain_failing(app.respond("ping", [], app.DEFAULT_MODEL), gr.Error)

    assert frames[-1][0] == [], "the failed turn is rolled back"
    assert textbox(frames[-1]) == "ping"
    assert locked(frames[-1]) is False


class Utf8SseServer:
    """A real HTTP server streaming raw UTF-8 SSE, as vLLM/llama.cpp do.
    `FakeStreamResponse` hands back `str` lines that are already decoded, so it
    cannot catch a decoding bug. Only a real socket exercises the path where
    `requests` has to pick a charset.
    """

    def __init__(self, content_type):
        import http.server
        import threading

        body = "".join(
            f"data: {json.dumps({'choices': [{'delta': {'content': token}}]}, ensure_ascii=False)}\n\n"
            for token in ("こんにちは", "、", "日本語", "です", "。")
        )
        raw = (body + "data: [DONE]\n\n").encode("utf-8")

        class Handler(http.server.BaseHTTPRequestHandler):
            def do_POST(self):
                self.rfile.read(int(self.headers.get("Content-Length") or 0))
                self.send_response(200)
                self.send_header("Content-Type", content_type)
                self.send_header("Transfer-Encoding", "chunked")
                self.end_headers()

                # Split on a 7-byte boundary so multi-byte characters land
                # across two chunks, as they do on a real connection.
                for start in range(0, len(raw), 7):
                    piece = raw[start : start + 7]
                    self.wfile.write(b"%x\r\n%s\r\n" % (len(piece), piece))
                    self.wfile.flush()

                self.wfile.write(b"0\r\n\r\n")

            def log_message(self, *args):
                pass

        self._server = http.server.HTTPServer(("127.0.0.1", 0), Handler)
        threading.Thread(target=self._server.serve_forever, daemon=True).start()

    @property
    def url(self):
        return f"http://127.0.0.1:{self._server.server_port}"

    def close(self):
        self._server.shutdown()


@pytest.mark.parametrize(
    "content_type",
    ["text/event-stream", "text/event-stream; charset=utf-8"],
)
def test_stream_completion_decodes_utf8_replies(monkeypatch, content_type):
    """Regression: a charset-less SSE response decoded Japanese as mojibake.
    `requests` falls back to ISO-8859-1 for a text/* response with no charset,
    so every non-ASCII reply came back garbled while English looked fine.
    """
    server = Utf8SseServer(content_type)
    monkeypatch.setattr(app, "ENDPOINT_URL", server.url)

    try:
        reply = "".join(app.stream_completion([], "m"))
    finally:
        server.close()

    assert reply == "こんにちは、日本語です。"