164 lines
5.9 KiB
Python
164 lines
5.9 KiB
Python
import unittest
|
|
import queue
|
|
from unittest.mock import patch
|
|
|
|
from deploy_tool.gui import DeployWorkbench, database_name_from_sql_files
|
|
import deploy_tool.gui as gui_module
|
|
|
|
|
|
class FakeWidget:
|
|
def __init__(self):
|
|
self.focused = False
|
|
self.seen = False
|
|
|
|
def focus_set(self):
|
|
self.focused = True
|
|
|
|
def see(self, index):
|
|
self.seen = index
|
|
|
|
|
|
class FakeButton:
|
|
def __init__(self):
|
|
self.state = None
|
|
|
|
def configure(self, **options):
|
|
self.state = options.get("state")
|
|
|
|
|
|
class GuiNavigationTests(unittest.TestCase):
|
|
def test_database_name_comes_from_first_sql_filename(self):
|
|
from pathlib import Path
|
|
|
|
self.assertEqual(database_name_from_sql_files([Path("db/movie.sql")]), "movie")
|
|
self.assertEqual(database_name_from_sql_files([]), "")
|
|
|
|
def test_deployment_running_state_updates_deploy_button(self):
|
|
app = DeployWorkbench.__new__(DeployWorkbench)
|
|
app.deployment_running = False
|
|
app.deploy_buttons = [FakeButton()]
|
|
set_running = getattr(DeployWorkbench, "_set_deployment_running", lambda self, running: None)
|
|
|
|
set_running(app, True)
|
|
|
|
self.assertTrue(app.deployment_running)
|
|
self.assertEqual([button.state for button in app.deploy_buttons], ["disabled"])
|
|
|
|
set_running(app, False)
|
|
|
|
self.assertFalse(app.deployment_running)
|
|
self.assertEqual([button.state for button in app.deploy_buttons], ["normal"])
|
|
|
|
def test_running_deployment_ignores_another_deploy_request(self):
|
|
app = DeployWorkbench.__new__(DeployWorkbench)
|
|
app.deployment_running = True
|
|
plan_calls = []
|
|
logs = []
|
|
app._current_plan = lambda: plan_calls.append("planned") or []
|
|
app._append_log = logs.append
|
|
|
|
DeployWorkbench.run_deploy_plan_async(app)
|
|
|
|
self.assertEqual(plan_calls, [])
|
|
self.assertTrue(any("正在执行" in line for line in logs))
|
|
|
|
def test_confirmed_deployment_locks_buttons_before_worker_starts(self):
|
|
app = DeployWorkbench.__new__(DeployWorkbench)
|
|
app.deployment_running = False
|
|
app.deploy_buttons = [FakeButton()]
|
|
app._current_plan = lambda: [object()]
|
|
app.ui_queue = object()
|
|
app._thread_log = lambda value: None
|
|
|
|
with patch.object(gui_module.messagebox, "askyesno", return_value=True), patch.object(gui_module.threading, "Thread") as thread_class:
|
|
DeployWorkbench.run_deploy_plan_async(app)
|
|
|
|
self.assertTrue(app.deployment_running)
|
|
self.assertEqual([button.state for button in app.deploy_buttons], ["disabled"])
|
|
thread_class.return_value.start.assert_called_once_with()
|
|
|
|
def test_thread_start_failure_restores_deploy_buttons(self):
|
|
app = DeployWorkbench.__new__(DeployWorkbench)
|
|
app.deployment_running = False
|
|
app.deploy_buttons = [FakeButton()]
|
|
app._current_plan = lambda: [object()]
|
|
logs = []
|
|
app._append_log = logs.append
|
|
|
|
with patch.object(gui_module.messagebox, "askyesno", return_value=True), patch.object(gui_module.threading, "Thread") as thread_class:
|
|
thread_class.return_value.start.side_effect = RuntimeError("thread unavailable")
|
|
try:
|
|
DeployWorkbench.run_deploy_plan_async(app)
|
|
except RuntimeError:
|
|
pass
|
|
|
|
self.assertFalse(app.deployment_running)
|
|
self.assertEqual([button.state for button in app.deploy_buttons], ["normal"])
|
|
self.assertTrue(any("thread unavailable" in line for line in logs))
|
|
|
|
def test_worker_exception_always_queues_deploy_done(self):
|
|
app = DeployWorkbench.__new__(DeployWorkbench)
|
|
app.ui_queue = queue.Queue()
|
|
run_worker = getattr(DeployWorkbench, "_run_deployment_worker", lambda self, plan: None)
|
|
|
|
with patch.object(gui_module, "run_steps", side_effect=RuntimeError("unexpected failure")):
|
|
run_worker(app, [object()])
|
|
|
|
events = []
|
|
while not app.ui_queue.empty():
|
|
events.append(app.ui_queue.get_nowait())
|
|
|
|
self.assertIn(("deploy_done", 1), events)
|
|
self.assertTrue(any(event == "log" and "unexpected failure" in str(payload) for event, payload in events))
|
|
|
|
def test_deploy_done_event_restores_deploy_buttons(self):
|
|
app = DeployWorkbench.__new__(DeployWorkbench)
|
|
app.deployment_running = True
|
|
app.deploy_buttons = [FakeButton()]
|
|
app.ui_queue = queue.Queue()
|
|
app.ui_queue.put(("deploy_done", 1))
|
|
logs = []
|
|
app._append_log = logs.append
|
|
app.after = lambda *args, **kwargs: None
|
|
|
|
DeployWorkbench._drain_queue(app)
|
|
|
|
self.assertFalse(app.deployment_running)
|
|
self.assertEqual([button.state for button in app.deploy_buttons], ["normal"])
|
|
self.assertTrue(any("退出码: 1" in line for line in logs))
|
|
|
|
def test_top_action_only_exposes_execution(self):
|
|
self.assertNotIn("plan", gui_module.TEXT)
|
|
self.assertEqual(gui_module.TEXT["deploy"], "执行部署")
|
|
|
|
def test_sidebar_navigation_focuses_logs(self):
|
|
app = DeployWorkbench.__new__(DeployWorkbench)
|
|
app.log_text = FakeWidget()
|
|
app.result_text = FakeWidget()
|
|
|
|
app.activate_sidebar_item("运行日志")
|
|
|
|
self.assertTrue(app.log_text.focused)
|
|
self.assertEqual(app.log_text.seen, "end")
|
|
|
|
def test_sidebar_navigation_focuses_database_config(self):
|
|
app = DeployWorkbench.__new__(DeployWorkbench)
|
|
app.database_widget = FakeWidget()
|
|
|
|
app.activate_sidebar_item("数据库")
|
|
|
|
self.assertTrue(app.database_widget.focused)
|
|
|
|
def test_sidebar_navigation_can_trigger_environment_check(self):
|
|
app = DeployWorkbench.__new__(DeployWorkbench)
|
|
calls = []
|
|
app.check_environment_async = lambda: calls.append("checked")
|
|
|
|
app.activate_sidebar_item("环境检测")
|
|
|
|
self.assertEqual(calls, ["checked"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|