Repository navigation
Expand file tree
/
Copy pathtest_warehouses.py
More file actions
78 lines (56 loc) · 2.64 KB
/
Copy pathtest_warehouses.py
File metadata and controls
78 lines (56 loc) · 2.64 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
78
"""SQL Warehouses API tests."""
import pytest
from databricks.sdk import WorkspaceClient
@pytest.mark.crud
def test_warehouse_create_returns_id(workspace_client: WorkspaceClient):
"""Test: POST /api/2.0/sql/warehouses creates warehouse and returns id."""
warehouse = workspace_client.warehouses.create(name="test_wh_create", cluster_size="Small")
assert warehouse.id is not None
assert len(warehouse.id) > 0
print(f"✓ Warehouse created with id: {warehouse.id}")
@pytest.mark.crud
def test_warehouse_get_returns_full_details(workspace_client: WorkspaceClient):
"""Test: GET /api/2.0/sql/warehouses/{id} returns full warehouse details."""
# Create warehouse
created = workspace_client.warehouses.create(name="test_wh_get", cluster_size="Small")
warehouse_id = created.id
# Get warehouse details
warehouse = workspace_client.warehouses.get(id=warehouse_id)
assert warehouse.id == warehouse_id
assert warehouse.name == "test_wh_get"
assert warehouse.state.value == "RUNNING"
assert warehouse.cluster_size == "Small"
print(f"✓ Warehouse details: {warehouse.name} ({warehouse.state.value})")
@pytest.mark.crud
def test_warehouse_list_returns_all_created(workspace_client: WorkspaceClient):
"""Test: GET /api/2.0/sql/warehouses lists all created warehouses."""
# Create multiple warehouses
names = ["list_wh_1", "list_wh_2", "list_wh_3"]
created_ids = set()
for name in names:
wh = workspace_client.warehouses.create(name=name)
created_ids.add(wh.id)
# List all
warehouses = list(workspace_client.warehouses.list())
listed_ids = {w.id for w in warehouses}
for wh_id in created_ids:
assert wh_id in listed_ids
print(f"✓ Listed {len(warehouses)} warehouses (including {len(created_ids)} created)")
@pytest.mark.crud
def test_warehouse_lifecycle_stop_and_delete(workspace_client: WorkspaceClient):
"""Test: Warehouse state transitions (RUNNING → STOPPED → deleted)."""
# Create
wh = workspace_client.warehouses.create(name="lifecycle_wh")
warehouse_id = wh.id
# Verify running
wh_started = workspace_client.warehouses.get(id=warehouse_id)
assert wh_started.state.value == "RUNNING"
# Stop
workspace_client.warehouses.stop(id=warehouse_id)
wh_stopped = workspace_client.warehouses.get(id=warehouse_id)
assert wh_stopped.state.value == "STOPPED"
# Delete
workspace_client.warehouses.delete(id=warehouse_id)
warehouses_after = list(workspace_client.warehouses.list())
assert not any(w.id == warehouse_id for w in warehouses_after)
print("✓ Warehouse lifecycle: RUNNING → STOPPED → deleted")