补齐一些获取用户名密码的代码 (#2918)

Author: LeoQuote

sql/binlog.py

@@ -134,12 +134,12 @@ def my2sql(request):
 
     # 提交给my2sql进行解析
     my2sql = My2SQL()
-
+    username, password = instance.get_username_password()
     # 准备参数
     args = {
         "host": instance.host,
-        "user": instance.user,
-        "password": instance.password,
+        "user": username,
+        "password": password,
         "port": instance.port,
         "work-type": work_type,
         "start-file": start_file,

sql/engines/__init__.py

@@ -49,6 +49,7 @@ def __del__(self):
             del self.remotessh
 
     def remote_instance_conn(self, instance=None):
+        user, password = instance.get_username_password()
         # 判断如果配置了隧道则连接隧道
         if not hasattr(self, "remotessh") and instance.tunnel:
             self.remotessh = SSHConnection(
@@ -62,13 +63,14 @@ def remote_instance_conn(self, instance=None):
                 instance.tunnel.pkey_password,
             )
             self.remote_host, self.remote_port = self.remotessh.get_ssh()
-            self.remote_user = instance.user
-            self.remote_password = instance.password
+            user, password = instance.get_username_password()
+            self.remote_user = user
+            self.remote_password = password
         elif not instance.tunnel:
             self.remote_host = instance.host
             self.remote_port = instance.port
-            self.remote_user = instance.user
-            self.remote_password = instance.password
+            self.remote_user = user
+            self.remote_password = password
         return (
             self.remote_host,
             self.remote_port,

sql/engines/tests.py

@@ -1,6 +1,7 @@
 import json
 from datetime import timedelta, datetime
 from unittest.mock import MagicMock, patch, Mock, ANY
+from pytest_mock import MockerFixture
 
 import sqlparse
 from django.contrib.auth import get_user_model
@@ -17,7 +18,14 @@
 from sql.engines.mongo import MongoEngine
 from sql.engines.clickhouse import ClickHouseEngine
 from sql.engines.odps import ODPSEngine
-from sql.models import DataMaskingColumns, Instance, SqlWorkflow, SqlWorkflowContent
+from sql.models import (
+    DataMaskingColumns,
+    Instance,
+    SqlWorkflow,
+    SqlWorkflowContent,
+    Tunnel,
+)
+
 
 User = get_user_model()
 
@@ -2501,3 +2509,22 @@ def test_get_all_columns_by_tb(self, mock_get_connection):
         self.assertEqual(
             result.column_list, ["COLUMN_NAME", "COLUMN_TYPE", "COLUMN_COMMENT"]
         )
+
+
+def test_ssh(db_instance, mocker: MockerFixture):
+    tunnel = Tunnel.objects.create(tunnel_name="test", host="test", port=22)
+    db_instance.tunnel = tunnel
+    db_instance.save()
+
+    class FakeTunnel:
+        def get_ssh(self):
+            return "remote_host", "remote_password"
+
+    mocker.patch("sql.engines.SSHConnection", return_value=FakeTunnel())
+    from sql.engines import EngineBase
+
+    engine = EngineBase(instance=db_instance)
+    remote_host, remote_password, _, _ = engine.remote_instance_conn(
+        instance=engine.instance
+    )
+    assert (remote_host, remote_password) == ("remote_host", "remote_password")

sql/instance.py

@@ -238,14 +238,18 @@ def schemasync(request):
     tag = int(time.time())
     output_directory = os.path.join(settings.BASE_DIR, "downloads/schemasync/")
     os.makedirs(output_directory, exist_ok=True)
+
+    username, password = instance.get_username_password()
+    target_username, target_password = target_instance.get_username_password()
+
     args = {
         "sync-auto-inc": sync_auto_inc,
         "sync-comments": sync_comments,
         "charset": "utf8mb4",
         "tag": tag,
         "output-directory": output_directory,
-        "source": f"mysql://{instance.user}:{instance.password}@{instance.host}:{instance.port}/{db_name}",
-        "target": f"mysql://{target_instance.user}:{target_instance.password}@{target_instance.host}:{target_instance.port}/{target_db_name}",
+        "source": f"mysql://{username}:{password}@{instance.host}:{instance.port}/{db_name}",
+        "target": f"mysql://{target_username}:{target_password}@{target_instance.host}:{target_instance.port}/{target_db_name}",
     }
     # 参数检查
     args_check_result = schema_sync.check_args(args)

sql/sql_analyze.py

@@ -66,7 +66,8 @@ def analyze(request):
                 )
             soar_test_dsn = SysConfig().get("soar_test_dsn")
             # 获取实例连接信息
-            online_dsn = f"{instance.user}:{instance.password}@{instance.host}:{instance.port}/{db_name}"
+            user, password = instance.get_username_password()
+            online_dsn = f"{user}:{password}@{instance.host}:{instance.port}/{db_name}"
         else:
             online_dsn = ""
             soar_test_dsn = ""