Repository navigation
Expand file tree
/
Copy pathtest_workspace.py
More file actions
171 lines (132 loc) · 6.13 KB
/
Copy pathtest_workspace.py
File metadata and controls
171 lines (132 loc) · 6.13 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
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
"""Workspace endpoints tests — real file-backed notebook storage."""
import base64
import pytest
from databricks.sdk import WorkspaceClient
from databricks.sdk.errors import DatabricksError
from databricks.sdk.service.workspace import ImportFormat, Language
SAMPLE_SCRIPT = b"print('hello from minilake workspace test')\n"
@pytest.mark.crud
def test_workspace_import_and_export(workspace_client: WorkspaceClient):
"""Test: Import a notebook, then export it back and get identical bytes."""
path = "/Shared/test_notebook.py"
workspace_client.workspace.import_(
path=path,
content=base64.b64encode(SAMPLE_SCRIPT).decode("ascii"),
format=ImportFormat.SOURCE,
language=Language.PYTHON,
overwrite=True,
)
exported = workspace_client.workspace.export(path=path)
assert base64.b64decode(exported.content) == SAMPLE_SCRIPT
print("✓ Workspace import/export round-trips real file content")
@pytest.mark.crud
def test_workspace_get_status(workspace_client: WorkspaceClient):
"""Test: get_status returns NOTEBOOK metadata for an imported file."""
path = "/Shared/status_notebook.py"
workspace_client.workspace.import_(
path=path,
content=base64.b64encode(SAMPLE_SCRIPT).decode("ascii"),
format=ImportFormat.SOURCE,
language=Language.PYTHON,
overwrite=True,
)
status = workspace_client.workspace.get_status(path=path)
assert status.path == path
assert status.object_type.value == "NOTEBOOK"
assert status.language.value == "PYTHON"
print("✓ Workspace get_status returns real metadata")
@pytest.mark.crud
def test_workspace_list_directory(workspace_client: WorkspaceClient):
"""Test: list returns imported files under a directory."""
workspace_client.workspace.mkdirs(path="/Shared/listdir")
path = "/Shared/listdir/script.py"
workspace_client.workspace.import_(
path=path,
content=base64.b64encode(SAMPLE_SCRIPT).decode("ascii"),
format=ImportFormat.SOURCE,
language=Language.PYTHON,
overwrite=True,
)
objects = list(workspace_client.workspace.list(path="/Shared/listdir"))
names = [o.path for o in objects]
assert path in names
print("✓ Workspace list shows real imported file")
@pytest.mark.crud
def test_workspace_mkdirs_and_delete(workspace_client: WorkspaceClient):
"""Test: mkdirs creates a real directory, delete removes it."""
workspace_client.workspace.mkdirs(path="/Shared/to_delete")
status = workspace_client.workspace.get_status(path="/Shared/to_delete")
assert status.object_type.value == "DIRECTORY"
workspace_client.workspace.delete(path="/Shared/to_delete", recursive=True)
with pytest.raises(Exception):
workspace_client.workspace.get_status(path="/Shared/to_delete")
print("✓ Workspace mkdirs/delete works for real")
@pytest.mark.workflow
def test_workspace_deleted_directory_can_be_reused_for_notebook(workspace_client: WorkspaceClient):
path = "/Shared/replaced_directory.py"
workspace_client.workspace.mkdirs(path=path)
old_directory = workspace_client.workspace.get_status(path=path)
workspace_client.workspace.delete(path=path, recursive=True)
workspace_client.workspace.import_(
path=path,
content=base64.b64encode(SAMPLE_SCRIPT).decode("ascii"),
format=ImportFormat.SOURCE,
language=Language.PYTHON,
overwrite=True,
)
status = workspace_client.workspace.get_status(path=path)
assert status.object_type.value == "NOTEBOOK"
assert status.object_id != old_directory.object_id
assert status.created_at is not None
assert base64.b64decode(workspace_client.workspace.export(path=path).content) == SAMPLE_SCRIPT
@pytest.mark.error
def test_workspace_notebook_import_does_not_replace_directory(workspace_client: WorkspaceClient):
path = "/Shared/not_a_notebook_directory"
workspace_client.workspace.mkdirs(path=path)
before = workspace_client.workspace.get_status(path=path)
with pytest.raises(DatabricksError) as error:
workspace_client.workspace.import_(
path=path,
content=base64.b64encode(SAMPLE_SCRIPT).decode("ascii"),
format=ImportFormat.SOURCE,
language=Language.PYTHON,
overwrite=False,
)
assert error.value.error_code == "RESOURCE_ALREADY_EXISTS"
after = workspace_client.workspace.get_status(path=path)
assert after.object_type.value == "DIRECTORY"
assert after.object_id == before.object_id
assert list(workspace_client.workspace.list(path=path)) == []
@pytest.mark.error
def test_workspace_export_nonexistent_fails(workspace_client: WorkspaceClient):
"""Test: Exporting a nonexistent path raises an error."""
with pytest.raises(Exception):
workspace_client.workspace.export(path="/Shared/does_not_exist.py")
print("✓ Workspace export of nonexistent path raises error")
@pytest.mark.crud
def test_workspace_import_auto_format_treated_as_source(workspace_client: WorkspaceClient):
"""Test: AUTO format is accepted (aliased to SOURCE), not rejected as unimplemented.
The VS Code extension's "New Notebook"/"New File" actions always send
format=AUTO, never SOURCE explicitly.
"""
path = "/Shared/auto_format_notebook.py"
workspace_client.workspace.import_(
path=path,
content=base64.b64encode(SAMPLE_SCRIPT).decode("ascii"),
format=ImportFormat.AUTO,
overwrite=True,
)
exported = workspace_client.workspace.export(path=path)
assert base64.b64decode(exported.content) == SAMPLE_SCRIPT
print("✓ AUTO import format is accepted like SOURCE")
@pytest.mark.error
def test_workspace_import_non_source_format_not_implemented(workspace_client: WorkspaceClient):
"""Test: Importing with a non-SOURCE format returns 501 (only SOURCE is emulated)."""
with pytest.raises(Exception):
workspace_client.workspace.import_(
path="/Shared/dbc_notebook",
content=base64.b64encode(b"fake dbc content").decode("ascii"),
format=ImportFormat.DBC,
overwrite=True,
)
print("✓ Non-SOURCE import format returns 501")