Add Gradio integration test for UI interactions

.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:

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):

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

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

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'])