Merge branch 'master' into enxdev/feat/enhance-database-modal-validation

# Conflicts:
#	superset-frontend/src/features/databases/DatabaseModal/DatabaseConnectionForm/TableCatalog.tsx

superset-frontend/cypress-base/cypress/e2e/database/modal.test.ts

@@ -63,56 +63,57 @@ describe('Add database', () => {
   it('show error alerts on dynamic form for bad host', () => {
     cy.get('.preferred > :nth-child(1)').click();
 
-    cy.get('input[name="host"]').type('badhost', { force: true });
-    cy.get('input[name="port"]').type('5432', { force: true });
-    cy.get('input[name="username"]').type('testusername', { force: true });
-    cy.get('input[name="database"]').type('testdb', { force: true });
-    cy.get('input[name="password"]').type('testpass', { force: true });
-
-    cy.get('body').click(0, 0);
-
+    cy.get('input[name="host"]').type('badhost', { force: true }).blur();
+    cy.wait('@validateParams', { timeout: 30000 });
+    cy.get('input[name="port"]').type('5432', { force: true }).blur();
+    cy.wait('@validateParams', { timeout: 30000 });
+    cy.get('input[name="username"]')
+      .type('testusername', { force: true })
+      .blur();
+    cy.wait('@validateParams', { timeout: 30000 });
+    cy.get('input[name="database"]').type('testdb', { force: true }).blur();
+    cy.wait('@validateParams', { timeout: 30000 });
+    cy.get('input[name="password"]').type('testpass', { force: true }).blur();
     cy.wait('@validateParams', { timeout: 30000 });
 
-    cy.getBySel('btn-submit-connection').should('not.be.disabled');
+    cy.getBySel('btn-submit-connection', { timeout: 60000 }).should(
+      'not.be.disabled',
+    );
     cy.getBySel('btn-submit-connection').click({ force: true });
 
-    cy.wait('@validateParams', { timeout: 30000 }).then(() => {
-      cy.wait('@createDb', { timeout: 60000 }).then(() => {
-        cy.contains(
-          '.ant-form-item-explain-error',
-          "The hostname provided can't be resolved",
-        ).should('exist');
-      });
+    cy.wait('@createDb', { timeout: 60000 }).then(() => {
+      cy.contains(
+        '.ant-form-item-explain-error',
+        "The hostname provided can't be resolved",
+      ).should('exist');
     });
   });
 
   it('show error alerts on dynamic form for bad port', () => {
     cy.get('.preferred > :nth-child(1)').click();
 
-    cy.get('input[name="host"]').type('localhost', { force: true });
-    cy.get('body').click(0, 0);
+    cy.get('input[name="host"]').type('localhost', { force: true }).blur();
+    cy.wait('@validateParams', { timeout: 30000 });
+    cy.get('input[name="port"]').type('5430', { force: true }).blur();
+    cy.wait('@validateParams', { timeout: 30000 });
+    cy.get('input[name="database"]').type('testdb', { force: true }).blur();
+    cy.wait('@validateParams', { timeout: 30000 });
+    cy.get('input[name="username"]')
+      .type('testusername', { force: true })
+      .blur();
+    cy.wait('@validateParams', { timeout: 30000 });
+    cy.get('input[name="password"]').type('testpass', { force: true }).blur();
     cy.wait('@validateParams', { timeout: 30000 });
 
-    cy.get('input[name="port"]').type('5430', { force: true });
-    cy.get('input[name="database"]').type('testdb', { force: true });
-    cy.get('input[name="username"]').type('testusername', { force: true });
-
-    cy.wait('@validateParams', { timeout: 30000 });
-
-    cy.get('input[name="password"]').type('testpass', { force: true });
-    cy.wait('@validateParams');
-
-    cy.getBySel('btn-submit-connection').should('not.be.disabled');
+    cy.getBySel('btn-submit-connection', { timeout: 60000 }).should(
+      'not.be.disabled',
+    );
     cy.getBySel('btn-submit-connection').click({ force: true });
-    cy.wait('@validateParams', { timeout: 30000 }).then(() => {
-      cy.get('body').click(0, 0);
-      cy.getBySel('btn-submit-connection').click({ force: true });
-      cy.wait('@createDb', { timeout: 60000 }).then(() => {
-        cy.contains(
-          '.ant-form-item-explain-error',
-          'The port is closed',
-        ).should('exist');
-      });
+
+    cy.wait('@createDb', { timeout: 60000 }).then(() => {
+      cy.contains('.ant-form-item-explain-error', 'The port is closed').should(
+        'exist',
+      );
     });
   });
 });

superset-frontend/packages/superset-ui-core/src/components/Form/LabeledErrorBoundInput.tsx

@@ -79,7 +79,7 @@ export const LabeledErrorBoundInput = ({
           isValidating ? 'validating' : hasError ? 'error' : 'success'
         }
         help={errorMessage || helpText}
-        hasFeedback={!!hasError}
+        hasFeedback={isValidating || !!hasError}
       >
         {visibilityToggle || props.name === 'password' ? (
           <StyledInputPassword

superset-frontend/packages/superset-ui-core/src/components/Form/types.ts

@@ -31,5 +31,6 @@ export interface LabeledErrorBoundInputProps {
   id?: string;
   classname?: string;
   visibilityToggle?: boolean;
+  isValidating?: boolean;
   [x: string]: any;
 }

superset-frontend/packages/superset-ui-core/src/components/Select/Select.tsx

@@ -519,7 +519,8 @@ const Select = forwardRef(
               handleSelectAll();
             }}
           >
-            {t('Select all')} {`(${formatNumber('SMART_NUMBER', bulkSelectCounts.selectable)})`}
+            {t('Select all')}{' '}
+            {`(${formatNumber('SMART_NUMBER', bulkSelectCounts.selectable)})`}
           </Button>
           <Button
             type="link"
@@ -536,7 +537,8 @@ const Select = forwardRef(
               handleDeselectAll();
             }}
           >
-            {t('Clear')} {`(${formatNumber('SMART_NUMBER', bulkSelectCounts.deselectable)})`}
+            {t('Clear')}{' '}
+            {`(${formatNumber('SMART_NUMBER', bulkSelectCounts.deselectable)})`}
           </Button>
         </StyledBulkActionsContainer>
       ),

superset-frontend/playwright/tests/dashboard/clear-all-filters.spec.ts

@@ -182,10 +182,7 @@ testWithAssets(
     // Now track POST /api/v1/chart/data requests around Clear All
     const postsAfterClearAll: string[] = [];
     const handler = (req: any) => {
-      if (
-        req.url().includes('/api/v1/chart/data') &&
-        req.method() === 'POST'
-      ) {
+      if (req.url().includes('/api/v1/chart/data') && req.method() === 'POST') {
         postsAfterClearAll.push(req.url());
       }
     };

superset-frontend/plugins/legacy-plugin-chart-world-map/src/WorldMap.ts

@@ -164,7 +164,8 @@ function WorldMap(element: HTMLElement, props: WorldMapProps): void {
     processedData = filteredData.map(d => ({
       ...d,
       radius: radiusScale(Math.sqrt(d.m2)),
-      fillColor: d.m1 != null ? colorFn(d.m1) ?? theme.colorBorder : theme.colorBorder,
+      fillColor:
+        d.m1 != null ? (colorFn(d.m1) ?? theme.colorBorder) : theme.colorBorder,
     }));
   }
 

superset-frontend/src/dashboard/containers/DashboardPage.test.tsx

@@ -285,7 +285,9 @@ test('clears undo history after hydrating the dashboard', async () => {
 
   expect(hydrateDashboard).toHaveBeenCalled();
   expect(clearDashboardHistory).toHaveBeenCalled();
-  const hydrateOrder = (hydrateDashboard as jest.Mock).mock.invocationCallOrder[0];
-  const clearOrder = (clearDashboardHistory as jest.Mock).mock.invocationCallOrder[0];
+  const hydrateOrder = (hydrateDashboard as jest.Mock).mock
+    .invocationCallOrder[0];
+  const clearOrder = (clearDashboardHistory as jest.Mock).mock
+    .invocationCallOrder[0];
   expect(clearOrder).toBeGreaterThan(hydrateOrder);
 });

superset-frontend/src/features/databases/DatabaseModal/DatabaseConnectionForm/CommonParameters.tsx

@@ -243,6 +243,7 @@ export const accessTokenField = ({
   validationErrors,
   db,
   isEditMode,
+  isValidating,
   default_value,
   description,
 }: FieldPropTypes) => (
@@ -250,6 +251,7 @@ export const accessTokenField = ({
     id="access_token"
     name="access_token"
     required={required}
+    isValidating={isValidating}
     visibilityToggle={!isEditMode}
     value={db?.parameters?.access_token}
     validationMethods={{ onBlur: getValidation }}

superset-frontend/src/features/databases/DatabaseModal/DatabaseConnectionForm/TableCatalog.tsx

@@ -33,6 +33,7 @@ export const TableCatalog = ({
   getValidation,
   validationErrors,
   db,
+  isValidating,
   isPublic = true,
 }: FieldPropTypes) => {
   const tableCatalog = db?.catalog || [];
@@ -53,6 +54,7 @@ export const TableCatalog = ({
               <ValidatedInput
                 className="catalog-name-input"
                 required={required}
+                isValidating={isValidating}
                 validationMethods={{ onBlur: getValidation }}
                 errorMessage={catalogError[idx]?.name}
                 placeholder={t('Enter a name for this sheet')}
@@ -86,6 +88,7 @@ export const TableCatalog = ({
             <ValidatedInput
               className="catalog-name-url"
               required={required}
+              isValidating={isValidating}
               validationMethods={{ onBlur: getValidation }}
               errorMessage={catalogError[idx]?.url}
               placeholder={t('Paste the shareable Google Sheet URL here')}

superset-frontend/src/features/databases/DatabaseModal/DatabaseConnectionForm/ValidatedInputField.tsx

@@ -49,11 +49,13 @@ export const validatedInputField = ({
   validationErrors,
   db,
   field,
+  isValidating,
 }: FieldPropTypes) => (
   <ValidatedInput
     id={field}
     name={field}
     required={required}
+    isValidating={isValidating}
     value={db?.parameters?.[field as keyof DatabaseParameters]}
     validationMethods={{ onBlur: getValidation }}
     errorMessage={validationErrors?.[field]}

superset-frontend/src/features/databases/DatabaseModal/SSHTunnelForm.tsx

@@ -18,18 +18,21 @@
  */
 import { useState } from 'react';
 import { t } from '@apache-superset/core/translation';
+import { JsonObject } from '@superset-ui/core';
 import { styled } from '@apache-superset/core/theme';
+import { Alert } from '@apache-superset/core/components';
 import {
   Form,
   FormLabel,
   Col,
   Row,
+  LabeledErrorBoundInput,
+  Icons,
   Tooltip,
 } from '@superset-ui/core/components';
 import { Input } from '@superset-ui/core/components/Input';
 import { Radio } from '@superset-ui/core/components/Radio';
-import { Icons } from '@superset-ui/core/components/Icons';
-import { DatabaseObject, FieldPropTypes } from '../types';
+import { DatabaseObject, CustomEventHandlerType } from '../types';
 import { AuthType } from '.';
 
 const StyledDiv = styled.div`
@@ -48,50 +51,73 @@ const StyledFormItem = styled(Form.Item)`
   margin-bottom: 0 !important;
 `;
 
-const StyledInputPassword = styled(Input.Password)`
-  margin: ${({ theme }) => `${theme.sizeUnit}px 0 ${theme.sizeUnit * 2}px`};
-`;
+interface SSHTunnelFormProps {
+  db: DatabaseObject | null;
+  onSSHTunnelParametersChange: CustomEventHandlerType;
+  setSSHTunnelLoginMethod: (method: AuthType) => void;
+  isValidating?: boolean;
+  validationErrors?: JsonObject | null;
+  getValidation: () => void;
+}
 
 const SSHTunnelForm = ({
   db,
   onSSHTunnelParametersChange,
   setSSHTunnelLoginMethod,
-}: {
-  db: DatabaseObject | null;
-  onSSHTunnelParametersChange: FieldPropTypes['changeMethods']['onSSHTunnelParametersChange'];
-  setSSHTunnelLoginMethod: (method: AuthType) => void;
-}) => {
+  isValidating = false,
+  validationErrors,
+  getValidation,
+}: SSHTunnelFormProps) => {
   const [usePassword, setUsePassword] = useState<AuthType>(AuthType.Password);
+  const sshErrors = validationErrors?.ssh_tunnel || {};
+  const sshSectionError = sshErrors?._error;
 
   return (
     <Form>
+      {sshSectionError && (
+        <StyledRow gutter={16}>
+          <Col xs={24}>
+            <Alert
+              type="error"
+              showIcon
+              message={sshSectionError}
+              data-test="ssh-tunnel-section-error"
+            />
+          </Col>
+        </StyledRow>
+      )}
       <StyledRow gutter={16}>
         <Col xs={24} md={12}>
           <StyledDiv>
-            <FormLabel htmlFor="server_address" required>
-              {t('SSH Host')}
-            </FormLabel>
-            <Input
+            <LabeledErrorBoundInput
+              id="server_address"
               name="server_address"
-              type="text"
+              label={t('SSH Host')}
+              required
               placeholder={t('e.g. 127.0.0.1')}
               value={db?.ssh_tunnel?.server_address || ''}
               onChange={onSSHTunnelParametersChange}
+              validationMethods={{ onBlur: getValidation }}
+              errorMessage={sshErrors?.server_address}
+              isValidating={isValidating}
               data-test="ssh-tunnel-server_address-input"
             />
           </StyledDiv>
         </Col>
         <Col xs={24} md={12}>
           <StyledDiv>
-            <FormLabel htmlFor="server_port" required>
-              {t('SSH Port')}
-            </FormLabel>
-            <Input
+            <LabeledErrorBoundInput
+              id="server_port"
               name="server_port"
+              label={t('SSH Port')}
+              required
               placeholder={t('22')}
               type="number"
               value={db?.ssh_tunnel?.server_port}
               onChange={onSSHTunnelParametersChange}
+              validationMethods={{ onBlur: getValidation }}
+              errorMessage={sshErrors?.server_port}
+              isValidating={isValidating}
               data-test="ssh-tunnel-server_port-input"
             />
           </StyledDiv>
@@ -100,15 +126,17 @@ const SSHTunnelForm = ({
       <StyledRow gutter={16}>
         <Col xs={24}>
           <StyledDiv>
-            <FormLabel htmlFor="username" required>
-              {t('Username')}
-            </FormLabel>
-            <Input
+            <LabeledErrorBoundInput
+              id="username"
               name="username"
-              type="text"
+              label={t('Username')}
+              required
               placeholder={t('e.g. Analytics')}
               value={db?.ssh_tunnel?.username || ''}
               onChange={onSSHTunnelParametersChange}
+              validationMethods={{ onBlur: getValidation }}
+              errorMessage={sshErrors?.username}
+              isValidating={isValidating}
               data-test="ssh-tunnel-username-input"
             />
           </StyledDiv>
@@ -148,16 +176,20 @@ const SSHTunnelForm = ({
         <StyledRow gutter={16}>
           <Col xs={24}>
             <StyledDiv>
-              <FormLabel htmlFor="password" required>
-                {t('SSH Password')}
-              </FormLabel>
-              <StyledInputPassword
+              <LabeledErrorBoundInput
+                id="password"
                 name="password"
+                label={t('SSH Password')}
+                required
+                visibilityToggle
                 placeholder={t('e.g. ********')}
                 value={db?.ssh_tunnel?.password || ''}
                 onChange={onSSHTunnelParametersChange}
+                validationMethods={{ onBlur: getValidation }}
+                errorMessage={sshErrors?.password}
+                isValidating={isValidating}
                 data-test="ssh-tunnel-password-input"
-                iconRender={visible =>
+                iconRender={(visible: boolean) =>
                   visible ? (
                     <Tooltip title={t('Hide password.')}>
                       <Icons.EyeInvisibleOutlined />
@@ -182,30 +214,47 @@ const SSHTunnelForm = ({
                 <FormLabel htmlFor="private_key" required>
                   {t('Private Key')}
                 </FormLabel>
-                <Input.TextArea
-                  name="private_key"
-                  placeholder={t('Paste Private Key here')}
-                  value={db?.ssh_tunnel?.private_key || ''}
-                  onChange={onSSHTunnelParametersChange}
-                  data-test="ssh-tunnel-private_key-input"
-                  rows={4}
-                />
+                <StyledFormItem
+                  validateStatus={
+                    isValidating
+                      ? 'validating'
+                      : sshErrors?.private_key
+                        ? 'error'
+                        : 'success'
+                  }
+                  help={sshErrors?.private_key}
+                  hasFeedback={isValidating || !!sshErrors?.private_key}
+                >
+                  <Input.TextArea
+                    name="private_key"
+                    placeholder={t('Paste Private Key here')}
+                    value={db?.ssh_tunnel?.private_key || ''}
+                    onChange={onSSHTunnelParametersChange}
+                    onBlur={getValidation}
+                    data-test="ssh-tunnel-private_key-input"
+                    rows={4}
+                  />
+                </StyledFormItem>
               </StyledDiv>
             </Col>
           </StyledRow>
           <StyledRow gutter={16}>
             <Col xs={24}>
               <StyledDiv>
-                <FormLabel htmlFor="private_key_password" required>
-                  {t('Private Key Password')}
-                </FormLabel>
-                <StyledInputPassword
+                <LabeledErrorBoundInput
+                  id="private_key_password"
                   name="private_key_password"
+                  label={t('Private Key Password')}
+                  required
+                  visibilityToggle
                   placeholder={t('e.g. ********')}
                   value={db?.ssh_tunnel?.private_key_password || ''}
                   onChange={onSSHTunnelParametersChange}
+                  validationMethods={{ onBlur: getValidation }}
+                  errorMessage={sshErrors?.private_key_password}
+                  isValidating={isValidating}
                   data-test="ssh-tunnel-private_key_password-input"
-                  iconRender={visible =>
+                  iconRender={(visible: boolean) =>
                     visible ? (
                       <Tooltip title={t('Hide password.')}>
                         <Icons.EyeInvisibleOutlined />

superset-frontend/src/features/databases/DatabaseModal/index.test.tsx

@@ -1212,26 +1212,40 @@ describe('DatabaseModal', () => {
             'ssh-tunnel-server_address-input',
           );
           expect(SSHTunnelServerAddressInput).toHaveValue('');
-          userEvent.type(SSHTunnelServerAddressInput, 'localhost');
-          expect(SSHTunnelServerAddressInput).toHaveValue('localhost');
+          fireEvent.change(SSHTunnelServerAddressInput, {
+            target: { value: 'localhost' },
+          });
+          await waitFor(() =>
+            expect(SSHTunnelServerAddressInput).toHaveValue('localhost'),
+          );
           const SSHTunnelServerPortInput = screen.getByTestId(
             'ssh-tunnel-server_port-input',
           );
           expect(SSHTunnelServerPortInput).toHaveValue(null);
-          userEvent.type(SSHTunnelServerPortInput, '22');
-          expect(SSHTunnelServerPortInput).toHaveValue(22);
+          fireEvent.change(SSHTunnelServerPortInput, {
+            target: { value: '22' },
+          });
+          await waitFor(() => expect(SSHTunnelServerPortInput).toHaveValue(22));
           const SSHTunnelUsernameInput = screen.getByTestId(
             'ssh-tunnel-username-input',
           );
           expect(SSHTunnelUsernameInput).toHaveValue('');
-          userEvent.type(SSHTunnelUsernameInput, 'test');
-          expect(SSHTunnelUsernameInput).toHaveValue('test');
+          fireEvent.change(SSHTunnelUsernameInput, {
+            target: { value: 'test' },
+          });
+          await waitFor(() =>
+            expect(SSHTunnelUsernameInput).toHaveValue('test'),
+          );
           const SSHTunnelPasswordInput = screen.getByTestId(
             'ssh-tunnel-password-input',
           );
           expect(SSHTunnelPasswordInput).toHaveValue('');
-          userEvent.type(SSHTunnelPasswordInput, 'pass');
-          expect(SSHTunnelPasswordInput).toHaveValue('pass');
+          fireEvent.change(SSHTunnelPasswordInput, {
+            target: { value: 'pass' },
+          });
+          await waitFor(() =>
+            expect(SSHTunnelPasswordInput).toHaveValue('pass'),
+          );
         });
 
         test('properly interacts with SSH Tunnel form textboxes', async () => {
@@ -1250,26 +1264,40 @@ describe('DatabaseModal', () => {
             'ssh-tunnel-server_address-input',
           );
           expect(SSHTunnelServerAddressInput).toHaveValue('');
-          userEvent.type(SSHTunnelServerAddressInput, 'localhost');
-          expect(SSHTunnelServerAddressInput).toHaveValue('localhost');
+          fireEvent.change(SSHTunnelServerAddressInput, {
+            target: { value: 'localhost' },
+          });
+          await waitFor(() =>
+            expect(SSHTunnelServerAddressInput).toHaveValue('localhost'),
+          );
           const SSHTunnelServerPortInput = screen.getByTestId(
             'ssh-tunnel-server_port-input',
           );
           expect(SSHTunnelServerPortInput).toHaveValue(null);
-          userEvent.type(SSHTunnelServerPortInput, '22');
-          expect(SSHTunnelServerPortInput).toHaveValue(22);
+          fireEvent.change(SSHTunnelServerPortInput, {
+            target: { value: '22' },
+          });
+          await waitFor(() => expect(SSHTunnelServerPortInput).toHaveValue(22));
           const SSHTunnelUsernameInput = screen.getByTestId(
             'ssh-tunnel-username-input',
           );
           expect(SSHTunnelUsernameInput).toHaveValue('');
-          userEvent.type(SSHTunnelUsernameInput, 'test');
-          expect(SSHTunnelUsernameInput).toHaveValue('test');
+          fireEvent.change(SSHTunnelUsernameInput, {
+            target: { value: 'test' },
+          });
+          await waitFor(() =>
+            expect(SSHTunnelUsernameInput).toHaveValue('test'),
+          );
           const SSHTunnelPasswordInput = screen.getByTestId(
             'ssh-tunnel-password-input',
           );
           expect(SSHTunnelPasswordInput).toHaveValue('');
-          userEvent.type(SSHTunnelPasswordInput, 'pass');
-          expect(SSHTunnelPasswordInput).toHaveValue('pass');
+          fireEvent.change(SSHTunnelPasswordInput, {
+            target: { value: 'pass' },
+          });
+          await waitFor(() =>
+            expect(SSHTunnelPasswordInput).toHaveValue('pass'),
+          );
         });
 
         test('if the SSH Tunneling toggle is not true, no inputs are displayed', async () => {
@@ -1364,7 +1392,10 @@ describe('DatabaseModal', () => {
           }),
         );
 
-        const textboxes = screen.getAllByRole('textbox');
+        // Wait for step 2 to render
+        expect(await screen.findByText(/step 2 of 3/i)).toBeInTheDocument();
+
+        const textboxes = await screen.findAllByRole('textbox');
         const hostField = textboxes[0];
         const portField = screen.getByRole('spinbutton');
         const databaseNameField = textboxes[1];
@@ -1380,15 +1411,20 @@ describe('DatabaseModal', () => {
 
         expect(connectButton).toBeDisabled();
 
-        userEvent.type(hostField, 'localhost');
-        userEvent.type(portField, '5432');
-        userEvent.type(databaseNameField, 'postgres');
-        userEvent.type(usernameField, 'testdb');
-        userEvent.type(passwordField, 'demoPassword');
+        fireEvent.change(hostField, { target: { value: 'localhost' } });
+        fireEvent.blur(hostField);
+        fireEvent.change(portField, { target: { value: '5432' } });
+        fireEvent.blur(portField);
+        fireEvent.change(databaseNameField, { target: { value: 'postgres' } });
+        fireEvent.blur(databaseNameField);
+        fireEvent.change(usernameField, { target: { value: 'testdb' } });
+        fireEvent.blur(usernameField);
+        fireEvent.change(passwordField, { target: { value: 'demoPassword' } });
+        fireEvent.blur(passwordField);
 
         await waitFor(() => expect(connectButton).toBeEnabled());
 
-        expect(await screen.findByDisplayValue(/5432/i)).toBeInTheDocument();
+        await waitFor(() => expect(portField).toHaveValue(5432));
         expect(hostField).toHaveValue('localhost');
         expect(portField).toHaveValue(5432);
         expect(databaseNameField).toHaveValue('postgres');
@@ -1397,10 +1433,48 @@ describe('DatabaseModal', () => {
 
         expect(connectButton).toBeEnabled();
         userEvent.click(connectButton);
+        // Verify that validation was called during the form interaction
+        // Note: With the optimized validation, redundant calls on the same db state are skipped
         await waitFor(() => {
           expect(
             fetchMock.callHistory.calls(VALIDATE_PARAMS_ENDPOINT).length,
-          ).toEqual(5);
+          ).toBeGreaterThan(0);
+        });
+      });
+
+      test('does not fire redundant validation on blur when db has not changed', async () => {
+        setup();
+
+        userEvent.click(
+          await screen.findByRole('button', {
+            name: /postgresql/i,
+          }),
+        );
+
+        expect(await screen.findByText(/step 2 of 3/i)).toBeInTheDocument();
+
+        const textboxes = await screen.findAllByRole('textbox');
+        const hostField = textboxes[0];
+
+        // Type a value and blur - should trigger validation
+        fireEvent.change(hostField, { target: { value: 'localhost' } });
+        fireEvent.blur(hostField);
+
+        await waitFor(() => {
+          expect(
+            fetchMock.callHistory.calls(VALIDATE_PARAMS_ENDPOINT).length,
+          ).toEqual(1);
+        });
+
+        // Blur again without changing the value - should NOT trigger another validation
+        fireEvent.focus(hostField);
+        fireEvent.blur(hostField);
+
+        // Wait a tick to ensure no additional calls are made
+        await waitFor(() => {
+          expect(
+            fetchMock.callHistory.calls(VALIDATE_PARAMS_ENDPOINT).length,
+          ).toEqual(1);
         });
       });
     });

superset-frontend/src/features/databases/DatabaseModal/index.tsx

@@ -668,6 +668,7 @@ const DatabaseModal: FunctionComponent<DatabaseModalProps> = ({
     hasValidated,
     setHasValidated,
   ] = useDatabaseValidation();
+  const lastValidatedDbSnapshotRef = useRef<string | null>(null);
   const [hasConnectedDb, setHasConnectedDb] = useState<boolean>(false);
   const [showCTAbtns, setShowCTAbtns] = useState(false);
   const [dbName, setDbName] = useState('');
@@ -775,6 +776,7 @@ const DatabaseModal: FunctionComponent<DatabaseModalProps> = ({
   const handleClearValidationErrors = useCallback(() => {
     setValidationErrors(null);
     setHasValidated(false);
+    lastValidatedDbSnapshotRef.current = null;
     clearError();
   }, [setValidationErrors, setHasValidated, clearError]);
 
@@ -851,6 +853,16 @@ const DatabaseModal: FunctionComponent<DatabaseModalProps> = ({
     [onChange],
   );
 
+  const handleTextChange = useCallback(
+    ({ target }: { target: HTMLInputElement }) => {
+      onChange(ActionType.TextChange, {
+        name: target.name,
+        value: target.value,
+      });
+    },
+    [onChange],
+  );
+
   const handleChangeWithValidation = useCallback(
     (
       actionType: ActionType,
@@ -862,6 +874,21 @@ const DatabaseModal: FunctionComponent<DatabaseModalProps> = ({
     [onChange, handleClearValidationErrors],
   );
 
+  const getBlurValidation = useCallback(async () => {
+    const currentDbSnapshot = JSON.stringify(db);
+    if (currentDbSnapshot === lastValidatedDbSnapshotRef.current) {
+      return [];
+    }
+    const result = await getValidation(db);
+    // Only cache after a request that produced a usable response. ``null``
+    // signals an unexpected/network failure, in which case we leave the
+    // snapshot untouched so the next blur retries.
+    if (result !== null) {
+      lastValidatedDbSnapshotRef.current = currentDbSnapshot;
+    }
+    return result;
+  }, [db, getValidation]);
+
   const onClose = () => {
     setDB({ type: ActionType.Reset });
     setHasConnectedDb(false);
@@ -940,7 +967,15 @@ const DatabaseModal: FunctionComponent<DatabaseModalProps> = ({
       }
 
       const errors = await getValidation(dbToUpdate, true);
-      if (!isEmpty(validationErrors) || errors?.length) {
+      // ``getValidation`` returns ``[]`` on success, a field-keyed object
+      // for blocking errors (e.g. the duplicate ``database_name`` check),
+      // and ``null`` for stale or unexpected responses. During save we
+      // cannot proceed without a usable result, so treat ``null`` as
+      // blocking too — only ``[]`` is a clean pass.
+      const hasReturnedErrors =
+        errors === null ||
+        (Array.isArray(errors) ? errors.length > 0 : !isEmpty(errors));
+      if (!isEmpty(validationErrors) || hasReturnedErrors) {
         addDangerToast(
           t('Connection failed, please check your connection settings.'),
         );
@@ -1847,7 +1882,6 @@ const DatabaseModal: FunctionComponent<DatabaseModalProps> = ({
           name: target.name,
           value: target.value,
         });
-        handleClearValidationErrors();
       }}
       setSSHTunnelLoginMethod={(method: AuthType) =>
         setDB({
@@ -1855,6 +1889,9 @@ const DatabaseModal: FunctionComponent<DatabaseModalProps> = ({
           payload: { login_method: method },
         })
       }
+      isValidating={isValidating}
+      validationErrors={validationErrors}
+      getValidation={getBlurValidation}
     />
   );
 
@@ -1930,13 +1967,8 @@ const DatabaseModal: FunctionComponent<DatabaseModalProps> = ({
           });
         }}
         onParametersChange={handleParametersChange}
-        onChange={({ target }: { target: HTMLInputElement }) =>
-          handleChangeWithValidation(ActionType.TextChange, {
-            name: target.name,
-            value: target.value,
-          })
-        }
-        getValidation={() => getValidation(db)}
+        onChange={handleTextChange}
+        getValidation={getBlurValidation}
         validationErrors={validationErrors}
         getPlaceholder={getPlaceholder}
         clearValidationErrors={handleClearValidationErrors}

superset-frontend/src/features/roles/RoleListEditModal.test.tsx

@@ -252,9 +252,7 @@ describe('RoleListEditModal', () => {
     const mockGet = SupersetClient.get as jest.Mock;
     mockGet.mockImplementation(({ endpoint }) => {
       if (
-        endpoint?.includes(
-          `/api/v1/security/roles/${mockRole.id}/permissions/`,
-        )
+        endpoint?.includes(`/api/v1/security/roles/${mockRole.id}/permissions/`)
       ) {
         // Only return permission id=10, not id=20
         return Promise.resolve({
@@ -298,9 +296,7 @@ describe('RoleListEditModal', () => {
     const mockGet = SupersetClient.get as jest.Mock;
     mockGet.mockImplementation(({ endpoint }) => {
       if (
-        endpoint?.includes(
-          `/api/v1/security/roles/${mockRole.id}/permissions/`,
-        )
+        endpoint?.includes(`/api/v1/security/roles/${mockRole.id}/permissions/`)
       ) {
         return Promise.reject(new Error('network error'));
       }
@@ -371,7 +367,9 @@ describe('RoleListEditModal', () => {
     };
 
     mockGet.mockImplementation(({ endpoint }) => {
-      if (endpoint?.includes(`/api/v1/security/roles/${roleA.id}/permissions/`)) {
+      if (
+        endpoint?.includes(`/api/v1/security/roles/${roleA.id}/permissions/`)
+      ) {
         return Promise.resolve({
           json: {
             result: roleA.permission_ids.map(pid => ({
@@ -382,7 +380,9 @@ describe('RoleListEditModal', () => {
           },
         });
       }
-      if (endpoint?.includes(`/api/v1/security/roles/${roleB.id}/permissions/`)) {
+      if (
+        endpoint?.includes(`/api/v1/security/roles/${roleB.id}/permissions/`)
+      ) {
         return Promise.resolve({
           json: {
             result: roleB.permission_ids.map(pid => ({

superset-frontend/src/middleware/loggerMiddleware.ts

@@ -33,7 +33,12 @@ import { ensureAppRoot } from '../utils/pathUtils';
 import type { DashboardInfo, DashboardLayoutState } from '../dashboard/types';
 import type { QueryEditor } from '../SqlLab/types';
 
-type LogEventSource = 'dashboard' | 'embedded_dashboard' | 'explore' | 'sqlLab' | 'slice';
+type LogEventSource =
+  | 'dashboard'
+  | 'embedded_dashboard'
+  | 'explore'
+  | 'sqlLab'
+  | 'slice';
 
 interface LogEventData {
   source?: LogEventSource;

superset-frontend/src/views/CRUD/hooks.ts

@@ -819,16 +819,13 @@ export function useDatabaseValidation() {
   );
   const [isValidating, setIsValidating] = useState(false);
   const [hasValidated, setHasValidated] = useState(false);
+  const latestRequestIdRef = useRef(0);
 
   const getValidation = useCallback(
     async (database: Partial<DatabaseObject> | null, onCreate = false) => {
-      if (database?.parameters?.ssh) {
-        setValidationErrors(null);
-        setIsValidating(false);
-        setHasValidated(true);
-        return Promise.resolve([]);
-      }
-
+      const requestId = latestRequestIdRef.current + 1;
+      latestRequestIdRef.current = requestId;
+      const isLatest = () => latestRequestIdRef.current === requestId;
       setIsValidating(true);
 
       try {
@@ -837,6 +834,9 @@ export function useDatabaseValidation() {
           body: JSON.stringify(transformDB(database)),
           headers: { 'Content-Type': 'application/json' },
         });
+        // Stale responses return ``null`` so callers can tell the result
+        // apart from a real, current outcome and skip caching it.
+        if (!isLatest()) return null;
         setValidationErrors(null);
         setIsValidating(false);
         setHasValidated(true);
@@ -845,12 +845,18 @@ export function useDatabaseValidation() {
         if (typeof error.json === 'function') {
           return error.json().then(({ errors = [] }) => {
             const parsedErrors = errors
-              .filter((err: { error_type: string }) => {
+              .filter((err: { error_type: string; extra?: JsonObject }) => {
                 const allowed = [
                   'CONNECTION_MISSING_PARAMETERS_ERROR',
                   'CONNECTION_ACCESS_DENIED_ERROR',
                   'INVALID_PAYLOAD_SCHEMA_ERROR',
                 ];
+                // SSH-tunnel section errors carry their own ``ssh_tunnel``
+                // marker and need to surface during blur validation too,
+                // otherwise feature-gate failures become invisible
+                // blockers (the save guard would still trip but with no
+                // hint about why).
+                if (err.extra?.ssh_tunnel) return true;
                 return allowed.includes(err.error_type) || onCreate;
               })
               .reduce((acc: JsonObject, err2: any) => {
@@ -866,6 +872,28 @@ export function useDatabaseValidation() {
                   return acc;
                 }
 
+                if (extra?.ssh_tunnel) {
+                  // Field-level errors come in via ``extra.missing``;
+                  // section-level errors (e.g. feature flag disabled) do
+                  // not name a specific field, so preserve the server
+                  // message under a reserved ``_error`` key so the SSH
+                  // form can render it instead of silently dropping it.
+                  const missingFields = extra.missing ?? [];
+                  acc.ssh_tunnel = {
+                    ...acc.ssh_tunnel,
+                    ...Object.fromEntries(
+                      missingFields.map((field: string) => [
+                        field,
+                        'This is a required field',
+                      ]),
+                    ),
+                    ...(missingFields.length === 0 && message
+                      ? { _error: message }
+                      : {}),
+                  };
+                  return acc;
+                }
+
                 if (extra?.invalid) {
                   extra.invalid.forEach((field: string) => {
                     acc[field] = message;
@@ -885,6 +913,7 @@ export function useDatabaseValidation() {
                 return acc;
               }, {});
 
+            if (!isLatest()) return null;
             setValidationErrors(parsedErrors);
             setIsValidating(false);
             setHasValidated(true);
@@ -893,9 +922,11 @@ export function useDatabaseValidation() {
         }
 
         console.error('Unexpected error during validation:', error);
-        setIsValidating(false);
-        setHasValidated(true);
-        return {};
+        if (isLatest()) {
+          setIsValidating(false);
+          setHasValidated(true);
+        }
+        return null;
       }
     },
     [setValidationErrors],

superset/commands/database/validate.py

@@ -19,6 +19,7 @@ from typing import Any, Optional
 
 from flask_babel import gettext as __
 
+from superset import is_feature_enabled
 from superset.commands.base import BaseCommand
 from superset.commands.database.exceptions import (
     DatabaseOfflineError,
@@ -33,6 +34,7 @@ from superset.errors import ErrorLevel, SupersetError, SupersetErrorType
 from superset.extensions import event_logger
 from superset.models.core import Database
 from superset.utils import json
+from superset.utils.ssh_tunnel import unmask_password_info
 
 BYPASS_VALIDATION_ENGINES = {"bigquery", "datastore", "snowflake"}
 
@@ -42,14 +44,23 @@ class ValidateDatabaseParametersCommand(BaseCommand):
         self._properties = properties.copy()
         self._model: Optional[Database] = None
 
-    def run(self) -> None:
+    def run(self) -> None:  # noqa: C901
         self.validate()
 
         engine = self._properties["engine"]
         driver = self._properties.get("driver")
 
         if engine in BYPASS_VALIDATION_ENGINES:
-            # Skip engines that are only validated onCreate
+            # Skip engines that are only validated onCreate, but still surface
+            # database_name uniqueness and SSH tunnel field errors so the
+            # progressive validation flow stays consistent across engines.
+            errors: list[SupersetError] = []
+            if database_name_error := self._get_database_name_error():
+                errors.append(database_name_error)
+            errors.extend(self._get_ssh_tunnel_errors())
+            if errors:
+                event_logger.log_with_context(action="validation_error", engine=engine)
+                raise InvalidParametersError(errors)
             return
 
         engine_spec = get_engine_spec(engine, driver)
@@ -65,8 +76,17 @@ class ValidateDatabaseParametersCommand(BaseCommand):
                 ),
             )
 
-        # perform initial validation
+        # perform initial validation (host, port, database, username)
         errors = engine_spec.validate_parameters(self._properties)  # type: ignore
+
+        # Collect database_name errors along with parameter errors
+        if database_name_error := self._get_database_name_error():
+            errors.append(database_name_error)
+
+        # Collect SSH tunnel errors
+        ssh_tunnel_errors = self._get_ssh_tunnel_errors()
+        errors.extend(ssh_tunnel_errors)
+
         if errors:
             event_logger.log_with_context(action="validation_error", engine=engine)
             raise InvalidParametersError(errors)
@@ -92,11 +112,23 @@ class ValidateDatabaseParametersCommand(BaseCommand):
         )
         if self._model and sqlalchemy_uri == self._model.safe_sqlalchemy_uri():
             sqlalchemy_uri = self._model.sqlalchemy_uri_decrypted
+
+        # Forward the SSH tunnel into the connection test so that
+        # tunnel-only databases are reached through the tunnel rather
+        # than directly, mirroring the existing test_connection flow.
+        ssh_tunnel_properties = self._properties.get("ssh_tunnel")
+        if ssh_tunnel_properties and self._model and self._model.ssh_tunnel:
+            ssh_tunnel_properties = unmask_password_info(
+                ssh_tunnel_properties,
+                self._model.ssh_tunnel,
+            )
+
         database = DatabaseDAO.build_db_for_connection_test(
             server_cert=self._properties.get("server_cert", ""),
             extra=self._properties.get("extra", "{}"),
             impersonate_user=self._properties.get("impersonate_user", False),
             encrypted_extra=json.dumps(encrypted_extra),
+            ssh_tunnel=ssh_tunnel_properties,
         )
         database.set_sqlalchemy_uri(sqlalchemy_uri)
         database.db_engine_spec.mutate_db_for_connection_test(database)
@@ -138,6 +170,116 @@ class ValidateDatabaseParametersCommand(BaseCommand):
                 ),
             )
 
-    def validate(self) -> None:
+    def _load_model(self) -> None:
+        """Load the existing database model if updating."""
         if (database_id := self._properties.get("id")) is not None:
             self._model = DatabaseDAO.find_by_id(database_id)
+
+    def _get_database_name_error(self) -> Optional[SupersetError]:
+        """Check for duplicate database name and return error if found."""
+        database_id = self._properties.get("id")
+
+        if database_name := self._properties.get("database_name"):
+            is_unique = (
+                DatabaseDAO.validate_update_uniqueness(database_id, database_name)
+                if database_id is not None
+                else DatabaseDAO.validate_uniqueness(database_name)
+            )
+            if not is_unique:
+                return SupersetError(
+                    message=__("A database with the same name already exists."),
+                    error_type=SupersetErrorType.INVALID_PAYLOAD_SCHEMA_ERROR,
+                    level=ErrorLevel.ERROR,
+                    extra={"invalid": ["database_name"]},
+                )
+        return None
+
+    def validate(self) -> None:
+        """Load the model in preparation for run()."""
+        self._load_model()
+
+    def _get_ssh_tunnel_errors(self) -> list[SupersetError]:
+        """Validate SSH tunnel fields and return list of errors."""
+        errors: list[SupersetError] = []
+        ssh_tunnel = self._properties.get("ssh_tunnel") or {}
+        parameters = self._properties.get("parameters") or {}
+
+        # ``parameters.ssh`` is the UI toggle. When it's explicitly false the
+        # user has turned SSH off and any stale ``ssh_tunnel`` payload should
+        # be ignored — otherwise toggling off after partially filling the
+        # tunnel form would still trip validation. When the toggle is absent
+        # (older callers / save-time payloads) fall back to inferring from
+        # the presence of an ``ssh_tunnel`` object.
+        if "ssh" in parameters:
+            if not parameters.get("ssh"):
+                return errors
+            ssh_enabled = True
+        else:
+            ssh_enabled = bool(ssh_tunnel)
+        if not ssh_enabled:
+            return errors
+
+        if not is_feature_enabled("SSH_TUNNELING"):
+            errors.append(
+                SupersetError(
+                    message=__("SSH Tunneling is not enabled"),
+                    error_type=SupersetErrorType.GENERIC_DB_ENGINE_ERROR,
+                    level=ErrorLevel.ERROR,
+                    extra={"ssh_tunnel": True},
+                )
+            )
+            return errors
+
+        if not parameters.get("port"):
+            # ``port`` is a database parameter (not a tunnel field), so it
+            # surfaces via the top-level ``missing`` path and lands on the
+            # database port input rather than the SSH tunnel section.
+            errors.append(
+                SupersetError(
+                    message=__(
+                        "A database port is required when connecting via SSH Tunnel."
+                    ),
+                    error_type=SupersetErrorType.CONNECTION_MISSING_PARAMETERS_ERROR,
+                    level=ErrorLevel.WARNING,
+                    extra={"missing": ["port"]},
+                )
+            )
+
+        # Required fields on the tunnel itself
+        required_fields = ["server_address", "server_port", "username"]
+        missing = [f for f in required_fields if not ssh_tunnel.get(f)]
+
+        if missing:
+            errors.append(
+                SupersetError(
+                    message=__(
+                        "One or more parameters are missing: %(missing)s",
+                        missing=", ".join(missing),
+                    ),
+                    error_type=SupersetErrorType.CONNECTION_MISSING_PARAMETERS_ERROR,
+                    level=ErrorLevel.WARNING,
+                    extra={"missing": missing, "ssh_tunnel": True},
+                )
+            )
+
+        # Either password or private_key is required. The active login
+        # method is a client-side toggle, so flag both fields as missing —
+        # the modal renders only one pane at a time and will surface the
+        # error on whichever field the user can see.
+        has_password = bool(ssh_tunnel.get("password"))
+        has_private_key = bool(ssh_tunnel.get("private_key"))
+
+        if not has_password and not has_private_key:
+            errors.append(
+                SupersetError(
+                    message=__("Must provide credentials for the SSH Tunnel"),
+                    error_type=SupersetErrorType.CONNECTION_MISSING_PARAMETERS_ERROR,
+                    level=ErrorLevel.WARNING,
+                    extra={
+                        "missing": ["password", "private_key"],
+                        "ssh_tunnel": True,
+                    },
+                )
+            )
+
+        return errors

superset/databases/schemas.py

@@ -443,6 +443,24 @@ class DatabaseValidateParametersSchema(Schema):
         required=True,
         metadata={"description": configuration_method_description},
     )
+    ssh_tunnel = fields.Nested("DatabaseSSHTunnelValidation", allow_none=True)
+
+
+class DatabaseSSHTunnelValidation(Schema):
+    """SSH Tunnel schema for validation.
+
+    Allows partial data without strict authentication requirements.
+    """
+
+    id = fields.Integer(
+        allow_none=True, metadata={"description": "SSH Tunnel ID (for updates)"}
+    )
+    server_address = fields.String(allow_none=True)
+    server_port = fields.Integer(allow_none=True)
+    username = fields.String(allow_none=True)
+    password = fields.String(required=False, allow_none=True)
+    private_key = fields.String(required=False, allow_none=True)
+    private_key_password = fields.String(required=False, allow_none=True)
 
 
 class DatabaseSSHTunnel(Schema):

tests/unit_tests/commands/databases/validate_test.py

@@ -205,3 +205,487 @@ 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 surfaces a field-level
+    SupersetError tagged with ``ssh_tunnel`` so the modal can map it back
+    to the SSH tunnel section instead of throwing a hard 400 toast.
+    """
+    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(InvalidParametersError) as excinfo:
+        command.run()
+    assert any(
+        err.extra is not None and err.extra.get("ssh_tunnel") is True
+        for err in excinfo.value.errors
+    )
+
+
+def test_validate_ssh_tunnel_missing_db_port(mocker: MockerFixture) -> None:
+    """
+    SSH tunneling requires a database port; the error is surfaced via the
+    top-level ``missing`` extra so it lands on the database port input.
+    """
+    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",
+        "ssh_tunnel": {"server_address": "localhost"},
+        "parameters": {"host": "localhost"},
+    }
+    command = ValidateDatabaseParametersCommand(properties)
+    with pytest.raises(InvalidParametersError) as excinfo:
+        command.run()
+    assert any(
+        err.extra is not None
+        and "port" in (err.extra.get("missing") or [])
+        and not err.extra.get("ssh_tunnel")
+        for err in excinfo.value.errors
+    )
+
+
+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", "private_key"]
+        for err in excinfo.value.errors
+    )
+
+
+def test_get_ssh_tunnel_errors_unencrypted_private_key_is_valid(
+    mocker: MockerFixture,
+) -> None:
+    """
+    An unencrypted private key (no ``private_key_password``) is a valid
+    SSH tunnel credential — validation should not flag it as missing.
+    """
+    mocker.patch(
+        "superset.commands.database.validate.is_feature_enabled",
+        return_value=True,
+    )
+
+    database = mocker.MagicMock()
+    with database.get_sqla_engine() as engine:
+        engine.dialect.do_ping.return_value = True
+    DatabaseDAO = mocker.patch(  # noqa: N806
+        "superset.commands.database.validate.DatabaseDAO"
+    )
+    DatabaseDAO.validate_uniqueness.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",
+        "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)
+    command.run()
+
+
+def test_ssh_tunnel_forwarded_to_connection_test(
+    mocker: MockerFixture,
+) -> None:
+    """
+    The SSH tunnel payload is forwarded into the connection test so
+    tunnel-only databases are reached through the tunnel rather than
+    pinged directly.
+    """
+    mocker.patch(
+        "superset.commands.database.validate.is_feature_enabled",
+        return_value=True,
+    )
+
+    database = mocker.MagicMock()
+    with database.get_sqla_engine() as engine:
+        engine.dialect.do_ping.return_value = True
+    DatabaseDAO = mocker.patch(  # noqa: N806
+        "superset.commands.database.validate.DatabaseDAO"
+    )
+    DatabaseDAO.validate_uniqueness.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="{}"),
+        ),
+    )
+
+    ssh_tunnel = {
+        "server_address": "ssh.example.com",
+        "server_port": 22,
+        "username": "ssh-user",
+        "password": "secret",
+    }
+    properties = {
+        "engine": "postgresql",
+        "parameters": {
+            "host": "localhost",
+            "port": 5432,
+            "username": "u",
+            "database": "d",
+        },
+        "ssh_tunnel": ssh_tunnel,
+    }
+    command = ValidateDatabaseParametersCommand(properties)
+    command.run()
+
+    DatabaseDAO.build_db_for_connection_test.assert_called_once()
+    assert (
+        DatabaseDAO.build_db_for_connection_test.call_args.kwargs["ssh_tunnel"]
+        == ssh_tunnel
+    )
+
+
+def test_get_ssh_tunnel_errors_skipped_when_parameters_ssh_false(
+    mocker: MockerFixture,
+) -> None:
+    """
+    An explicit ``parameters.ssh == False`` is authoritative and skips SSH
+    tunnel validation even when a stale ``ssh_tunnel`` object is still in
+    the payload — otherwise toggling SSH off after partial entry would
+    leave hidden validation errors blocking save.
+    """
+    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",
+            "ssh": False,
+        },
+        # Stale partial tunnel payload from before the user toggled off:
+        "ssh_tunnel": {"server_address": "ssh.example.com"},
+    }
+    command = ValidateDatabaseParametersCommand(properties)
+    command.run()
+
+
+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()
+
+
+def test_bypass_engine_surfaces_ssh_tunnel_errors(mocker: MockerFixture) -> None:
+    """
+    Bypass engines also surface SSH tunnel field errors so the progressive
+    validation flow stays consistent across engine types.
+    """
+    mocker.patch(
+        "superset.commands.database.validate.is_feature_enabled",
+        return_value=True,
+    )
+    DatabaseDAO = mocker.patch(  # noqa: N806
+        "superset.commands.database.validate.DatabaseDAO"
+    )
+    DatabaseDAO.validate_uniqueness.return_value = True
+
+    properties = {
+        "engine": "snowflake",
+        "database_name": "ok",
+        "parameters": {"port": 443},
+        "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
+        for err in excinfo.value.errors
+    )
+
+
+def test_validate_ssh_tunnel_feature_disabled_via_parameters_ssh(
+    mocker: MockerFixture,
+) -> None:
+    """
+    The SSH feature-flag guard fires when the UI marks ``parameters.ssh``
+    even if ``ssh_tunnel`` itself is empty during progressive validation.
+    """
+    mocker.patch(
+        "superset.commands.database.validate.is_feature_enabled",
+        return_value=False,
+    )
+
+    properties = {
+        "engine": "postgresql",
+        "parameters": {"host": "localhost", "port": 5432, "ssh": True},
+        "ssh_tunnel": {},
+    }
+    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
+        for err in excinfo.value.errors
+    )
+
+
+def test_ssh_tunnel_missing_message_is_interpolated(
+    mocker: MockerFixture,
+) -> None:
+    """
+    The translated ``parameters are missing`` message is interpolated with
+    the actual missing fields rather than the raw ``%(missing)s`` token.
+    """
+    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()
+    missing_field_messages = [
+        err.message
+        for err in excinfo.value.errors
+        if err.extra is not None
+        and err.extra.get("missing")
+        and err.extra.get("ssh_tunnel")  # noqa: E501
+    ]
+    assert missing_field_messages
+    assert all("%(missing)s" not in msg for msg in missing_field_messages)
+    assert any("server_port" in msg for msg in missing_field_messages)