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
|
# Copyright (c) Jupyter Development Team.
# Distributed under the terms of the Modified BSD License.
"""Test the kernels service API."""
import json
import os
import pytest
import tornado.httpclient
from strict_rfc3339 import rfc3339_to_timestamp # type:ignore
from jupyterlab_server.test_utils import (
big_unicode_string,
expected_http_error,
maybe_patch_ioloop,
validate_request,
)
maybe_patch_ioloop()
async def test_delete(jp_fetch, labserverapp):
orig = "f/o/o"
copy = "baz"
r = await jp_fetch("lab", "api", "workspaces", orig)
validate_request(r)
res = r.body.decode()
data = json.loads(res)
data["metadata"]["id"] = copy
r2 = await jp_fetch("lab", "api", "workspaces", copy, method="PUT", body=json.dumps(data))
assert r2.code == 204
r3 = await jp_fetch(
"lab",
"api",
"workspaces",
copy,
method="DELETE",
)
assert r3.code == 204
with pytest.raises(tornado.httpclient.HTTPClientError) as e:
await jp_fetch(
"lab",
"api",
"workspaces",
"does_not_exist",
method="DELETE",
)
assert expected_http_error(e, 404)
with pytest.raises(tornado.httpclient.HTTPClientError) as e:
await jp_fetch(
"lab",
"api",
"workspaces",
"",
method="DELETE",
)
assert expected_http_error(e, 400)
async def test_get_non_existant(jp_fetch, labserverapp):
id = "foo"
r = await jp_fetch("lab", "api", "workspaces", id)
validate_request(r)
data = json.loads(r.body.decode())
r2 = await jp_fetch("lab", "api", "workspaces", id, method="PUT", body=json.dumps(data))
validate_request(r2)
r3 = await jp_fetch("lab", "api", "workspaces", id)
validate_request(r3)
data = json.loads(r3.body.decode())
first_metadata = data["metadata"]
first_created = rfc3339_to_timestamp(first_metadata["created"])
first_modified = rfc3339_to_timestamp(first_metadata["last_modified"])
r4 = await jp_fetch("lab", "api", "workspaces", id, method="PUT", body=json.dumps(data))
validate_request(r4)
r5 = await jp_fetch("lab", "api", "workspaces", id)
validate_request(r5)
data = json.loads(r5.body.decode())
second_metadata = data["metadata"]
second_created = rfc3339_to_timestamp(second_metadata["created"])
second_modified = rfc3339_to_timestamp(second_metadata["last_modified"])
assert first_created <= second_created
assert first_modified < second_modified
@pytest.mark.skipif(os.name == "nt", reason="Temporal failure on windows")
async def test_get(jp_fetch, labserverapp):
id = "foo"
r = await jp_fetch("lab", "api", "workspaces", id)
validate_request(r)
data = json.loads(r.body.decode())
metadata = data["metadata"]
assert metadata["id"] == id
assert rfc3339_to_timestamp(metadata["created"])
assert rfc3339_to_timestamp(metadata["last_modified"])
r2 = await jp_fetch("lab", "api", "workspaces", id)
validate_request(r2)
data = json.loads(r.body.decode())
assert data["metadata"]["id"] == id
async def test_listing(jp_fetch, labserverapp):
# ID fields are from workspaces/*.jupyterlab-workspace
listing = {"foo", "f/o/o/"}
r = await jp_fetch("lab", "api", "workspaces/")
validate_request(r)
res = r.body.decode()
data = json.loads(res)
output = set(data["workspaces"]["ids"])
assert output == listing
async def test_listing_dates(jp_fetch, labserverapp):
r = await jp_fetch("lab", "api", "workspaces")
data = json.loads(r.body.decode())
values = data["workspaces"]["values"]
workspaces = [
[ws["metadata"].get("last_modified"), ws["metadata"].get("created")] for ws in values
]
times = [time for workspace in workspaces for time in workspace]
assert None not in times
[rfc3339_to_timestamp(t) for t in times]
async def test_put(jp_fetch, labserverapp):
id = "foo"
r = await jp_fetch("lab", "api", "workspaces", id)
assert r.code == 200
res = r.body.decode()
data = json.loads(res)
data["metadata"]["big-unicode-string"] = big_unicode_string[::-1]
r2 = await jp_fetch("lab", "api", "workspaces", id, method="PUT", body=json.dumps(data))
assert r2.code == 204
async def test_bad_put(jp_fetch, labserverapp):
orig = "foo"
copy = "bar"
r = await jp_fetch("lab", "api", "workspaces", orig)
assert r.code == 200
res = r.body.decode()
data = json.loads(res)
with pytest.raises(tornado.httpclient.HTTPClientError) as e:
await jp_fetch("lab", "api", "workspaces", copy, method="PUT", body=json.dumps(data))
assert expected_http_error(e, 400)
async def test_blank_put(jp_fetch, labserverapp):
orig = "foo"
r = await jp_fetch("lab", "api", "workspaces", orig)
assert r.code == 200
res = r.body.decode()
data = json.loads(res)
with pytest.raises(tornado.httpclient.HTTPClientError) as e:
await jp_fetch("lab", "api", "workspaces", method="PUT", body=json.dumps(data))
assert expected_http_error(e, 400)
|
