diff --git a/.github/workflows/dev.yml b/.github/workflows/dev.yml index 8374f99..5233e68 100644 --- a/.github/workflows/dev.yml +++ b/.github/workflows/dev.yml @@ -37,6 +37,15 @@ jobs: python -m pip install --upgrade pip pip install poetry tox tox-gh-actions poetry install --extras dev + poetry add selenium + poetry add webdriver-manager + + - name: Set up ChromeDriver + uses: nanasess/setup-chromedriver@v2 + + - name: Set display environment variable(for ubuntu) + if: runner.os == 'Linux' + run: export DISPLAY=:99 - name: test with tox run: diff --git a/gui_agent_loop_core/backend/sandbox/sandbox_server_impl_gradio.py b/gui_agent_loop_core/backend/sandbox/sandbox_server_impl_gradio.py index adf7b44..a1ee11c 100644 --- a/gui_agent_loop_core/backend/sandbox/sandbox_server_impl_gradio.py +++ b/gui_agent_loop_core/backend/sandbox/sandbox_server_impl_gradio.py @@ -13,7 +13,7 @@ ) -def sandbox_server(interpreter_manager: InterpreterManager): +def sandbox_server(interpreter_manager: InterpreterManager, test_mode: bool = False): with gr.Blocks() as app: component_dict = get_gui_common_component(GuiBackendType.GRADIO, interpreter_manager.create_session_instance()) agent_name = component_dict[GuiComponentName.AGENT_NAME] @@ -36,8 +36,11 @@ def _change_agent(agent_name): outputs=[agent_name_radio], ) - app.queue() - app.launch(server_name="0.0.0.0", debug=True) + if not test_mode: + app.queue() + app.launch(server_name="0.0.0.0", debug=True) + else: + return app class DummyGuiAgentInterpreter(GuiAgentInterpreterABC): diff --git a/tests/gui_agent_loop_core/backend/test_gradio_integration.py b/tests/gui_agent_loop_core/backend/test_gradio_integration.py new file mode 100644 index 0000000..e9e70e2 --- /dev/null +++ b/tests/gui_agent_loop_core/backend/test_gradio_integration.py @@ -0,0 +1,87 @@ +import os +import tempfile +import time +import unittest +from unittest.mock import patch + +from selenium import webdriver +from selenium.webdriver.chrome.options import Options +from selenium.webdriver.chrome.service import Service +from selenium.webdriver.common.by import By +from selenium.webdriver.common.keys import Keys +from selenium.webdriver.support import expected_conditions as EC +from selenium.webdriver.support.ui import WebDriverWait +from webdriver_manager.chrome import ChromeDriverManager + +from gui_agent_loop_core.backend.sandbox.sandbox_server_impl_gradio import DummyGuiAgentInterpreter, sandbox_server +from gui_agent_loop_core.core.interpreter_manager import InterpreterManager + + +class TestGradioIntegration(unittest.TestCase): + @classmethod + def setUpClass(cls): + # 一意の `user-data-dir` を作成（競合防止） + user_data_dir = tempfile.mkdtemp() + + # Chrome オプションを設定 + options = Options() + options.add_argument("--headless") # UIなしで動作（CI環境向け） + options.add_argument("--no-sandbox") # サンドボックスを無効化（権限問題回避） + options.add_argument("--disable-dev-shm-usage") # 共有メモリ問題を回避 + options.add_argument(f"--user-data-dir={user_data_dir}") # 競合しないプロファイルを使用 + + try: + # `webdriver_manager` を使用して `chromedriver` を取得 + service = Service(ChromeDriverManager().install()) + cls.driver = webdriver.Chrome(service=service, options=options) + except Exception as e: + print(f"WebDriver の起動に失敗しました: {e}") + raise # エラーを発生させてテストを中断 + + @classmethod + def tearDownClass(cls): + if hasattr(cls, "driver"): + cls.driver.quit() # Chrome を適切に終了 + + def setUp(self): + interpreter = DummyGuiAgentInterpreter() + interpreter_manager = InterpreterManager(interpreter) + self.app = sandbox_server(interpreter_manager, test_mode=True) + self.app.launch(share=True) + + def tearDown(self): + self.app.close() + + def test_gradio_ui_interactions(self): + input_box = self.driver.find_element(By.TAG_NAME, "input") + input_box.send_keys("Hello, how are you?") + input_box.send_keys(Keys.RETURN) + + wait = WebDriverWait(self.driver, 10) + response = wait.until(EC.presence_of_element_located((By.CLASS_NAME, "output"))) + + self.assertIn("chat_core response", response.text) + + def test_streaming_response_from_llm(self): + input_box = self.driver.find_element(By.TAG_NAME, "input") + input_box.send_keys("Stream this response") + input_box.send_keys(Keys.RETURN) + + wait = WebDriverWait(self.driver, 10) + response = wait.until(EC.presence_of_element_located((By.CLASS_NAME, "output"))) + + self.assertIn("chat_core response", response.text) + + def test_synchronous_response_from_llm(self): + input_box = self.driver.find_element(By.TAG_NAME, "input") + input_box.send_keys("Give me a synchronous response") + input_box.send_keys(Keys.RETURN) + + wait = WebDriverWait(self.driver, 10) + response = wait.until(EC.presence_of_element_located((By.CLASS_NAME, "output"))) + + self.assertIn("chat_core response", response.text) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/gui_agent_loop_core/backend/test_gradio_ui_integration.py b/tests/gui_agent_loop_core/backend/test_gradio_ui_integration.py new file mode 100644 index 0000000..e9e70e2 --- /dev/null +++ b/tests/gui_agent_loop_core/backend/test_gradio_ui_integration.py @@ -0,0 +1,87 @@ +import os +import tempfile +import time +import unittest +from unittest.mock import patch + +from selenium import webdriver +from selenium.webdriver.chrome.options import Options +from selenium.webdriver.chrome.service import Service +from selenium.webdriver.common.by import By +from selenium.webdriver.common.keys import Keys +from selenium.webdriver.support import expected_conditions as EC +from selenium.webdriver.support.ui import WebDriverWait +from webdriver_manager.chrome import ChromeDriverManager + +from gui_agent_loop_core.backend.sandbox.sandbox_server_impl_gradio import DummyGuiAgentInterpreter, sandbox_server +from gui_agent_loop_core.core.interpreter_manager import InterpreterManager + + +class TestGradioIntegration(unittest.TestCase): + @classmethod + def setUpClass(cls): + # 一意の `user-data-dir` を作成（競合防止） + user_data_dir = tempfile.mkdtemp() + + # Chrome オプションを設定 + options = Options() + options.add_argument("--headless") # UIなしで動作（CI環境向け） + options.add_argument("--no-sandbox") # サンドボックスを無効化（権限問題回避） + options.add_argument("--disable-dev-shm-usage") # 共有メモリ問題を回避 + options.add_argument(f"--user-data-dir={user_data_dir}") # 競合しないプロファイルを使用 + + try: + # `webdriver_manager` を使用して `chromedriver` を取得 + service = Service(ChromeDriverManager().install()) + cls.driver = webdriver.Chrome(service=service, options=options) + except Exception as e: + print(f"WebDriver の起動に失敗しました: {e}") + raise # エラーを発生させてテストを中断 + + @classmethod + def tearDownClass(cls): + if hasattr(cls, "driver"): + cls.driver.quit() # Chrome を適切に終了 + + def setUp(self): + interpreter = DummyGuiAgentInterpreter() + interpreter_manager = InterpreterManager(interpreter) + self.app = sandbox_server(interpreter_manager, test_mode=True) + self.app.launch(share=True) + + def tearDown(self): + self.app.close() + + def test_gradio_ui_interactions(self): + input_box = self.driver.find_element(By.TAG_NAME, "input") + input_box.send_keys("Hello, how are you?") + input_box.send_keys(Keys.RETURN) + + wait = WebDriverWait(self.driver, 10) + response = wait.until(EC.presence_of_element_located((By.CLASS_NAME, "output"))) + + self.assertIn("chat_core response", response.text) + + def test_streaming_response_from_llm(self): + input_box = self.driver.find_element(By.TAG_NAME, "input") + input_box.send_keys("Stream this response") + input_box.send_keys(Keys.RETURN) + + wait = WebDriverWait(self.driver, 10) + response = wait.until(EC.presence_of_element_located((By.CLASS_NAME, "output"))) + + self.assertIn("chat_core response", response.text) + + def test_synchronous_response_from_llm(self): + input_box = self.driver.find_element(By.TAG_NAME, "input") + input_box.send_keys("Give me a synchronous response") + input_box.send_keys(Keys.RETURN) + + wait = WebDriverWait(self.driver, 10) + response = wait.until(EC.presence_of_element_located((By.CLASS_NAME, "output"))) + + self.assertIn("chat_core response", response.text) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/gui_agent_loop_core/backend/test_server_impl_gradio.py b/tests/gui_agent_loop_core/backend/test_server_impl_gradio.py index 0640920..1331624 100644 --- a/tests/gui_agent_loop_core/backend/test_server_impl_gradio.py +++ b/tests/gui_agent_loop_core/backend/test_server_impl_gradio.py @@ -74,6 +74,53 @@ def test_create_interface_chat(mock_get_components, mock_interpreter_manager): # Additional assertions can be added here for specific component properties +# Integration test for Gradio UI interactions +def test_gradio_ui_interactions(mock_interpreter_manager): + app, chat_iface, chatbot = _create_interface_chat(mock_interpreter_manager) + + # Simulate user interaction + user_message = "Hello, how are you?" + history = [] + response = list(mock_interpreter_manager.chat_gradio_like(user_message, history)) + + # Assert the response + assert response == ["test_response"] + assert history == [{"role": "user", "content": user_message}, {"role": "assistant", "content": "test_response"}] + + +# Test for streaming response from LLM +def test_streaming_response_from_llm(mock_interpreter_manager): + app, chat_iface, chatbot = _create_interface_chat(mock_interpreter_manager) + + # Simulate streaming response + user_message = "Stream this response" + history = [] + response_stream = mock_interpreter_manager.chat(user_message, is_auto=False) + + # Collect the streaming response + response = [] + for chunk in response_stream: + response.append(chunk) + + # Assert the response + assert response == ["test_response"] + assert history == [{"role": "user", "content": user_message}, {"role": "assistant", "content": "test_response"}] + + +# Test for synchronous response from LLM +def test_synchronous_response_from_llm(mock_interpreter_manager): + app, chat_iface, chatbot = _create_interface_chat(mock_interpreter_manager) + + # Simulate synchronous response + user_message = "Give me a synchronous response" + history = [] + response = list(mock_interpreter_manager.chat(user_message, is_auto=False)) + + # Assert the response + assert response == ["test_response"] + assert history == [{"role": "user", "content": user_message}, {"role": "assistant", "content": "test_response"}] + + def main(): """Test runner function for local testing""" pytest.main([__file__, '-v'])