291 lines
11 KiB
Python
291 lines
11 KiB
Python
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()
|