from __future__ import annotations import unittest from unittest.mock import patch from app.services.ops_migration_service import execute_ops_migration, preview_ops_migration class OpsMigrationServiceTestCase(unittest.TestCase): @patch("app.services.ops_migration_service._collect_remote_checks") @patch("app.services.ops_migration_service._inspect_target_database") @patch("app.services.ops_migration_service._find_managed_node") def test_preview_returns_blocking_reason_when_remote_tools_missing( self, mock_find_managed_node, mock_inspect_target_database, mock_collect_remote_checks, ) -> None: mock_find_managed_node.return_value = { "node_code": "target-a", "ssh_host": "10.0.0.8", "ssh_user": "root", "ssh_port": 22, "auth_mode": "key", } mock_collect_remote_checks.return_value = { "tools": { "python3": True, "node": False, "npm": False, "systemctl": True, "psql": True, "pg_dump": True, "curl": True, }, "paths": { "repo_exists": True, "repo_git": True, "domain_root_exists": True, "api_root_exists": True, "web_root_exists": True, }, "remote_db_config": { "DB_HOST": "127.0.0.1", "DB_PORT": "5432", "DB_DATABASE": "domain", "DB_USER": "domainuser", }, "blocking_reasons": ["目标机缺少 node。", "目标机缺少 npm。"], "warnings": [], } mock_inspect_target_database.return_value = {"available": False} ok, message, data = preview_ops_migration({"target_node_code": "target-a"}) self.assertFalse(ok) self.assertEqual("迁移预检查未通过", message) self.assertIn("目标机缺少 node。", data["blocking_reasons"]) self.assertIn("目标机缺少 npm。", data["blocking_reasons"]) @patch("app.services.ops_migration_service._collect_remote_checks") @patch("app.services.ops_migration_service._inspect_target_database") @patch("app.services.ops_migration_service._find_managed_node") def test_preview_uses_remote_env_as_target_db_default( self, mock_find_managed_node, mock_inspect_target_database, mock_collect_remote_checks, ) -> None: mock_find_managed_node.return_value = { "node_code": "target-a", "ssh_host": "10.0.0.8", "ssh_user": "root", "ssh_port": 22, "auth_mode": "key", } mock_collect_remote_checks.return_value = { "tools": { "python3": True, "node": True, "npm": True, "systemctl": True, "psql": True, "pg_dump": True, "curl": True, }, "paths": { "repo_exists": True, "repo_git": True, "domain_root_exists": True, "api_root_exists": True, "web_root_exists": True, }, "git_commit": "abc123", "remote_db_config": { "DB_HOST": "127.0.0.1", "DB_PORT": "5433", "DB_DATABASE": "domain_remote", "DB_USER": "remote_user", }, "blocking_reasons": [], "warnings": [], } mock_inspect_target_database.return_value = { "available": True, "has_business_data": False, } ok, message, data = preview_ops_migration({"target_node_code": "target-a", "overwrite_database": True}) self.assertTrue(ok) self.assertEqual("迁移预检查完成", message) self.assertEqual("domain_remote", data["target_db_config"]["database"]) self.assertEqual("remote_user", data["target_db_config"]["user"]) self.assertEqual(5433, data["target_db_config"]["port"]) self.assertTrue(bool(data["execution_guard"]["token"])) @patch("app.services.ops_migration_service._run_remote_health_check") @patch("app.services.ops_migration_service._restart_remote_services") @patch("app.services.ops_migration_service._build_remote_frontend") @patch("app.services.ops_migration_service._inspect_target_database") @patch("app.services.ops_migration_service._collect_remote_checks") @patch("app.services.ops_migration_service._find_managed_node") def test_execute_rejects_missing_confirmation_text_for_nonempty_target_db( self, mock_find_managed_node, mock_collect_remote_checks, mock_inspect_target_database, mock_build_remote_frontend, mock_restart_remote_services, mock_run_remote_health_check, ) -> None: mock_find_managed_node.return_value = { "node_code": "target-a", "ssh_host": "10.0.0.8", "ssh_user": "root", "ssh_port": 22, "auth_mode": "key", } mock_collect_remote_checks.return_value = { "tools": { "python3": True, "node": True, "npm": True, "systemctl": True, "psql": True, "pg_dump": True, "curl": True, }, "paths": { "repo_exists": True, "repo_git": True, "domain_root_exists": True, "api_root_exists": True, "web_root_exists": True, }, "git_commit": "abc123", "remote_db_config": { "DB_HOST": "127.0.0.1", "DB_PORT": "5432", "DB_DATABASE": "domain_remote", "DB_USER": "remote_user", }, "blocking_reasons": [], "warnings": [], } mock_inspect_target_database.return_value = { "available": True, "database": "domain_remote", "has_business_data": True, "public_table_count": 10, "business_table_count": 5, } mock_build_remote_frontend.return_value = (True, "ok", {}) mock_restart_remote_services.return_value = (True, "ok", {}) mock_run_remote_health_check.return_value = (True, "ok", {}) preview_ok, _preview_message, preview_data = preview_ops_migration( {"target_node_code": "target-a", "overwrite_database": True} ) self.assertTrue(preview_ok) token = preview_data["execution_guard"]["token"] required_confirmation_text = preview_data["execution_guard"]["required_confirmation_text"] self.assertEqual("OVERWRITE domain_remote", required_confirmation_text) execute_ok, execute_message, execute_data = execute_ops_migration( { "target_node_code": "target-a", "overwrite_database": True, "execute_confirmation_token": token, } ) self.assertFalse(execute_ok) self.assertEqual("缺少数据库覆盖确认文案，执行被拒绝。", execute_message) self.assertIn("missing execute_confirmation_text", execute_data["blocking_reasons"]) @patch("app.services.ops_migration_service._start_migration_dispatch_thread") @patch("app.services.ops_migration_service.create_ops_job") @patch("app.services.ops_migration_service._collect_remote_checks") @patch("app.services.ops_migration_service._inspect_target_database") @patch("app.services.ops_migration_service._find_managed_node") def test_execute_creates_async_job_for_long_running_migration( self, mock_find_managed_node, mock_inspect_target_database, mock_collect_remote_checks, mock_create_ops_job, mock_start_thread, ) -> None: mock_find_managed_node.return_value = { "node_code": "target-a", "ssh_host": "10.0.0.8", "ssh_user": "root", "ssh_port": 22, "auth_mode": "key", } mock_collect_remote_checks.return_value = { "tools": { "python3": True, "node": True, "npm": True, "systemctl": True, "psql": True, "pg_dump": True, "curl": True, }, "paths": { "repo_exists": True, "repo_git": True, "domain_root_exists": True, "api_root_exists": True, "web_root_exists": True, }, "git_commit": "abc123", "remote_db_config": { "DB_HOST": "127.0.0.1", "DB_PORT": "5432", "DB_DATABASE": "domain_remote", "DB_USER": "remote_user", }, "blocking_reasons": [], "warnings": [], } mock_inspect_target_database.return_value = { "available": True, "database": "domain_remote", "has_business_data": False, "public_table_count": 0, "business_table_count": 0, } mock_create_ops_job.return_value = ( True, "ok", { "job": { "id": 88, "job_code": "ops-20260422160000-abc123", "action": "migration.execute", "status": "queued", "target_node_code": "target-a", } }, ) preview_ok, _preview_message, preview_data = preview_ops_migration({"target_node_code": "target-a"}) self.assertTrue(preview_ok) token = preview_data["execution_guard"]["token"] execute_ok, execute_message, execute_data = execute_ops_migration( { "target_node_code": "target-a", "execute_confirmation_token": token, } ) self.assertTrue(execute_ok) self.assertEqual("迁移任务已创建，后台开始执行。", execute_message) self.assertEqual(88, execute_data["job"]["id"]) self.assertEqual("migration.execute", mock_create_ops_job.call_args.args[0]["action"]) self.assertFalse(bool(mock_create_ops_job.call_args.args[0]["run_now"])) self.assertEqual("control-plane", mock_create_ops_job.call_args.args[0]["execution_mode"]) self.assertEqual("", mock_create_ops_job.call_args.args[0]["payload"]["target_db_password"]) mock_start_thread.assert_called_once_with(88) if __name__ == "__main__": unittest.main()