test(database): cover new validate command branches

Adds unit tests for the duplicate-database-name check (create + update
paths, plus the bypass-engine path), the SSH tunnel feature-flag and
db-port guards, and the SSH tunnel field-level error collection
(missing required fields, missing credentials, private key without
password). Brings patch coverage on commands/database/validate.py up
from ~44%.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

tests/unit_tests/commands/databases/validate_test.py

@@ -23,6 +23,10 @@ from superset.commands.database.exceptions import (
     DatabaseTestConnectionFailedError,
     InvalidParametersError,
 )
+from superset.commands.database.ssh_tunnel.exceptions import (
+    SSHTunnelDatabasePortError,
+    SSHTunnelingNotEnabledError,
+)
 from superset.commands.database.validate import ValidateDatabaseParametersCommand
 from superset.errors import ErrorLevel, SupersetError, SupersetErrorType
 
@@ -205,3 +209,264 @@ def test_command_with_oauth2_not_configured(mocker: MockerFixture) -> None:
             extra={"engine_name": "gsheets"},
         )
     ]
+
+
+def test_command_duplicate_database_name(mocker: MockerFixture) -> None:
+    """
+    Validation surfaces a duplicate-name error for a new database with a
+    name already in use.
+    """
+    DatabaseDAO = mocker.patch(  # noqa: N806
+        "superset.commands.database.validate.DatabaseDAO"
+    )
+    DatabaseDAO.validate_uniqueness.return_value = False
+    mocker.patch(
+        "superset.commands.database.validate.get_engine_spec",
+        return_value=mocker.MagicMock(
+            validate_parameters=mocker.MagicMock(return_value=[]),
+        ),
+    )
+
+    properties = {
+        "engine": "postgresql",
+        "database_name": "duplicate",
+        "parameters": {
+            "host": "localhost",
+            "port": 5432,
+            "username": "u",
+            "database": "d",
+        },
+    }
+    command = ValidateDatabaseParametersCommand(properties)
+    with pytest.raises(InvalidParametersError) as excinfo:
+        command.run()
+    assert any(
+        err.error_type == SupersetErrorType.INVALID_PAYLOAD_SCHEMA_ERROR
+        and err.extra is not None
+        and err.extra.get("invalid") == ["database_name"]
+        for err in excinfo.value.errors
+    )
+
+
+def test_command_duplicate_database_name_on_update(mocker: MockerFixture) -> None:
+    """
+    Validation uses ``validate_update_uniqueness`` when an ``id`` is provided.
+    """
+    DatabaseDAO = mocker.patch(  # noqa: N806
+        "superset.commands.database.validate.DatabaseDAO"
+    )
+    DatabaseDAO.find_by_id.return_value = mocker.MagicMock()
+    DatabaseDAO.validate_update_uniqueness.return_value = False
+    mocker.patch(
+        "superset.commands.database.validate.get_engine_spec",
+        return_value=mocker.MagicMock(
+            validate_parameters=mocker.MagicMock(return_value=[]),
+        ),
+    )
+
+    properties = {
+        "id": 1,
+        "engine": "postgresql",
+        "database_name": "existing",
+        "parameters": {
+            "host": "localhost",
+            "port": 5432,
+            "username": "u",
+            "database": "d",
+        },
+    }
+    command = ValidateDatabaseParametersCommand(properties)
+    with pytest.raises(InvalidParametersError):
+        command.run()
+    DatabaseDAO.validate_update_uniqueness.assert_called_once_with(1, "existing")
+
+
+def test_command_duplicate_database_name_bypass_engine(
+    mocker: MockerFixture,
+) -> None:
+    """
+    Bypass engines (e.g. ``bigquery``) still validate database name uniqueness.
+    """
+    DatabaseDAO = mocker.patch(  # noqa: N806
+        "superset.commands.database.validate.DatabaseDAO"
+    )
+    DatabaseDAO.validate_uniqueness.return_value = False
+
+    properties = {
+        "engine": "bigquery",
+        "database_name": "duplicate",
+    }
+    command = ValidateDatabaseParametersCommand(properties)
+    with pytest.raises(InvalidParametersError) as excinfo:
+        command.run()
+    assert excinfo.value.errors[0].error_type == (
+        SupersetErrorType.INVALID_PAYLOAD_SCHEMA_ERROR
+    )
+
+
+def test_validate_ssh_tunnel_feature_disabled(mocker: MockerFixture) -> None:
+    """
+    Enabling SSH tunnel without the feature flag raises an error.
+    """
+    mocker.patch(
+        "superset.commands.database.validate.is_feature_enabled",
+        return_value=False,
+    )
+
+    properties = {
+        "engine": "postgresql",
+        "ssh_tunnel": {"server_address": "localhost"},
+    }
+    command = ValidateDatabaseParametersCommand(properties)
+    with pytest.raises(SSHTunnelingNotEnabledError):
+        command.run()
+
+
+def test_validate_ssh_tunnel_missing_db_port(mocker: MockerFixture) -> None:
+    """
+    SSH tunneling requires a database port.
+    """
+    mocker.patch(
+        "superset.commands.database.validate.is_feature_enabled",
+        return_value=True,
+    )
+
+    properties = {
+        "engine": "postgresql",
+        "ssh_tunnel": {"server_address": "localhost"},
+        "parameters": {"host": "localhost"},
+    }
+    command = ValidateDatabaseParametersCommand(properties)
+    with pytest.raises(SSHTunnelDatabasePortError):
+        command.run()
+
+
+def test_get_ssh_tunnel_errors_missing_required_fields(
+    mocker: MockerFixture,
+) -> None:
+    """
+    SSH tunnel collects missing required fields (server_address, server_port,
+    username) and missing credentials.
+    """
+    mocker.patch(
+        "superset.commands.database.validate.is_feature_enabled",
+        return_value=True,
+    )
+    mocker.patch(
+        "superset.commands.database.validate.get_engine_spec",
+        return_value=mocker.MagicMock(
+            validate_parameters=mocker.MagicMock(return_value=[]),
+        ),
+    )
+
+    properties = {
+        "engine": "postgresql",
+        "parameters": {
+            "host": "localhost",
+            "port": 5432,
+            "username": "u",
+            "database": "d",
+        },
+        "ssh_tunnel": {"server_address": "ssh.example.com"},
+    }
+    command = ValidateDatabaseParametersCommand(properties)
+    with pytest.raises(InvalidParametersError) as excinfo:
+        command.run()
+
+    assert any(
+        err.extra is not None
+        and err.extra.get("ssh_tunnel") is True
+        and err.extra.get("missing") == ["server_port", "username"]
+        for err in excinfo.value.errors
+    )
+    assert any(
+        err.extra is not None
+        and err.extra.get("ssh_tunnel") is True
+        and err.extra.get("missing") == ["password"]
+        for err in excinfo.value.errors
+    )
+
+
+def test_get_ssh_tunnel_errors_private_key_without_password(
+    mocker: MockerFixture,
+) -> None:
+    """
+    Providing a private_key without private_key_password raises a missing
+    parameters error.
+    """
+    mocker.patch(
+        "superset.commands.database.validate.is_feature_enabled",
+        return_value=True,
+    )
+    mocker.patch(
+        "superset.commands.database.validate.get_engine_spec",
+        return_value=mocker.MagicMock(
+            validate_parameters=mocker.MagicMock(return_value=[]),
+        ),
+    )
+
+    properties = {
+        "engine": "postgresql",
+        "parameters": {
+            "host": "localhost",
+            "port": 5432,
+            "username": "u",
+            "database": "d",
+        },
+        "ssh_tunnel": {
+            "server_address": "ssh.example.com",
+            "server_port": 22,
+            "username": "ssh-user",
+            "private_key": "----- KEY -----",
+        },
+    }
+    command = ValidateDatabaseParametersCommand(properties)
+    with pytest.raises(InvalidParametersError) as excinfo:
+        command.run()
+
+    assert any(
+        err.extra is not None
+        and err.extra.get("ssh_tunnel") is True
+        and err.extra.get("missing") == ["private_key_password"]
+        for err in excinfo.value.errors
+    )
+
+
+def test_get_ssh_tunnel_errors_skipped_when_not_enabled(
+    mocker: MockerFixture,
+) -> None:
+    """
+    SSH tunnel validation is a no-op when ssh is not enabled and no tunnel
+    is provided.
+    """
+    DatabaseDAO = mocker.patch(  # noqa: N806
+        "superset.commands.database.validate.DatabaseDAO"
+    )
+    DatabaseDAO.validate_uniqueness.return_value = True
+
+    database = mocker.MagicMock()
+    with database.get_sqla_engine() as engine:
+        engine.dialect.do_ping.return_value = True
+    DatabaseDAO.build_db_for_connection_test.return_value = database
+
+    mocker.patch(
+        "superset.commands.database.validate.get_engine_spec",
+        return_value=mocker.MagicMock(
+            validate_parameters=mocker.MagicMock(return_value=[]),
+            build_sqlalchemy_uri=mocker.MagicMock(return_value="postgresql://"),
+            unmask_encrypted_extra=mocker.MagicMock(return_value="{}"),
+        ),
+    )
+
+    properties = {
+        "engine": "postgresql",
+        "database_name": "ok",
+        "parameters": {
+            "host": "localhost",
+            "port": 5432,
+            "username": "u",
+            "database": "d",
+        },
+    }
+    command = ValidateDatabaseParametersCommand(properties)
+    command.run()