Repository navigation
Expand file tree
/
Copy pathconftest.py
More file actions
77 lines (56 loc) · 2.46 KB
/
Copy pathconftest.py
File metadata and controls
77 lines (56 loc) · 2.46 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
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,
}