Validate Connector Auth

Connector Auth Validate

Goal

Generate and run an authentication verification test for the {{source_name}} connector to confirm that the supplied credentials are valid.

Prerequisites

• API doc must exist at src/databricks/labs/community_connector/sources/{{source_name}}/{{source_name}}_api_doc.md
• Credentials must be supplied via one of:
  • CONNECTOR_TEST_CONFIG_JSON env var — inline JSON.
  • CONNECTOR_TEST_CONFIG_PATH env var — path to a JSON file.
  • A local file at any path the developer chooses, passed to load_config.

If no credentials resolve, stop and report that credentials have not been collected yet and ask user to provide.

Output

• tests/unit/sources/{{source_name}}/auth_test.py — a passing auth verification test

Steps

Step 0: Check if auth_test.py already exists

Check if tests/unit/sources/{{source_name}}/auth_test.py already exists.

• If it exists: skip to Step 3 — just run the existing test to validate credentials. Do not regenerate.
• If it does not exist: proceed to Step 1.

Do NOT search for file locations — all paths are known. Do not run Glob/Search to discover files. Credential field names come from connector_spec.yaml (connection_parameters section), not from any specific on-disk config file.

Step 1: Read the API Doc and Spec for Auth Details

Read src/databricks/labs/community_connector/sources/{{source_name}}/connector_spec.yaml to get the credential field names (connection_parameters), and read the authentication section of src/databricks/labs/community_connector/sources/{{source_name}}/{{source_name}}_api_doc.md to determine:

• The auth method (API key, Bearer token, Basic auth, OAuth, etc.)
• How credentials are passed (headers, query params, etc.)
• The base URL
• A simple read-only endpoint suitable for verifying connectivity

Step 2: Generate the Auth Test

Generate a Python test file at tests/unit/sources/{{source_name}}/auth_test.py.

This script must:

1. Use load_config from tests.unit.sources.test_utils to load credentials
2. Make the simplest possible API call using those credentials
3. Assert the response indicates successful authentication (HTTP 200, no auth errors)
4. Print a clear success or failure message

Template:

"""
Auth verification test for {SourceName} connector.
Run this script to verify your credentials are correctly configured.

Usage:
    # via env var (inline JSON):
    CONNECTOR_TEST_CONFIG_JSON='{"token":"..."}' \\
        python tests/unit/sources/{source_name}/auth_test.py

    # via env var (path to a JSON file at any location):
    CONNECTOR_TEST_CONFIG_PATH=~/secrets/{source_name}.json \\
        python tests/unit/sources/{source_name}/auth_test.py
"""
import sys
import os

sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', '..', '..', '..', '..'))

from tests.unit.sources.test_utils import load_config
import requests


def test_auth():
    """Verify supplied credentials are valid by making a simple API call."""
    config = load_config()  # honors CONNECTOR_TEST_CONFIG_JSON / _PATH env vars

    # Build auth headers/params from config — customize based on auth method
    # Example for Bearer token:
    # headers = {"Authorization": f"Bearer {config['access_token']}"}
    # Example for API key:
    # headers = {"Authorization": f"Token token={config['api_key']}"}
    # Example for Basic auth:
    # auth = (config['email'], config['api_token'])

    response = requests.get(
        "{base_url}{verification_endpoint_path}",
        headers=headers,  # or auth=auth, etc.
        timeout=10
    )

    if response.status_code == 200:
        print(f"Authentication successful! Connected to {SourceName}.")
        print(f"   Response: {response.json()}")
        return True
    elif response.status_code == 401:
        print(f"Authentication failed: Invalid credentials (HTTP 401).")
        print(f"   Check the credentials supplied via CONNECTOR_TEST_CONFIG_JSON / "
              f"CONNECTOR_TEST_CONFIG_PATH.")
        return False
    elif response.status_code == 403:
        print(f"Authorization failed: Insufficient permissions (HTTP 403).")
        print(f"   Ensure your credentials have the required scopes/permissions.")
        return False
    else:
        print(f"Unexpected response: HTTP {response.status_code}")
        print(f"   Body: {response.text}")
        return False


if __name__ == "__main__":
    success = test_auth()
    sys.exit(0 if success else 1)

Customize the template based on:

• The actual auth method from the API doc
• The actual base URL and a simple verification endpoint
• The response structure (extract a useful field like username or account name to print)

Step 3: Run the Auth Test

Run the test using the project virtual environment (Python 3.10+):

source .venv/bin/activate
python tests/unit/sources/{source_name}/auth_test.py

Debug if authentication fails and report the issue clearly.

Quality Standards

• Never hardcode credentials — always load from config files via load_config
• Keep the auth test minimal — one HTTP request, clear output, no complex logic
• Be precise about field names — use the exact field names from the API documentation
• Provide useful error messages — distinguish 401 (wrong credentials) from 403 (wrong permissions)
• Use requests library unless the source has an official Python SDK that simplifies auth significantly

Edge Cases

• OAuth 2.0: When the spec declares a connection.oauth block, the auth test obtains an access token per the spec's flow and calls the API with it as a Bearer credential (Authorization: Bearer <access_token>). For m2m (client-credentials), POST client_id + client_secret (+ scopes) to token_url to get the access token — no user, no refresh token. For u2m / u2m_per_user, the collected credentials typically include client_id, client_secret, and a refresh_token; exchange the refresh token for an access token first.
• Subdomain-based URLs: Build the base URL from the subdomain field in config.
• Multiple auth methods: Use whichever method's credentials are present in the supplied config.