Files
gradio-app--gradio/test/test_chat_interface.py
wehub-resource-sync adf0d17497
publish / version_or_publish (push) Waiting to run
storybook-build / changes (push) Waiting to run
storybook-build / :storybook-build (push) Blocked by required conditions
Sync Gradio Skills to Hugging Face / sync-skills (push) Waiting to run
functional / changes (push) Waiting to run
functional / build-frontend (push) Blocked by required conditions
functional / functional-test-SSR=false (push) Blocked by required conditions
functional / functional-reload (push) Blocked by required conditions
functional / functional-test-SSR=true (push) Blocked by required conditions
hygiene / hygiene-test (push) Waiting to run
python / changes (push) Waiting to run
python / build (push) Blocked by required conditions
python / test-ubuntu-latest-flaky (push) Blocked by required conditions
python / test-ubuntu-latest-not-flaky (push) Blocked by required conditions
python / test-windows-latest-flaky (push) Blocked by required conditions
python / test-windows-latest-not-flaky (push) Blocked by required conditions
js / changes (push) Waiting to run
js / js-test (push) Blocked by required conditions
docs-build / changes (push) Waiting to run
docs-build / docs-build (push) Blocked by required conditions
docs-build / website-build (push) Blocked by required conditions
chore: import upstream snapshot with attribution
2026-07-13 13:17:32 +08:00

576 lines
22 KiB
Python

import tempfile
import warnings
from concurrent.futures import wait
from pathlib import Path
from unittest.mock import patch
import pytest
from gradio_client import handle_file
import gradio as gr
from gradio.components.chatbot import Message, TextMessage
def invalid_fn(message):
return message
def double(message, history):
return message + " " + message
async def async_greet(message, history):
return "hi, " + message
def stream(message, history):
for i in range(len(message)):
yield message[: i + 1]
async def async_stream(message, history):
for i in range(len(message)):
yield message[: i + 1]
def count(message, history):
return str(len(history))
def echo_system_prompt_plus_message(message, history, system_prompt, tokens):
response = f"{system_prompt} {message}"
for i in range(min(len(response), int(tokens))):
yield response[: i + 1]
class TestInit:
def test_no_fn(self):
with pytest.raises(TypeError):
gr.ChatInterface() # type: ignore
def test_concurrency_limit(self):
chat = gr.ChatInterface(double, concurrency_limit=10)
assert chat.concurrency_limit == 10
fns = [
fn
for fn in chat.fns.values()
if fn.name in {"_submit_fn", "_api_submit_fn"}
]
assert all(fn.concurrency_limit == 10 for fn in fns)
def test_custom_textbox(self):
def chat():
return "Hello"
gr.ChatInterface(
chat,
chatbot=gr.Chatbot(height=400),
textbox=gr.Textbox(placeholder="Type Message", container=False, scale=7),
title="Test",
)
gr.ChatInterface(
chat,
chatbot=gr.Chatbot(height=400),
textbox=gr.MultimodalTextbox(container=False, scale=7),
title="Test",
)
def test_events_attached(self):
chatbot = gr.ChatInterface(double)
dependencies = chatbot.fns.values()
textbox = chatbot.textbox._id
assert next(
(d for d in dependencies if d.targets == [(textbox, "submit")]),
None,
)
def test_example_caching(self, connect):
with patch(
"gradio.utils.get_cache_folder", return_value=Path(tempfile.mkdtemp())
):
chatbot = gr.ChatInterface(
double, examples=["hello", "hi"], cache_examples=True
)
with connect(chatbot):
prediction_hello = chatbot.examples_handler.load_from_cache(0)
prediction_hi = chatbot.examples_handler.load_from_cache(1)
assert prediction_hello[0].root == [
Message(role="user", content=[TextMessage(text="hello")]),
Message(role="assistant", content=[TextMessage(text="hello hello")]),
]
assert prediction_hi[0].root == [
Message(role="user", content=[TextMessage(text="hi")]),
Message(role="assistant", content=[TextMessage(text="hi hi")]),
]
@pytest.mark.asyncio
async def test_example_caching_lazy(self):
with patch(
"gradio.utils.get_cache_folder", return_value=Path(tempfile.mkdtemp())
):
chatbot = gr.ChatInterface(
double,
examples=["hello", "hi"],
cache_examples=True,
cache_mode="lazy",
)
prediction_hello = chatbot.examples_handler.load_from_cache(0)
assert prediction_hello[0].root == [
Message(role="user", content=[TextMessage(text="hello")]),
Message(role="assistant", content=[TextMessage(text="hello hello")]),
]
prediction_hi = chatbot.examples_handler.load_from_cache(1)
assert prediction_hi[0].root == [
Message(role="user", content=[TextMessage(text="hi")]),
Message(role="assistant", content=[TextMessage(text="hi hi")]),
]
def test_example_caching_async(self, connect):
with patch(
"gradio.utils.get_cache_folder", return_value=Path(tempfile.mkdtemp())
):
chatbot = gr.ChatInterface(
async_greet, examples=["abubakar", "tom"], cache_examples=True
)
with connect(chatbot):
prediction_hello = chatbot.examples_handler.load_from_cache(0)
prediction_hi = chatbot.examples_handler.load_from_cache(1)
assert prediction_hello[0].root == [
Message(role="user", content=[TextMessage(text="abubakar")]),
Message(role="assistant", content=[TextMessage(text="hi, abubakar")]),
]
assert prediction_hi[0].root == [
Message(role="user", content=[TextMessage(text="tom")]),
Message(role="assistant", content=[TextMessage(text="hi, tom")]),
]
def test_example_caching_with_streaming(self, connect):
with patch(
"gradio.utils.get_cache_folder", return_value=Path(tempfile.mkdtemp())
):
chatbot = gr.ChatInterface(
stream, examples=["hello", "hi"], cache_examples=True
)
with connect(chatbot):
prediction_hello = chatbot.examples_handler.load_from_cache(0)
prediction_hi = chatbot.examples_handler.load_from_cache(1)
assert prediction_hello[0].root == [
Message(role="user", content=[TextMessage(text="hello")]),
Message(role="assistant", content=[TextMessage(text="hello")]),
]
assert prediction_hi[0].root == [
Message(role="user", content=[TextMessage(text="hi")]),
Message(role="assistant", content=[TextMessage(text="hi")]),
]
def test_example_caching_with_streaming_async(self, connect):
with patch(
"gradio.utils.get_cache_folder", return_value=Path(tempfile.mkdtemp())
):
chatbot = gr.ChatInterface(
async_stream, examples=["hello", "hi"], cache_examples=True
)
with connect(chatbot):
prediction_hello = chatbot.examples_handler.load_from_cache(0)
prediction_hi = chatbot.examples_handler.load_from_cache(1)
assert prediction_hello[0].root == [
Message(role="user", content=[TextMessage(text="hello")]),
Message(role="assistant", content=[TextMessage(text="hello")]),
]
assert prediction_hi[0].root == [
Message(role="user", content=[TextMessage(text="hi")]),
Message(role="assistant", content=[TextMessage(text="hi")]),
]
def test_default_accordion_params(self):
chatbot = gr.ChatInterface(
echo_system_prompt_plus_message,
additional_inputs=["textbox", "slider"],
)
accordion = [
comp
for comp in chatbot.blocks.values()
if comp.get_config().get("name") == "accordion"
][0]
assert accordion.get_config().get("open") is False
assert (
accordion.get_config().get("label").key # type: ignore
== "chat_interface.additional_inputs"
)
def test_setting_accordion_params(self, monkeypatch):
chatbot = gr.ChatInterface(
echo_system_prompt_plus_message,
additional_inputs=["textbox", "slider"],
additional_inputs_accordion=gr.Accordion(open=True, label="MOAR"),
)
accordion = [
comp
for comp in chatbot.blocks.values()
if comp.get_config().get("name") == "accordion"
][0]
assert accordion.get_config().get("open") is True
assert accordion.get_config().get("label") == "MOAR"
def test_example_caching_with_additional_inputs(self, monkeypatch, connect):
with patch(
"gradio.utils.get_cache_folder", return_value=Path(tempfile.mkdtemp())
):
chatbot = gr.ChatInterface(
echo_system_prompt_plus_message,
additional_inputs=["textbox", "slider"],
examples=[["hello", "robot", 100], ["hi", "robot", 2]],
cache_examples=True,
)
with connect(chatbot):
prediction_hello = chatbot.examples_handler.load_from_cache(0)
prediction_hi = chatbot.examples_handler.load_from_cache(1)
assert prediction_hello[0].root == [
Message(role="user", content=[TextMessage(text="hello")]),
Message(role="assistant", content=[TextMessage(text="robot hello")]),
]
assert prediction_hi[0].root == [
Message(role="user", content=[TextMessage(text="hi")]),
Message(role="assistant", content=[TextMessage(text="ro")]),
]
def test_example_caching_with_additional_inputs_already_rendered(
self, monkeypatch, connect
):
with patch(
"gradio.utils.get_cache_folder", return_value=Path(tempfile.mkdtemp())
):
with gr.Blocks() as demo:
with gr.Accordion("Inputs"):
text = gr.Textbox()
slider = gr.Slider()
chatbot = gr.ChatInterface(
echo_system_prompt_plus_message,
additional_inputs=[text, slider],
examples=[["hello", "robot", 100], ["hi", "robot", 2]],
cache_examples=True,
)
with connect(demo):
prediction_hello = chatbot.examples_handler.load_from_cache(0)
prediction_hi = chatbot.examples_handler.load_from_cache(1)
assert prediction_hello[0].root == [
Message(role="user", content=[TextMessage(text="hello")]),
Message(role="assistant", content=[TextMessage(text="robot hello")]),
]
assert prediction_hi[0].root == [
Message(role="user", content=[TextMessage(text="hi")]),
Message(role="assistant", content=[TextMessage(text="ro")]),
]
def test_custom_chatbot_with_events(self):
with gr.Blocks() as demo:
chatbot = gr.Chatbot()
chatbot.like(lambda: None, None, None)
gr.ChatInterface(fn=lambda x, y: x, chatbot=chatbot)
dependencies = demo.fns.values()
assert next(
(d for d in dependencies if d.targets == [(chatbot._id, "like")]),
None,
)
class TestAPI:
def test_get_api_info(self):
chatbot = gr.ChatInterface(double, api_name="chat")
api_info = chatbot.get_api_info()
assert api_info
assert len(api_info["named_endpoints"]) == 1
assert len(api_info["unnamed_endpoints"]) == 0
assert "/chat" in api_info["named_endpoints"]
def test_streaming_api(self, connect):
chatbot = gr.ChatInterface(stream).queue()
with connect(chatbot) as client:
job = client.submit("hello")
wait([job])
assert job.outputs() == ["h", "he", "hel", "hell", "hello"]
def test_streaming_api_async(self, connect):
chatbot = gr.ChatInterface(async_stream).queue()
with connect(chatbot) as client:
job = client.submit("hello")
wait([job])
assert job.outputs() == ["h", "he", "hel", "hell", "hello"]
def test_non_streaming_api(self, connect):
chatbot = gr.ChatInterface(double)
with connect(chatbot) as client:
result = client.predict("hello")
assert result == "hello hello"
def test_non_streaming_api_default(self, connect):
chatbot = gr.ChatInterface(double, api_name="double")
with connect(chatbot) as client:
result = client.predict("hello", api_name="/double")
assert result == "hello hello"
def test_non_streaming_api_async(self, connect):
chatbot = gr.ChatInterface(async_greet)
with connect(chatbot) as client:
result = client.predict("gradio")
assert result == "hi, gradio"
def test_streaming_api_with_additional_inputs(self, connect):
chatbot = gr.ChatInterface(
echo_system_prompt_plus_message,
additional_inputs=["textbox", "slider"],
).queue()
with connect(chatbot) as client:
job = client.submit("hello", "robot", 7)
wait([job])
assert job.outputs() == [
"r",
"ro",
"rob",
"robo",
"robot",
"robot ",
"robot h",
]
def test_multimodal_api(self, connect):
def double_multimodal(msg, history):
return msg["text"] + " " + msg["text"]
chatbot = gr.ChatInterface(
double_multimodal,
multimodal=True,
api_name="chat",
)
with connect(chatbot) as client:
result = client.predict({"text": "hello", "files": []}, api_name="/chat")
assert result == "hello hello"
def test_component_returned(self, connect):
def mock_chat_fn(msg, history):
return gr.Audio("test/test_files/audio_sample.wav")
chatbot = gr.ChatInterface(
mock_chat_fn,
multimodal=True,
api_name="chat",
)
with connect(chatbot) as client:
result = client.predict(
{
"text": "hello",
"files": [handle_file("test/test_files/audio_sample.wav")],
},
api_name="/chat",
)
assert result["value"] == "test/test_files/audio_sample.wav"
def test_multiple_messages(self, connect):
def multiple_messages(msg, history):
return [msg["text"], msg["text"]]
chatbot = gr.ChatInterface(
multiple_messages,
multimodal=True,
api_name="chat",
)
with connect(chatbot) as client:
result = client.predict({"text": "hello", "files": []}, api_name="/chat")
assert result == ["hello", "hello"]
class TestExampleMessages:
def test_setup_example_messages_with_strings(self):
chat = gr.ChatInterface(
double,
examples=["hello", "hi", "hey"],
example_labels=["Greeting 1", "Greeting 2", "Greeting 3"],
)
assert len(chat.examples_messages) == 3
assert chat.examples_messages[0] == {
"text": "hello",
"display_text": "Greeting 1",
}
assert chat.examples_messages[1] == {
"text": "hi",
"display_text": "Greeting 2",
}
assert chat.examples_messages[2] == {
"text": "hey",
"display_text": "Greeting 3",
}
def test_setup_example_messages_with_multimodal(self):
chat = gr.ChatInterface(
double,
examples=[ # type: ignore
{"text": "hello", "files": ["file1.txt"]},
{"text": "hi", "files": ["file2.txt", "file3.txt"]},
{"text": "", "files": ["file4.txt"]},
],
)
assert len(chat.examples_messages) == 3
assert chat.examples_messages[0]["text"] == "hello" # type: ignore
assert chat.examples_messages[0]["files"][0]["path"].endswith("file1.txt") # type: ignore
def test_setup_example_messages_with_lists(self):
chat = gr.ChatInterface(
double,
examples=[
["hello", "other_value"],
["hi", "another_value"],
],
)
assert len(chat.examples_messages) == 2
assert chat.examples_messages[0] == {"text": "hello"}
assert chat.examples_messages[1] == {"text": "hi"}
def test_setup_example_messages_empty(self):
chat = gr.ChatInterface(double)
chat._setup_example_messages(None)
assert chat.examples_messages == []
def test_chat_interface_api_name(self, connect):
chat = gr.ChatInterface(double, api_visibility="private")
assert chat.api_visibility == "private"
with connect(chat) as client:
assert client.view_api(return_format="dict")["named_endpoints"] == {}
chat = gr.ChatInterface(double, api_name="double")
with connect(chat) as client:
assert "/double" in client.view_api(return_format="dict")["named_endpoints"]
def test_chat_interface_api_names_with_additional_inputs(self, connect):
def response(message, history, random_number: int):
return str(random_number)
chat = gr.ChatInterface(
response,
additional_inputs=[gr.Textbox(label="Random number")],
api_name="chat",
)
with connect(chat) as client:
endpoints = client.view_api(return_format="dict")["named_endpoints"]
assert "/chat" in endpoints
assert any(
p["parameter_name"] == "random_number"
for p in endpoints["/chat"]["parameters"]
)
def test_example_icons_set_if_multimodal_false(self):
demo = gr.ChatInterface(
fn=double,
title="🌤️ Weather Assistant",
description="Ask about the weather anywhere! Watch as I gather the information step by step.",
examples=[
"What's the weather like in Tokyo?",
"Is it sunny in Paris right now?",
"Should I bring an umbrella in New York today?",
],
example_icons=[
"https://cdn3.iconfinder.com/data/icons/landmark-outline/432/japan_tower_tokyo_landmark_travel_architecture_tourism_view-256.png",
"https://cdn2.iconfinder.com/data/icons/city-building-1/200/ArcdeTriomphe-256.png",
"https://cdn2.iconfinder.com/data/icons/city-icons-for-offscreen-magazine/80/new-york-256.png",
],
)
assert len(demo.examples_messages) == 3
assert (
demo.examples_messages[0].get("icon", {}).get("url") # type: ignore
== "https://cdn3.iconfinder.com/data/icons/landmark-outline/432/japan_tower_tokyo_landmark_travel_architecture_tourism_view-256.png"
)
assert (
demo.examples_messages[1].get("icon", {}).get("url") # type: ignore
== "https://cdn2.iconfinder.com/data/icons/city-building-1/200/ArcdeTriomphe-256.png"
)
assert (
demo.examples_messages[2].get("icon", {}).get("url") # type: ignore
== "https://cdn2.iconfinder.com/data/icons/city-icons-for-offscreen-magazine/80/new-york-256.png"
)
def chat(message, history):
return f"Echo: {message}"
class TestTextboxParameterConflicts:
"""Test that warnings are shown for conflicting textbox parameters."""
def test_warning_with_custom_textbox_and_submit_btn(self):
"""Should warn when submit_btn is set on ChatInterface with custom Textbox."""
with pytest.warns(UserWarning, match="submit_btn.*will be ignored"):
gr.ChatInterface(
chat,
textbox=gr.Textbox(placeholder="Custom textbox"),
submit_btn="submit",
)
def test_warning_with_custom_textbox_and_stop_btn(self):
"""Should warn when stop_btn is set on ChatInterface with custom Textbox."""
with pytest.warns(UserWarning, match="stop_btn.*will be ignored"):
gr.ChatInterface(
chat,
textbox=gr.Textbox(placeholder="Custom textbox"),
stop_btn="Stop",
)
def test_warning_with_multiple_conflicts(self):
"""Should warn about all conflicting parameters."""
with pytest.warns(UserWarning, match="submit_btn.*stop_btn.*will be ignored"):
gr.ChatInterface(
chat,
textbox=gr.Textbox(placeholder="Custom textbox"),
submit_btn="Send",
stop_btn="Stop",
)
def test_no_warning_when_params_set_on_textbox(self):
"""Should NOT warn when params are correctly set on the textbox itself."""
with warnings.catch_warnings():
warnings.simplefilter("error")
gr.ChatInterface(
chat,
textbox=gr.Textbox(placeholder="Custom textbox", submit_btn="submit"),
)
def test_no_warning_without_custom_textbox(self):
"""Should NOT warn when using default textbox with ChatInterface params."""
with warnings.catch_warnings():
warnings.simplefilter("error")
gr.ChatInterface(
chat,
submit_btn="Submit",
)
def test_no_warning_when_textbox_already_has_matching_value(self):
"""Should NOT warn if textbox already has the same value as ChatInterface param."""
with warnings.catch_warnings():
warnings.simplefilter("error")
gr.ChatInterface(
chat,
textbox=gr.Textbox(placeholder="Test", submit_btn="Send"),
submit_btn="Send",
)
def test_warning_with_multimodal_textbox(self):
"""Should warn for MultimodalTextbox conflicts too."""
with pytest.warns(UserWarning, match="stop_btn.*will be ignored"):
gr.ChatInterface(
chat,
multimodal=True,
textbox=gr.MultimodalTextbox(placeholder="Custom"),
stop_btn="Stop",
)
def test_no_warning_multimodal_with_correct_usage(self):
"""Should NOT warn when MultimodalTextbox params are set correctly."""
with warnings.catch_warnings():
warnings.simplefilter("error")
gr.ChatInterface(
chat,
multimodal=True,
textbox=gr.MultimodalTextbox(placeholder="Custom", stop_btn="Stop"),
)