Repository navigation
Expand file tree
/
Copy pathtest_scim.py
More file actions
278 lines (217 loc) · 10.4 KB
/
Copy pathtest_scim.py
File metadata and controls
278 lines (217 loc) · 10.4 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
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
"""SCIM tests for Users, Groups and Service Principals.
Driven through `w.users`, `w.groups` and `w.service_principals`. These pin the
camelCase wire contract above all: SCIM sends `userName`, `displayName` and a
capital-R `Resources`, and getting any of those wrong does not raise — it hands the
SDK an object with every field `None`. So every assertion here reads a field back
through the SDK rather than off the raw JSON.
"""
import pytest
from databricks.sdk import WorkspaceClient
from databricks.sdk.errors import DatabricksError
from databricks.sdk.service.iam import ComplexValue, Patch, PatchOp
_PATCH_ATOMICITY_CASES = [
(
"users",
"user_name",
"userName",
"display_name",
"displayName",
"alice@example.com",
"bob@example.com",
),
("groups", "display_name", "displayName", "external_id", "externalId", "engineering", "product"),
(
"service_principals",
"application_id",
"applicationId",
"display_name",
"displayName",
"11111111-2222-3333-4444-555555555555",
"66666666-7777-8888-9999-000000000000",
),
]
def _create_patch_test_resource(client, unique_attr, unique_value, stable_attr):
return client.create(**{unique_attr: unique_value, stable_attr: "Original metadata"})
# --------------------------------------------------------------------------- users
@pytest.mark.crud
def test_current_user_is_listed(workspace_client: WorkspaceClient):
"""`Me` and `Users` must agree on who is signed in."""
me = workspace_client.current_user.me()
assert me.user_name in {u.user_name for u in workspace_client.users.list()}
@pytest.mark.crud
@pytest.mark.parametrize(
"user_name,display_name",
[("alice@example.com", "Alice"), ("bob@example.com", "Bob")],
)
def test_create_and_get_user(workspace_client: WorkspaceClient, user_name: str, display_name: str):
created = workspace_client.users.create(
user_name=user_name,
display_name=display_name,
emails=[ComplexValue(value=user_name, primary=True)],
)
assert created.id
assert created.user_name == user_name
assert created.display_name == display_name
fetched = workspace_client.users.get(id=created.id)
assert fetched.user_name == user_name
assert fetched.emails[0].value == user_name
assert fetched.active is True
@pytest.mark.crud
def test_list_users_filter(workspace_client: WorkspaceClient):
"""`userName eq "..."` is the filter the CLI and Terraform actually send."""
workspace_client.users.create(user_name="carol@example.com")
workspace_client.users.create(user_name="dave@example.com")
found = list(workspace_client.users.list(filter='userName eq "carol@example.com"'))
assert [u.user_name for u in found] == ["carol@example.com"]
@pytest.mark.crud
def test_update_user_replaces(workspace_client: WorkspaceClient):
created = workspace_client.users.create(user_name="erin@example.com", display_name="Erin")
workspace_client.users.update(id=created.id, user_name="erin@example.com", display_name="Erin Updated")
assert workspace_client.users.get(id=created.id).display_name == "Erin Updated"
@pytest.mark.crud
def test_patch_user_replaces_one_attribute(workspace_client: WorkspaceClient):
created = workspace_client.users.create(user_name="frank@example.com", display_name="Frank")
workspace_client.users.patch(
id=created.id,
operations=[Patch(op=PatchOp.REPLACE, path="displayName", value="Franklin")],
)
fetched = workspace_client.users.get(id=created.id)
assert fetched.display_name == "Franklin"
assert fetched.user_name == "frank@example.com", "unrelated fields survive a patch"
@pytest.mark.crud
def test_delete_user(workspace_client: WorkspaceClient):
created = workspace_client.users.create(user_name="gone@example.com")
workspace_client.users.delete(id=created.id)
with pytest.raises(DatabricksError):
workspace_client.users.get(id=created.id)
@pytest.mark.error
def test_duplicate_user_name_rejected(workspace_client: WorkspaceClient):
workspace_client.users.create(user_name="dupe@example.com")
with pytest.raises(DatabricksError) as exc:
workspace_client.users.create(user_name="dupe@example.com")
assert "already exists" in str(exc.value).lower()
@pytest.mark.error
@pytest.mark.parametrize(
"resource_name,unique_attr,unique_path,stable_attr,stable_path,first_value,target_value",
_PATCH_ATOMICITY_CASES,
)
def test_duplicate_scim_patch_preserves_resource(
workspace_client: WorkspaceClient,
resource_name: str,
unique_attr: str,
unique_path: str,
stable_attr: str,
stable_path: str,
first_value: str,
target_value: str,
):
resource = getattr(workspace_client, resource_name)
resource.create(**{unique_attr: first_value, stable_attr: "Existing resource"})
target = _create_patch_test_resource(resource, unique_attr, target_value, stable_attr)
with pytest.raises(DatabricksError):
resource.patch(
id=target.id,
operations=[Patch(op=PatchOp.REPLACE, path=unique_path, value=first_value)],
)
fetched = resource.get(id=target.id)
assert getattr(fetched, unique_attr) == target_value
assert getattr(fetched, stable_attr) == "Original metadata"
assert len(list(resource.list(filter=f'{unique_path} eq "{first_value}"'))) == 1
@pytest.mark.error
@pytest.mark.parametrize(
"resource_name,unique_attr,unique_path,stable_attr,stable_path,first_value,target_value",
_PATCH_ATOMICITY_CASES,
)
def test_rejected_multi_operation_scim_patch_does_not_commit_prior_operations(
workspace_client: WorkspaceClient,
resource_name: str,
unique_attr: str,
unique_path: str,
stable_attr: str,
stable_path: str,
first_value: str,
target_value: str,
):
resource = getattr(workspace_client, resource_name)
resource.create(**{unique_attr: first_value, stable_attr: "Existing resource"})
target = _create_patch_test_resource(resource, unique_attr, target_value, stable_attr)
with pytest.raises(DatabricksError):
resource.patch(
id=target.id,
operations=[
Patch(op=PatchOp.REPLACE, path=stable_path, value="Must not be saved"),
Patch(op=PatchOp.REPLACE, path=unique_path, value=first_value),
],
)
fetched = resource.get(id=target.id)
assert getattr(fetched, unique_attr) == target_value
assert getattr(fetched, stable_attr) == "Original metadata"
@pytest.mark.error
def test_unsupported_filter_is_rejected(workspace_client: WorkspaceClient):
"""An unsupported filter fails loudly rather than quietly returning everything."""
with pytest.raises(DatabricksError) as exc:
list(workspace_client.users.list(filter="userName pr"))
assert "filter" in str(exc.value).lower()
# -------------------------------------------------------------------------- groups
@pytest.mark.crud
def test_create_group_with_members(workspace_client: WorkspaceClient):
user = workspace_client.users.create(user_name="member@example.com")
group = workspace_client.groups.create(display_name="engineering", members=[ComplexValue(value=user.id)])
assert group.id
assert group.display_name == "engineering"
fetched = workspace_client.groups.get(id=group.id)
assert [m.value for m in fetched.members] == [user.id]
@pytest.mark.crud
def test_patch_group_adds_and_removes_members(workspace_client: WorkspaceClient):
"""Membership is managed by PATCH, including the `members[value eq "id"]` path
form Terraform sends to drop one member."""
first = workspace_client.users.create(user_name="one@example.com")
second = workspace_client.users.create(user_name="two@example.com")
group = workspace_client.groups.create(display_name="team", members=[ComplexValue(value=first.id)])
workspace_client.groups.patch(
id=group.id,
operations=[Patch(op=PatchOp.ADD, path="members", value=[{"value": second.id}])],
)
members = {m.value for m in workspace_client.groups.get(id=group.id).members}
assert members == {first.id, second.id}
workspace_client.groups.patch(
id=group.id,
operations=[Patch(op=PatchOp.REMOVE, path=f'members[value eq "{first.id}"]')],
)
members = {m.value for m in (workspace_client.groups.get(id=group.id).members or [])}
assert members == {second.id}
@pytest.mark.crud
def test_list_and_delete_group(workspace_client: WorkspaceClient):
group = workspace_client.groups.create(display_name="temporary")
assert "temporary" in {g.display_name for g in workspace_client.groups.list()}
workspace_client.groups.delete(id=group.id)
with pytest.raises(DatabricksError):
workspace_client.groups.get(id=group.id)
@pytest.mark.error
def test_duplicate_group_name_rejected(workspace_client: WorkspaceClient):
workspace_client.groups.create(display_name="dupes")
with pytest.raises(DatabricksError):
workspace_client.groups.create(display_name="dupes")
# --------------------------------------------------------------- service principals
@pytest.mark.crud
def test_create_service_principal_mints_application_id(workspace_client: WorkspaceClient):
"""`applicationId` is how everything downstream addresses a service principal,
so one is generated when the client does not supply it."""
created = workspace_client.service_principals.create(display_name="ci-bot")
assert created.id
assert created.application_id, "an application_id must always be present"
assert created.display_name == "ci-bot"
assert workspace_client.service_principals.get(id=created.id).display_name == "ci-bot"
@pytest.mark.crud
def test_service_principal_honours_supplied_application_id(workspace_client: WorkspaceClient):
created = workspace_client.service_principals.create(
display_name="pinned", application_id="11111111-2222-3333-4444-555555555555"
)
assert created.application_id == "11111111-2222-3333-4444-555555555555"
@pytest.mark.crud
def test_list_and_delete_service_principal(workspace_client: WorkspaceClient):
created = workspace_client.service_principals.create(display_name="short-lived")
assert "short-lived" in {sp.display_name for sp in workspace_client.service_principals.list()}
workspace_client.service_principals.delete(id=created.id)
with pytest.raises(DatabricksError):
workspace_client.service_principals.get(id=created.id)