conftest.py

import json
import os

import pytest


def _test_config_from_file():
 """Connection details from the JSON file named by ``DATABRICKS_TEST_CONFIG_FILE``,
 or ``{}`` when the variable is unset/empty/unreadable.

 Why this indirection exists: the engineer-bot (databricks-bot-engine) runs
 this e2e suite inside an agent-driven subprocess whose environment has every
 credential-shaped variable — anything matching ``*TOKEN*`` / ``*SECRET*`` /
 ``*PASSWORD*`` etc. — stripped for safety (the engine's ``shared/env_scrub.py``).
 ``DATABRICKS_TOKEN`` is therefore removed before pytest starts, so the token
 can't reach the agent's test run as a plain env var. The bot instead writes the
 connection details (token included) to a file and points at it with
 ``DATABRICKS_TEST_CONFIG_FILE`` — a name the scrub deliberately preserves.

 Normal CI (and local dev) leaves the variable unset, so this returns ``{}`` and
 every fixture below resolves purely from its env var, unchanged.
 """
 path = os.getenv("DATABRICKS_TEST_CONFIG_FILE")
 if not path:
 return {}
 try:
 with open(path) as f:
 return json.load(f)
 except (OSError, ValueError):
 return {}


@pytest.fixture(scope="session")
def _test_config():
 return _test_config_from_file()


@pytest.fixture(scope="session")
def host(_test_config):
 return os.getenv("DATABRICKS_SERVER_HOSTNAME") or _test_config.get("host")


@pytest.fixture(scope="session")
def http_path(_test_config):
 return os.getenv("DATABRICKS_HTTP_PATH") or _test_config.get("http_path")


@pytest.fixture(scope="session")
def access_token(_test_config):
 return os.getenv("DATABRICKS_TOKEN") or _test_config.get("access_token")


@pytest.fixture(scope="session")
def ingestion_user(_test_config):
 return os.getenv("DATABRICKS_USER") or _test_config.get("ingestion_user")


@pytest.fixture(scope="session")
def catalog(_test_config):
 return os.getenv("DATABRICKS_CATALOG") or _test_config.get("catalog")


@pytest.fixture(scope="session")
def schema(_test_config):
 return os.getenv("DATABRICKS_SCHEMA") or _test_config.get("schema") or "default"


@pytest.fixture(scope="session", autouse=True)
def connection_details(host, http_path, access_token, ingestion_user, catalog, schema):
 return {
 "host": host,
 "http_path": http_path,
 "access_token": access_token,
 "ingestion_user": ingestion_user,
 "catalog": catalog,
 "schema": schema,
 }