from __future__ import annotations import json import os import sys import warnings from collections.abc import Sequence from pathlib import Path from typing import Literal from unittest.mock import patch import numpy as np import pytest from gradio_client.exceptions import AppError from hypothesis import given, settings # ty: ignore[unresolved-import] from hypothesis import strategies as st # ty: ignore[unresolved-import] import gradio as gr from gradio import EventData, Request from gradio.exceptions import Error from gradio.external_utils import format_ner_list from gradio.utils import ( FileSize, SyncToAsyncIterator, UnhashableKeyDict, _parse_file_size, abspath, append_unique_suffix, assert_configs_are_equivalent_besides_ids, check_function_inputs_match, colab_check, delete_none, diff, download_if_url, error_payload, get_extension_from_file_path_or_url, get_function_description, get_function_params, get_heartbeat_rate, get_icon_path, get_type_hints, ipython_check, is_allowed_file, is_hosted_notebook, is_in_or_equal, is_special_typed_parameter, parse_escaped_json, safe_aclose_iterator, safe_deepcopy, sanitize_list_for_csv, sanitize_value_for_csv, tex2svg, validate_url, ) os.environ["GRADIO_ANALYTICS_ENABLED"] = "False" class TestUtils: @patch("IPython.get_ipython") def test_colab_check_no_ipython(self, mock_get_ipython): mock_get_ipython.return_value = None assert colab_check() is False @patch("IPython.get_ipython") def test_ipython_check_import_fail(self, mock_get_ipython): mock_get_ipython.side_effect = ImportError() assert ipython_check() is False @patch("IPython.get_ipython") def test_ipython_check_no_ipython(self, mock_get_ipython): mock_get_ipython.return_value = None assert ipython_check() is False def test_download_if_url_doesnt_crash_on_connection_error(self): in_article = "placeholder" out_article = download_if_url(in_article) assert out_article == in_article # non-printable characters are not allowed in URL address in_article = "text\twith\rnon-printable\nASCII\x00characters" out_article = download_if_url(in_article) assert out_article == in_article # only files with HTTP(S) URL can be downloaded in_article = "ftp://localhost/tmp/index.html" out_article = download_if_url(in_article) assert out_article == in_article in_article = "file:///C:/tmp/index.html" out_article = download_if_url(in_article) assert out_article == in_article # this address will raise ValueError during parsing in_article = "https://[unmatched_bracket#?:@/index.html" out_article = download_if_url(in_article) assert out_article == in_article @pytest.mark.flaky def test_download_if_url_correct_parse(self): in_article = "https://huggingface.co/datasets/gradio/custom-html-gallery/blob/main/manifest.json" out_article = download_if_url(in_article) assert out_article != in_article def test_is_hosted_notebook_false(self): assert not is_hosted_notebook() def test_kaggle_check_true_when_run_type_set(self): with patch.dict( os.environ, {"KAGGLE_KERNEL_RUN_TYPE": "Interactive"}, clear=True ): assert is_hosted_notebook() def test_assert_configs_are_equivalent(): test_dir = Path(__file__).parent / "test_files" with open(test_dir / "xray_config.json") as fp: xray_config = json.load(fp) with open(test_dir / "xray_config_diff_ids.json") as fp: xray_config_diff_ids = json.load(fp) with open(test_dir / "xray_config_wrong.json") as fp: xray_config_wrong = json.load(fp) assert assert_configs_are_equivalent_besides_ids(xray_config, xray_config) assert assert_configs_are_equivalent_besides_ids(xray_config, xray_config_diff_ids) with pytest.raises(ValueError): assert_configs_are_equivalent_besides_ids(xray_config, xray_config_wrong) def test_get_heartbeat_rate(monkeypatch): # Defaults to 15 seconds when nothing is configured. monkeypatch.delenv("GRADIO_HEARTBEAT_INTERVAL", raising=False) monkeypatch.delenv("GRADIO_IS_E2E_TEST", raising=False) assert get_heartbeat_rate() == 15 # GRADIO_HEARTBEAT_INTERVAL overrides the default with a float value. monkeypatch.setenv("GRADIO_HEARTBEAT_INTERVAL", "1.5") assert get_heartbeat_rate() == 1.5 # It takes precedence over the GRADIO_IS_E2E_TEST fallback. monkeypatch.setenv("GRADIO_IS_E2E_TEST", "1") assert get_heartbeat_rate() == 1.5 # Falls back to the E2E rate when only GRADIO_IS_E2E_TEST is set. monkeypatch.delenv("GRADIO_HEARTBEAT_INTERVAL", raising=False) assert get_heartbeat_rate() == 0.25 # An invalid value warns and falls back to the default. monkeypatch.delenv("GRADIO_IS_E2E_TEST", raising=False) monkeypatch.setenv("GRADIO_HEARTBEAT_INTERVAL", "not-a-number") with pytest.warns(UserWarning): assert get_heartbeat_rate() == 15 # A zero or negative value would busy-loop asyncio.sleep(), so it warns # and falls back to the default. for bad_value in ("0", "-5"): monkeypatch.setenv("GRADIO_HEARTBEAT_INTERVAL", bad_value) with pytest.warns(UserWarning): assert get_heartbeat_rate() == 15 class TestFormatNERList: def test_format_ner_list_standard(self): string = "Wolfgang lives in Berlin" groups = [ {"entity_group": "PER", "start": 0, "end": 8}, {"entity_group": "LOC", "start": 18, "end": 24}, ] result = [ ("", None), ("Wolfgang", "PER"), (" lives in ", None), ("Berlin", "LOC"), ("", None), ] assert format_ner_list(string, groups) == result def test_format_ner_list_empty(self): string = "I live in a city" groups = [] result = [("I live in a city", None)] assert format_ner_list(string, groups) == result class TestDeleteNone: """Credit: https://stackoverflow.com/questions/33797126/proper-way-to-remove-keys-in-dictionary-with-none-values-in-python""" def test_delete_none(self): input = { "a": 12, "b": 34, "c": None, "k": { "d": 34, "t": None, "m": [{"k": 23, "t": None}, [None, 1, 2, 3], {1, 2, None}], None: 123, }, } truth = { "a": 12, "b": 34, "k": { "d": 34, "t": None, "m": [{"k": 23, "t": None}, [None, 1, 2, 3], {1, 2, None}], None: 123, }, } assert delete_none(input) == truth class TestSanitizeForCSV: def test_unsafe_value(self): assert sanitize_value_for_csv("=OPEN()") == "'=OPEN()" assert sanitize_value_for_csv("=1+2") == "'=1+2" assert sanitize_value_for_csv('=1+2";=1+2') == "'=1+2\";=1+2" def test_safe_value(self): assert sanitize_value_for_csv(4) == 4 assert sanitize_value_for_csv(-44.44) == -44.44 assert sanitize_value_for_csv("1+1=2") == "1+1=2" assert sanitize_value_for_csv("1aaa2") == "1aaa2" def test_list(self): assert sanitize_list_for_csv([4, "def=", "=gh+ij"]) == [4, "def=", "'=gh+ij"] assert sanitize_list_for_csv( [["=abc", "def", "gh,+ij"], ["abc", "=def", "+ghij"]] ) == [["'=abc", "def", "'gh,+ij"], ["abc", "'=def", "'+ghij"]] assert sanitize_list_for_csv([1, ["ab", "=de"]]) == [1, ["ab", "'=de"]] def test_parse_escaped_json(self): assert parse_escaped_json("-0.5678") == -0.5678 # Negative numbers get CSV-escaped with a leading "'" on write (#13591) assert parse_escaped_json("'-0.5678") == -0.5678 with pytest.raises(json.JSONDecodeError): parse_escaped_json("'not json") class TestValidateURL: @pytest.mark.flaky def test_valid_urls(self): assert validate_url("https://www.gradio.app") assert validate_url("http://gradio.dev") assert validate_url( "https://upload.wikimedia.org/wikipedia/commons/b/b0/Bengal_tiger_%28Panthera_tigris_tigris%29_female_3_crop.jpg" ) assert validate_url( "https://huggingface.co/datasets/Xenova/transformers.js-docs/resolve/main/bread_small.png" ) def test_invalid_urls(self): assert not (validate_url("C:/Users/")) assert not (validate_url("C:\\Users\\")) assert not (validate_url("/home/user")) class TestAppendUniqueSuffix: def test_no_suffix(self): name = "test" list_of_names = ["test_1", "test_2"] assert append_unique_suffix(name, list_of_names) == name def test_first_suffix(self): name = "test" list_of_names = ["test", "test_-1"] assert append_unique_suffix(name, list_of_names) == "test_1" def test_later_suffix(self): name = "test" list_of_names = ["test", "test_1", "test_2", "test_3"] assert append_unique_suffix(name, list_of_names) == "test_4" class TestAbspath: def test_abspath_no_symlink(self): resolved_path = str(abspath("../gradio/gradio/test_data/lion.jpg")) assert ".." not in resolved_path @pytest.mark.skipif( sys.platform.startswith("win"), reason="Windows doesn't allow creation of sym links without administrative privileges", ) def test_abspath_symlink_path(self): os.symlink("gradio/test_data", "gradio/test_link", True) resolved_path = str(abspath("../gradio/gradio/test_link/lion.jpg")) os.unlink("gradio/test_link") assert "test_link" in resolved_path @pytest.mark.skipif( sys.platform.startswith("win"), reason="Windows doesn't allow creation of sym links without administrative privileges", ) def test_abspath_symlink_dir(self): os.symlink("gradio/test_data", "gradio/test_link", True) full_path = os.path.join(os.getcwd(), "gradio/test_link/lion.jpg") resolved_path = str(abspath(full_path)) os.unlink("gradio/test_link") assert "test_link" in resolved_path assert full_path == resolved_path class TestGetTypeHints: def test_get_type_hints(self): class F: def __call__(self, s: str): return s class C: def f(self, s: str): return s def f(s: str): return s class GenericObject: pass test_objs = [F(), C().f, f] for x in test_objs: hints = get_type_hints(x) assert len(hints) == 1 assert hints["s"] is str assert len(get_type_hints(GenericObject())) == 0 def test_get_type_hints_with_unresolvable_forward_ref(self): """get_type_hints should return {} when annotations can't be resolved at runtime.""" def func(x: str) -> "NonExistentType": # noqa: F821, UP037 # ty: ignore[unresolved-reference] return x hints = get_type_hints(func) assert hints == {} def test_is_special_typed_parameter(self): def func(a: list[str], b: Literal["a", "b"], c, d: Request, e: Request | None): pass hints = get_type_hints(func) assert not is_special_typed_parameter("a", hints) assert not is_special_typed_parameter("b", hints) assert not is_special_typed_parameter("c", hints) assert is_special_typed_parameter("d", hints) assert is_special_typed_parameter("e", hints) def test_is_special_typed_parameter_with_pipe(self): def func(a: Request, b: str | int, c: list[str]): pass hints = get_type_hints(func) assert is_special_typed_parameter("a", hints) assert not is_special_typed_parameter("b", hints) assert not is_special_typed_parameter("c", hints) class TestCheckFunctionInputsMatch: def test_check_function_inputs_match(self): class F: def __call__(self, s: str, evt: EventData): return s class C: def f(self, s: str, evt: EventData): return s def f(s: str, evt: EventData): return s test_objs = [F(), C().f, f] with warnings.catch_warnings(): warnings.simplefilter("error") # Ensure there're no warnings raised here. for x in test_objs: check_function_inputs_match(x, [None], False) def test_tex2svg_preserves_matplotlib_backend(): import matplotlib matplotlib.use("svg") tex2svg("1+1=2") assert matplotlib.get_backend() == "svg" with pytest.raises( Exception # specifically a pyparsing.ParseException but not important here ): tex2svg("$$$1+1=2$$$") assert matplotlib.get_backend() == "svg" def test_is_in_or_equal(): assert is_in_or_equal("files/lion.jpg", "files/lion.jpg") assert is_in_or_equal("files/lion.jpg", "files") assert is_in_or_equal("files/lion.._M.jpg", "files") assert not is_in_or_equal("files", "files/lion.jpg") assert is_in_or_equal("/home/usr/notes.txt", "/home/usr/") assert not is_in_or_equal("/home/usr/subdirectory", "/home/usr/notes.txt") assert not is_in_or_equal("/home/usr/../../etc/notes.txt", "/home/usr/") assert not is_in_or_equal("/safe_dir/subdir/../../unsafe_file.txt", "/safe_dir/") @pytest.mark.skipif( sys.platform.startswith("win"), reason="Windows doesn't support POSIX double-slash notation", ) def test_is_in_or_equal_posix_specific_paths(): assert is_in_or_equal("//foo/..a", "//foo") assert is_in_or_equal("//foo/asd/", "/foo") assert is_in_or_equal("//foo/..²", "/foo") def create_path_string(): return st.lists( st.one_of( st.text( alphabet="ab@1/(", min_size=1, ), st.just(".."), st.just("."), ), min_size=1, max_size=10, # Limit depth to avoid excessively long paths ).map(lambda x: os.path.join(*x).replace("(", "..")) def create_path_list(): return st.lists(create_path_string(), min_size=0, max_size=5) def my_check(path_1, path_2): try: path_1 = Path(path_1).resolve() path_2 = Path(path_2).resolve() _ = path_1.relative_to(path_2) return True except ValueError: return False @settings(derandomize=os.getenv("CI") is not None) @given( path_1=create_path_string(), path_2=create_path_string(), ) def test_is_in_or_equal_fuzzer(path_1, path_2): try: # Convert to absolute paths abs_path_1 = abspath(path_1) abs_path_2 = abspath(path_2) result = is_in_or_equal(abs_path_1, abs_path_2) assert result == my_check(abs_path_1, abs_path_2) except Exception as e: pytest.fail(f"Exception raised: {e}") @settings(derandomize=os.getenv("CI") is not None) @given( path=create_path_string(), blocked_paths=create_path_list(), allowed_paths=create_path_list(), created_paths=create_path_list(), ) def test_is_allowed_file_fuzzer( path: Path, blocked_paths: Sequence[Path], allowed_paths: Sequence[Path], created_paths: Sequence[Path], ): result, reason = is_allowed_file(path, blocked_paths, allowed_paths, created_paths) assert isinstance(result, bool) assert reason in [ "in_blocklist", "allowed", "not_created_or_allowed", "created", ] if result: assert reason in ("allowed", "created") elif reason == "in_blocklist": assert any(is_in_or_equal(path, blocked_path) for blocked_path in blocked_paths) elif reason == "not_created_or_allowed": assert not any( is_in_or_equal(path, allowed_path) for allowed_path in allowed_paths ) if reason == "allowed": assert any(is_in_or_equal(path, allowed_path) for allowed_path in allowed_paths) elif reason == "created": assert any(is_in_or_equal(path, created_path) for created_path in created_paths) @pytest.mark.parametrize( "path,blocked_paths,allowed_paths", [ ("/a/foo.txt", ["/a"], ["/b"], False), ("/b/foo.txt", ["/a"], ["/b"], True), ("/a/../c/foo.txt", ["/c/"], ["/a/"], False), ("/c/../a/foo.txt", ["/c/"], ["/a/"], True), ("/c/foo.txt", ["/c/"], ["/c/foo.txt"], True), ], ) def is_allowed_file_corner_cases(path, blocked_paths, allowed_paths, result): assert is_allowed_file(path, blocked_paths, allowed_paths, []) == result # Additional test for known edge cases @pytest.mark.parametrize( "path_1,path_2,expected", [ ("/AAA/a/../a", "/AAA", True), ("//AA/a", "/tmp", False), ("/AAA/..", "/AAA", False), ("/a/b/c", "/d/e/f", False), (".", "..", True), ("..", ".", False), ("/a/b/./c", "/a/b", True), ("/a/b/../c", "/a", True), ("/a/b/c", "/a/b/c/../d", False), ("/", "/a", False), ("/a", "/", True), ], ) def test_is_in_or_equal_edge_cases(path_1, path_2, expected): assert is_in_or_equal(path_1, path_2) == expected @pytest.mark.parametrize( "path_or_url, extension", [ ("https://example.com/avatar/xxxx.mp4?se=2023-11-16T06:51:23Z&sp=r", "mp4"), ("/home/user/documents/example.pdf", "pdf"), ("C:\\Users\\user\\documents\\example.png", "png"), ("C:/Users/user/documents/example", ""), ], ) def test_get_extension_from_file_path_or_url(path_or_url, extension): assert get_extension_from_file_path_or_url(path_or_url) == extension @pytest.mark.parametrize( "old, new, expected_diff", [ ({"a": 1, "b": 2}, {"a": 1, "b": 2}, []), ({}, {"a": 1, "b": 2}, [["add", ["a"], 1], ["add", ["b"], 2]]), (["a", "b"], {"a": 1, "b": 2}, [["replace", [], {"a": 1, "b": 2}]]), ("abc", "abcdef", [["append", [], "def"]]), ( [ {"role": "user", "content": "Hello!", "metadata": {"id": 1}}, {"role": "assistant", "content": "b"}, ], [ { "role": "assistant", "content": "Thinking...", "metadata": {"title": "Thinking..."}, }, {"role": "assistant", "content": "b"}, ], [ ["replace", [0, "role"], "assistant"], ["replace", [0, "content"], "Thinking..."], ["delete", [0, "metadata", "id"], None], ["add", [0, "metadata", "title"], "Thinking..."], ], ), ( [ {"role": "user", "content": "Hello!", "metadata": {"id": 1}}, {"role": "assistant", "content": "b"}, ], [ {"role": "user", "content": "No metadata", "metadata": {}}, {"role": "assistant", "content": "b"}, ], [ ["replace", [0, "content"], "No metadata"], ["delete", [0, "metadata", "id"], None], ], ), ( {"data": [[1, 1], [2, 2], [3, 3]]}, {"data": [[1, 1]]}, [ ["delete", ["data", 1], None], ["delete", ["data", 1], None], ], ), ], ) def test_diff(old, new, expected_diff): assert diff(old, new) == expected_diff class TestFunctionParams: def test_regular_function(self): def func(a: int, b: int = 10, c: str = "default", d=None): pass assert get_function_params(func) == [ ("a", False, None, int), ("b", True, 10, int), ("c", True, "default", str), ("d", True, None, None), ] def test_function_no_params(self): def func(): pass assert get_function_params(func) == [] def test_lambda_function(self): assert get_function_params(lambda x, y: x + y) == [ ("x", False, None, None), ("y", False, None, None), ] def test_function_with_args(self): def func(a, *args): pass assert get_function_params(func) == [("a", False, None, None)] def test_function_with_kwargs(self): def func(a, **kwargs): pass assert get_function_params(func) == [("a", False, None, None)] def test_function_with_special_args(self): def func(a, r: Request, b=10): pass assert get_function_params(func) == [ ("a", False, None, None), ("b", True, 10, None), ] def func2(a, r: Request | None = None, b="abc"): pass assert get_function_params(func2) == [ ("a", False, None, None), ("b", True, "abc", None), ] def test_class_method_skip_first_param(self): class MyClass: def method(self, arg1, arg2=42): pass assert get_function_params(MyClass().method) == [ ("arg1", False, None, None), ("arg2", True, 42, None), ] def test_static_method_no_skip(self): class MyClass: @staticmethod def method(arg1, arg2=42): pass assert get_function_params(MyClass.method) == [ ("arg1", False, None, None), ("arg2", True, 42, None), ] def test_class_method_with_args(self): class MyClass: def method(self, a, *args, b=42): pass assert get_function_params(MyClass().method) == [("a", False, None, None)] def test_lambda_with_args(self): assert get_function_params(lambda x, *args: x) == [("x", False, None, None)] def test_lambda_with_kwargs(self): assert get_function_params(lambda x, **kwargs: x) == [("x", False, None, None)] def test_parse_file_size(): assert _parse_file_size("1kb") == 1 * FileSize.KB assert _parse_file_size("1mb") == 1 * FileSize.MB assert _parse_file_size("505 Mb") == 505 * FileSize.MB class TestUnhashableKeyDict: def test_set_get_simple(self): d = UnhashableKeyDict() d["a"] = 1 assert d["a"] == 1 def test_set_get_unhashable(self): d = UnhashableKeyDict() key = [1, 2, 3] key2 = [1, 2, 3] d[key] = "value" assert d[key] == "value" assert d[key2] == "value" def test_set_get_numpy_array(self): d = UnhashableKeyDict() key = np.array([1, 2, 3]) key2 = np.array([1, 2, 3]) d[key] = "numpy value" assert d[key2] == "numpy value" def test_overwrite(self): d = UnhashableKeyDict() d["key"] = "old" d["key"] = "new" assert d["key"] == "new" def test_delete(self): d = UnhashableKeyDict() d["key"] = "value" del d["key"] assert len(d) == 0 with pytest.raises(KeyError): d["key"] def test_delete_nonexistent(self): d = UnhashableKeyDict() with pytest.raises(KeyError): del d["nonexistent"] def test_len(self): d = UnhashableKeyDict() assert len(d) == 0 d["a"] = 1 d["b"] = 2 assert len(d) == 2 def test_contains(self): d = UnhashableKeyDict() d["key"] = "value" assert "key" in d assert "nonexistent" not in d def test_get_nonexistent(self): d = UnhashableKeyDict() with pytest.raises(KeyError): d["nonexistent"] class TestSafeDeepCopy: def test_safe_deepcopy_dict(self): original = {"key1": [1, 2, {"nested_key": "value"}], "key2": "simple_string"} copied = safe_deepcopy(original) assert copied == original assert copied is not original assert copied["key1"] is not original["key1"] assert copied["key1"][2] is not original["key1"][2] def test_safe_deepcopy_list(self): original = [1, 2, [3, 4, {"key": "value"}]] copied = safe_deepcopy(original) assert copied == original assert copied is not original assert copied[2] is not original[2] assert copied[2][2] is not original[2][2] # type: ignore def test_safe_deepcopy_custom_object(self): class CustomClass: def __init__(self, value): self.value = value original = CustomClass(10) copied = safe_deepcopy(original) assert copied.value == original.value assert copied is not original def test_safe_deepcopy_handles_undeepcopyable(self): class Uncopyable: def __deepcopy__(self, memo): raise TypeError("Can't deepcopy") original = Uncopyable() result = safe_deepcopy(original) assert result is not original assert type(result) is type(original) def test_get_icon_path(): assert get_icon_path("plus.svg").endswith("plus.svg") assert get_icon_path("huggingface-logo.svg").endswith("huggingface-logo.svg") def test_error_payload(): result = error_payload(None, False) assert result == {"error": None} result = error_payload(Exception("test error"), True) assert result == {"error": "test error", "visible": True} gr_error = Error("custom error", duration=1.5, visible=True, title="Error Title") result = error_payload(gr_error, False) assert result == { "error": "custom error", "duration": 1.5, "visible": True, "title": "Error Title", } app_error = AppError("custom error") result = error_payload(app_error, False) assert result == { "error": "custom error", "duration": 10, "visible": True, "title": "Error", } class TestGetFunctionDescription: def test_basic_function_description(self): def test_func(param1, param2): """This is a test function. Args: param1: First parameter param2: Second parameter Returns: - First return value - Second return value """ pass description, parameters, returns = get_function_description(test_func) assert description == "This is a test function." assert parameters == {"param1": "First parameter", "param2": "Second parameter"} assert returns == ["- First return value", "- Second return value"] def test_function_with_extended_returns(self): def test_func(param1, param2): """This is a test function. Args: param1: First parameter param2: Second parameter Returns: - First return value - Second return value Examples: - Example 1 - Example 2 """ pass description, parameters, returns = get_function_description(test_func) assert description == "This is a test function." assert parameters == {"param1": "First parameter", "param2": "Second parameter"} assert returns == [ "- First return value", "- Second return value", "Examples:", "- Example 1", "- Example 2", ] def test_function_with_no_docstring(self): def test_func(): pass description, parameters, returns = get_function_description(test_func) assert description == "" assert parameters == {} assert returns == [] def test_function_with_no_parameters(self): def test_func(): """This is a test function with no parameters.""" pass description, parameters, returns = get_function_description(test_func) assert description == "This is a test function with no parameters." assert parameters == {} assert returns == [] def test_function_with_alternate_parameter_section(self): def test_func(param1, param2): """This is a test function. Parameters: param1: First parameter param2: Second parameter """ pass description, parameters, returns = get_function_description(test_func) assert description == "This is a test function." assert parameters == {"param1": "First parameter", "param2": "Second parameter"} assert returns == [] def test_function_with_arguments_section(self): def test_func(param1, param2): """This is a test function. Arguments: param1: First parameter param2: Second parameter Returns: x: First return value y: Second return value """ pass description, parameters, returns = get_function_description(test_func) assert description == "This is a test function." assert parameters == {"param1": "First parameter", "param2": "Second parameter"} assert returns == ["x: First return value", "y: Second return value"] def test_function_with_multiline_description(self): def test_func(param1, param2): """This is a test function. It has multiple lines of description. Args: param1: First parameter param2: Second parameter """ pass description, parameters, returns = get_function_description(test_func) assert ( description == "This is a test function. It has multiple lines of description." ) assert parameters == {"param1": "First parameter", "param2": "Second parameter"} assert returns == [] def test_function_with_missing_params(self): def test_func(param1, param2, param3): """This is a test function. Args: param1: description1 param3: description3 Returns: - First return value - Second return value """ pass description, parameters, returns = get_function_description(test_func) assert description == "This is a test function." assert parameters == { "param1": "description1", "param2": "", "param3": "description3", } assert returns == ["- First return value", "- Second return value"] def test_function_with_nested_colons(self): def test_func(param1, param2): """This is a test function. Args: param1: description1: with nested colon param2: description2 """ pass description, parameters, returns = get_function_description(test_func) assert description == "This is a test function." assert parameters == { "param1": "description1: with nested colon", "param2": "description2", } assert returns == [] def test_get_function_description_inherits_parent_docstring(self): """ Test that get_function_description correctly retrieves docstrings from a method's parent class if the method itself has none. """ class Parent: def method(self): """This is the docstring from the parent class.""" pass class Child(Parent): def method(self): pass # No docstring here description, parameters, returns = get_function_description(Child().method) assert parameters == {} assert returns == [] assert description == "This is the docstring from the parent class." class TestConnectHeartbeat: def test_unload_registered_last(self): with gr.Blocks() as demo: msg = gr.Markdown("# Test") gr.Button("Click").click(lambda: "# TADA!", outputs=[msg]) demo.unload(lambda: None) config = demo.get_config_file() assert config["connect_heartbeat"] is True def test_unload_registered_first(self): with gr.Blocks() as demo: demo.unload(lambda: None) msg = gr.Markdown("# Test") gr.Button("Click").click(lambda: "# TADA!", outputs=[msg]) config = demo.get_config_file() assert config["connect_heartbeat"] is True def test_unload_registered_in_middle(self): with gr.Blocks() as demo: msg = gr.Markdown("# Test") gr.Button("Click1").click(lambda: "# 1", outputs=[msg]) demo.unload(lambda: None) gr.Button("Click2").click(lambda: "# 2", outputs=[msg]) config = demo.get_config_file() assert config["connect_heartbeat"] is True def test_no_unload_no_heartbeat(self): with gr.Blocks() as demo: msg = gr.Markdown("# Test") gr.Button("Click").click(lambda: "# TADA!", outputs=[msg]) config = demo.get_config_file() assert config["connect_heartbeat"] is False def test_per_session_manual_cache_connects_heartbeat(self): def predict(value, cache=gr.Cache(per_session=True)): hit = cache.get(value) if hit is not None: return hit["value"] cache.set(value, value=value) return value with gr.Blocks() as demo: text = gr.Textbox() output = gr.Textbox() gr.Button("Click").click(predict, [text], [output]) config = demo.get_config_file() assert config["connect_heartbeat"] is True def test_per_session_decorator_cache_connects_heartbeat(self): @gr.cache(per_session=True) def predict(value): return value with gr.Blocks() as demo: text = gr.Textbox() output = gr.Textbox() gr.Button("Click").click(predict, [text], [output]) config = demo.get_config_file() assert config["connect_heartbeat"] is True class _FakeIterator: """A fake iterator with a controllable close() for testing.""" def __init__(self, fail_count=0): self.close_call_count = 0 self._fail_count = fail_count self._closed = False def close(self): self.close_call_count += 1 if self.close_call_count <= self._fail_count: raise ValueError("generator already executing") self._closed = True def __next__(self): raise StopIteration def __iter__(self): return self class TestSyncToAsyncIteratorAclose: @pytest.mark.asyncio async def test_aclose_closes_iterator(self): """aclose() should call close() on the underlying sync iterator.""" def gen(): yield 1 yield 2 iterator = SyncToAsyncIterator(gen(), limiter=None) await iterator.aclose() assert list(iterator.iterator) == [] @pytest.mark.asyncio async def test_aclose_retries_on_already_executing(self): """aclose() should retry when ValueError('generator already executing') is raised.""" fake = _FakeIterator(fail_count=2) iterator = SyncToAsyncIterator(fake, limiter=None) await iterator.aclose(retry_interval=0.01) assert fake.close_call_count == 3 assert fake._closed is True @pytest.mark.asyncio async def test_aclose_timeout_raises(self): """aclose() should raise ValueError after timeout is exceeded.""" fake = _FakeIterator(fail_count=999) iterator = SyncToAsyncIterator(fake, limiter=None) with pytest.raises(ValueError, match="already executing"): await iterator.aclose(timeout=0.1, retry_interval=0.02) @pytest.mark.asyncio async def test_aclose_raises_other_value_error(self): """aclose() should not retry on ValueError without 'already executing'.""" class _BadIterator: def close(self): raise ValueError("some other error") iterator = SyncToAsyncIterator(_BadIterator(), limiter=None) with pytest.raises(ValueError, match="some other error"): await iterator.aclose() class TestSafeAcloseIterator: @pytest.mark.asyncio async def test_delegates_to_aclose(self): """safe_aclose_iterator() should call aclose() on the iterator.""" def gen(): yield 1 iterator = SyncToAsyncIterator(gen(), limiter=None) await safe_aclose_iterator(iterator) assert list(iterator.iterator) == [] @pytest.mark.asyncio async def test_works_with_async_generator(self): """safe_aclose_iterator() should work with native async generators.""" async def gen(): yield 1 yield 2 ag = gen() await safe_aclose_iterator(ag) # After aclose, async generator should raise StopAsyncIteration with pytest.raises(StopAsyncIteration): await ag.__anext__()