##// END OF EJS Templates
STY: Define dirs as a single expression.
Scott Sanderson -
Show More
@@ -1,542 +1,542 b''
1 # coding: utf-8
1 # coding: utf-8
2 """Test the contents webservice API."""
2 """Test the contents webservice API."""
3
3
4 import base64
4 import base64
5 import io
5 import io
6 import json
6 import json
7 import os
7 import os
8 import shutil
8 import shutil
9 from unicodedata import normalize
9 from unicodedata import normalize
10
10
11 pjoin = os.path.join
11 pjoin = os.path.join
12
12
13 import requests
13 import requests
14
14
15 from IPython.html.utils import url_path_join, url_escape, to_os_path
15 from IPython.html.utils import url_path_join, url_escape, to_os_path
16 from IPython.html.tests.launchnotebook import NotebookTestBase, assert_http_error
16 from IPython.html.tests.launchnotebook import NotebookTestBase, assert_http_error
17 from IPython.nbformat import read, write, from_dict
17 from IPython.nbformat import read, write, from_dict
18 from IPython.nbformat.v4 import (
18 from IPython.nbformat.v4 import (
19 new_notebook, new_markdown_cell,
19 new_notebook, new_markdown_cell,
20 )
20 )
21 from IPython.nbformat import v2
21 from IPython.nbformat import v2
22 from IPython.utils import py3compat
22 from IPython.utils import py3compat
23 from IPython.utils.data import uniq_stable
23 from IPython.utils.data import uniq_stable
24
24
25
25
26 def notebooks_only(dir_model):
26 def notebooks_only(dir_model):
27 return [nb for nb in dir_model['content'] if nb['type']=='notebook']
27 return [nb for nb in dir_model['content'] if nb['type']=='notebook']
28
28
29 def dirs_only(dir_model):
29 def dirs_only(dir_model):
30 return [x for x in dir_model['content'] if x['type']=='directory']
30 return [x for x in dir_model['content'] if x['type']=='directory']
31
31
32
32
33 class API(object):
33 class API(object):
34 """Wrapper for contents API calls."""
34 """Wrapper for contents API calls."""
35 def __init__(self, base_url):
35 def __init__(self, base_url):
36 self.base_url = base_url
36 self.base_url = base_url
37
37
38 def _req(self, verb, path, body=None, params=None):
38 def _req(self, verb, path, body=None, params=None):
39 response = requests.request(verb,
39 response = requests.request(verb,
40 url_path_join(self.base_url, 'api/contents', path),
40 url_path_join(self.base_url, 'api/contents', path),
41 data=body, params=params,
41 data=body, params=params,
42 )
42 )
43 response.raise_for_status()
43 response.raise_for_status()
44 return response
44 return response
45
45
46 def list(self, path='/'):
46 def list(self, path='/'):
47 return self._req('GET', path)
47 return self._req('GET', path)
48
48
49 def read(self, path, type=None, format=None):
49 def read(self, path, type=None, format=None):
50 params = {}
50 params = {}
51 if type is not None:
51 if type is not None:
52 params['type'] = type
52 params['type'] = type
53 if format is not None:
53 if format is not None:
54 params['format'] = format
54 params['format'] = format
55 return self._req('GET', path, params=params)
55 return self._req('GET', path, params=params)
56
56
57 def create_untitled(self, path='/', ext='.ipynb'):
57 def create_untitled(self, path='/', ext='.ipynb'):
58 body = None
58 body = None
59 if ext:
59 if ext:
60 body = json.dumps({'ext': ext})
60 body = json.dumps({'ext': ext})
61 return self._req('POST', path, body)
61 return self._req('POST', path, body)
62
62
63 def mkdir_untitled(self, path='/'):
63 def mkdir_untitled(self, path='/'):
64 return self._req('POST', path, json.dumps({'type': 'directory'}))
64 return self._req('POST', path, json.dumps({'type': 'directory'}))
65
65
66 def copy(self, copy_from, path='/'):
66 def copy(self, copy_from, path='/'):
67 body = json.dumps({'copy_from':copy_from})
67 body = json.dumps({'copy_from':copy_from})
68 return self._req('POST', path, body)
68 return self._req('POST', path, body)
69
69
70 def create(self, path='/'):
70 def create(self, path='/'):
71 return self._req('PUT', path)
71 return self._req('PUT', path)
72
72
73 def upload(self, path, body):
73 def upload(self, path, body):
74 return self._req('PUT', path, body)
74 return self._req('PUT', path, body)
75
75
76 def mkdir(self, path='/'):
76 def mkdir(self, path='/'):
77 return self._req('PUT', path, json.dumps({'type': 'directory'}))
77 return self._req('PUT', path, json.dumps({'type': 'directory'}))
78
78
79 def copy_put(self, copy_from, path='/'):
79 def copy_put(self, copy_from, path='/'):
80 body = json.dumps({'copy_from':copy_from})
80 body = json.dumps({'copy_from':copy_from})
81 return self._req('PUT', path, body)
81 return self._req('PUT', path, body)
82
82
83 def save(self, path, body):
83 def save(self, path, body):
84 return self._req('PUT', path, body)
84 return self._req('PUT', path, body)
85
85
86 def delete(self, path='/'):
86 def delete(self, path='/'):
87 return self._req('DELETE', path)
87 return self._req('DELETE', path)
88
88
89 def rename(self, path, new_path):
89 def rename(self, path, new_path):
90 body = json.dumps({'path': new_path})
90 body = json.dumps({'path': new_path})
91 return self._req('PATCH', path, body)
91 return self._req('PATCH', path, body)
92
92
93 def get_checkpoints(self, path):
93 def get_checkpoints(self, path):
94 return self._req('GET', url_path_join(path, 'checkpoints'))
94 return self._req('GET', url_path_join(path, 'checkpoints'))
95
95
96 def new_checkpoint(self, path):
96 def new_checkpoint(self, path):
97 return self._req('POST', url_path_join(path, 'checkpoints'))
97 return self._req('POST', url_path_join(path, 'checkpoints'))
98
98
99 def restore_checkpoint(self, path, checkpoint_id):
99 def restore_checkpoint(self, path, checkpoint_id):
100 return self._req('POST', url_path_join(path, 'checkpoints', checkpoint_id))
100 return self._req('POST', url_path_join(path, 'checkpoints', checkpoint_id))
101
101
102 def delete_checkpoint(self, path, checkpoint_id):
102 def delete_checkpoint(self, path, checkpoint_id):
103 return self._req('DELETE', url_path_join(path, 'checkpoints', checkpoint_id))
103 return self._req('DELETE', url_path_join(path, 'checkpoints', checkpoint_id))
104
104
105 class APITest(NotebookTestBase):
105 class APITest(NotebookTestBase):
106 """Test the kernels web service API"""
106 """Test the kernels web service API"""
107 dirs_nbs = [('', 'inroot'),
107 dirs_nbs = [('', 'inroot'),
108 ('Directory with spaces in', 'inspace'),
108 ('Directory with spaces in', 'inspace'),
109 (u'unicodΓ©', 'innonascii'),
109 (u'unicodΓ©', 'innonascii'),
110 ('foo', 'a'),
110 ('foo', 'a'),
111 ('foo', 'b'),
111 ('foo', 'b'),
112 ('foo', 'name with spaces'),
112 ('foo', 'name with spaces'),
113 ('foo', u'unicodΓ©'),
113 ('foo', u'unicodΓ©'),
114 ('foo/bar', 'baz'),
114 ('foo/bar', 'baz'),
115 ('ordering', 'A'),
115 ('ordering', 'A'),
116 ('ordering', 'b'),
116 ('ordering', 'b'),
117 ('ordering', 'C'),
117 ('ordering', 'C'),
118 (u'Γ₯ b', u'Γ§ d'),
118 (u'Γ₯ b', u'Γ§ d'),
119 ]
119 ]
120 hidden_dirs = ['.hidden', '__pycache__']
120 hidden_dirs = ['.hidden', '__pycache__']
121
121
122 dirs = uniq_stable([py3compat.cast_unicode(d) for (d,n) in dirs_nbs])
122 # Don't include root dir.
123 del dirs[0] # remove ''
123 dirs = uniq_stable([py3compat.cast_unicode(d) for (d,n) in dirs_nbs[1:]])
124 top_level_dirs = {normalize('NFC', d.split('/')[0]) for d in dirs}
124 top_level_dirs = {normalize('NFC', d.split('/')[0]) for d in dirs}
125
125
126 @staticmethod
126 @staticmethod
127 def _blob_for_name(name):
127 def _blob_for_name(name):
128 return name.encode('utf-8') + b'\xFF'
128 return name.encode('utf-8') + b'\xFF'
129
129
130 @staticmethod
130 @staticmethod
131 def _txt_for_name(name):
131 def _txt_for_name(name):
132 return u'%s text file' % name
132 return u'%s text file' % name
133
133
134 def to_os_path(self, api_path):
134 def to_os_path(self, api_path):
135 return to_os_path(api_path, root=self.notebook_dir.name)
135 return to_os_path(api_path, root=self.notebook_dir.name)
136
136
137 def make_dir(self, api_path):
137 def make_dir(self, api_path):
138 """Create a directory at api_path"""
138 """Create a directory at api_path"""
139 os_path = self.to_os_path(api_path)
139 os_path = self.to_os_path(api_path)
140 try:
140 try:
141 os.makedirs(os_path)
141 os.makedirs(os_path)
142 except OSError:
142 except OSError:
143 print("Directory already exists: %r" % os_path)
143 print("Directory already exists: %r" % os_path)
144
144
145 def make_txt(self, api_path, txt):
145 def make_txt(self, api_path, txt):
146 """Make a text file at a given api_path"""
146 """Make a text file at a given api_path"""
147 os_path = self.to_os_path(api_path)
147 os_path = self.to_os_path(api_path)
148 with io.open(os_path, 'w', encoding='utf-8') as f:
148 with io.open(os_path, 'w', encoding='utf-8') as f:
149 f.write(txt)
149 f.write(txt)
150
150
151 def make_blob(self, api_path, blob):
151 def make_blob(self, api_path, blob):
152 """Make a binary file at a given api_path"""
152 """Make a binary file at a given api_path"""
153 os_path = self.to_os_path(api_path)
153 os_path = self.to_os_path(api_path)
154 with io.open(os_path, 'wb') as f:
154 with io.open(os_path, 'wb') as f:
155 f.write(blob)
155 f.write(blob)
156
156
157 def make_nb(self, api_path, nb):
157 def make_nb(self, api_path, nb):
158 """Make a notebook file at a given api_path"""
158 """Make a notebook file at a given api_path"""
159 os_path = self.to_os_path(api_path)
159 os_path = self.to_os_path(api_path)
160
160
161 with io.open(os_path, 'w', encoding='utf-8') as f:
161 with io.open(os_path, 'w', encoding='utf-8') as f:
162 write(nb, f, version=4)
162 write(nb, f, version=4)
163
163
164 def delete_dir(self, api_path):
164 def delete_dir(self, api_path):
165 """Delete a directory at api_path, removing any contents."""
165 """Delete a directory at api_path, removing any contents."""
166 os_path = self.to_os_path(api_path)
166 os_path = self.to_os_path(api_path)
167 shutil.rmtree(os_path, ignore_errors=True)
167 shutil.rmtree(os_path, ignore_errors=True)
168
168
169 def delete_file(self, api_path):
169 def delete_file(self, api_path):
170 """Delete a file at the given path if it exists."""
170 """Delete a file at the given path if it exists."""
171 if self.isfile(api_path):
171 if self.isfile(api_path):
172 os.unlink(self.to_os_path(api_path))
172 os.unlink(self.to_os_path(api_path))
173
173
174 def isfile(self, api_path):
174 def isfile(self, api_path):
175 return os.path.isfile(self.to_os_path(api_path))
175 return os.path.isfile(self.to_os_path(api_path))
176
176
177 def isdir(self, api_path):
177 def isdir(self, api_path):
178 return os.path.isdir(self.to_os_path(api_path))
178 return os.path.isdir(self.to_os_path(api_path))
179
179
180 def setUp(self):
180 def setUp(self):
181
181
182 for d in (self.dirs + self.hidden_dirs):
182 for d in (self.dirs + self.hidden_dirs):
183 self.make_dir(d)
183 self.make_dir(d)
184
184
185 for d, name in self.dirs_nbs:
185 for d, name in self.dirs_nbs:
186 # create a notebook
186 # create a notebook
187 nb = new_notebook()
187 nb = new_notebook()
188 self.make_nb(u'{}/{}.ipynb'.format(d, name), nb)
188 self.make_nb(u'{}/{}.ipynb'.format(d, name), nb)
189
189
190 # create a text file
190 # create a text file
191 txt = self._txt_for_name(name)
191 txt = self._txt_for_name(name)
192 self.make_txt(u'{}/{}.txt'.format(d, name), txt)
192 self.make_txt(u'{}/{}.txt'.format(d, name), txt)
193
193
194 # create a binary file
194 # create a binary file
195 blob = self._blob_for_name(name)
195 blob = self._blob_for_name(name)
196 self.make_blob(u'{}/{}.blob'.format(d, name), blob)
196 self.make_blob(u'{}/{}.blob'.format(d, name), blob)
197
197
198 self.api = API(self.base_url())
198 self.api = API(self.base_url())
199
199
200 def tearDown(self):
200 def tearDown(self):
201 for dname in (list(self.top_level_dirs) + self.hidden_dirs):
201 for dname in (list(self.top_level_dirs) + self.hidden_dirs):
202 self.delete_dir(dname)
202 self.delete_dir(dname)
203 self.delete_file('inroot.ipynb')
203 self.delete_file('inroot.ipynb')
204
204
205 def test_list_notebooks(self):
205 def test_list_notebooks(self):
206 nbs = notebooks_only(self.api.list().json())
206 nbs = notebooks_only(self.api.list().json())
207 self.assertEqual(len(nbs), 1)
207 self.assertEqual(len(nbs), 1)
208 self.assertEqual(nbs[0]['name'], 'inroot.ipynb')
208 self.assertEqual(nbs[0]['name'], 'inroot.ipynb')
209
209
210 nbs = notebooks_only(self.api.list('/Directory with spaces in/').json())
210 nbs = notebooks_only(self.api.list('/Directory with spaces in/').json())
211 self.assertEqual(len(nbs), 1)
211 self.assertEqual(len(nbs), 1)
212 self.assertEqual(nbs[0]['name'], 'inspace.ipynb')
212 self.assertEqual(nbs[0]['name'], 'inspace.ipynb')
213
213
214 nbs = notebooks_only(self.api.list(u'/unicodΓ©/').json())
214 nbs = notebooks_only(self.api.list(u'/unicodΓ©/').json())
215 self.assertEqual(len(nbs), 1)
215 self.assertEqual(len(nbs), 1)
216 self.assertEqual(nbs[0]['name'], 'innonascii.ipynb')
216 self.assertEqual(nbs[0]['name'], 'innonascii.ipynb')
217 self.assertEqual(nbs[0]['path'], u'unicodΓ©/innonascii.ipynb')
217 self.assertEqual(nbs[0]['path'], u'unicodΓ©/innonascii.ipynb')
218
218
219 nbs = notebooks_only(self.api.list('/foo/bar/').json())
219 nbs = notebooks_only(self.api.list('/foo/bar/').json())
220 self.assertEqual(len(nbs), 1)
220 self.assertEqual(len(nbs), 1)
221 self.assertEqual(nbs[0]['name'], 'baz.ipynb')
221 self.assertEqual(nbs[0]['name'], 'baz.ipynb')
222 self.assertEqual(nbs[0]['path'], 'foo/bar/baz.ipynb')
222 self.assertEqual(nbs[0]['path'], 'foo/bar/baz.ipynb')
223
223
224 nbs = notebooks_only(self.api.list('foo').json())
224 nbs = notebooks_only(self.api.list('foo').json())
225 self.assertEqual(len(nbs), 4)
225 self.assertEqual(len(nbs), 4)
226 nbnames = { normalize('NFC', n['name']) for n in nbs }
226 nbnames = { normalize('NFC', n['name']) for n in nbs }
227 expected = [ u'a.ipynb', u'b.ipynb', u'name with spaces.ipynb', u'unicodΓ©.ipynb']
227 expected = [ u'a.ipynb', u'b.ipynb', u'name with spaces.ipynb', u'unicodΓ©.ipynb']
228 expected = { normalize('NFC', name) for name in expected }
228 expected = { normalize('NFC', name) for name in expected }
229 self.assertEqual(nbnames, expected)
229 self.assertEqual(nbnames, expected)
230
230
231 nbs = notebooks_only(self.api.list('ordering').json())
231 nbs = notebooks_only(self.api.list('ordering').json())
232 nbnames = [n['name'] for n in nbs]
232 nbnames = [n['name'] for n in nbs]
233 expected = ['A.ipynb', 'b.ipynb', 'C.ipynb']
233 expected = ['A.ipynb', 'b.ipynb', 'C.ipynb']
234 self.assertEqual(nbnames, expected)
234 self.assertEqual(nbnames, expected)
235
235
236 def test_list_dirs(self):
236 def test_list_dirs(self):
237 dirs = dirs_only(self.api.list().json())
237 dirs = dirs_only(self.api.list().json())
238 dir_names = {normalize('NFC', d['name']) for d in dirs}
238 dir_names = {normalize('NFC', d['name']) for d in dirs}
239 self.assertEqual(dir_names, self.top_level_dirs) # Excluding hidden dirs
239 self.assertEqual(dir_names, self.top_level_dirs) # Excluding hidden dirs
240
240
241 def test_list_nonexistant_dir(self):
241 def test_list_nonexistant_dir(self):
242 with assert_http_error(404):
242 with assert_http_error(404):
243 self.api.list('nonexistant')
243 self.api.list('nonexistant')
244
244
245 def test_get_nb_contents(self):
245 def test_get_nb_contents(self):
246 for d, name in self.dirs_nbs:
246 for d, name in self.dirs_nbs:
247 path = url_path_join(d, name + '.ipynb')
247 path = url_path_join(d, name + '.ipynb')
248 nb = self.api.read(path).json()
248 nb = self.api.read(path).json()
249 self.assertEqual(nb['name'], u'%s.ipynb' % name)
249 self.assertEqual(nb['name'], u'%s.ipynb' % name)
250 self.assertEqual(nb['path'], path)
250 self.assertEqual(nb['path'], path)
251 self.assertEqual(nb['type'], 'notebook')
251 self.assertEqual(nb['type'], 'notebook')
252 self.assertIn('content', nb)
252 self.assertIn('content', nb)
253 self.assertEqual(nb['format'], 'json')
253 self.assertEqual(nb['format'], 'json')
254 self.assertIn('content', nb)
254 self.assertIn('content', nb)
255 self.assertIn('metadata', nb['content'])
255 self.assertIn('metadata', nb['content'])
256 self.assertIsInstance(nb['content']['metadata'], dict)
256 self.assertIsInstance(nb['content']['metadata'], dict)
257
257
258 def test_get_contents_no_such_file(self):
258 def test_get_contents_no_such_file(self):
259 # Name that doesn't exist - should be a 404
259 # Name that doesn't exist - should be a 404
260 with assert_http_error(404):
260 with assert_http_error(404):
261 self.api.read('foo/q.ipynb')
261 self.api.read('foo/q.ipynb')
262
262
263 def test_get_text_file_contents(self):
263 def test_get_text_file_contents(self):
264 for d, name in self.dirs_nbs:
264 for d, name in self.dirs_nbs:
265 path = url_path_join(d, name + '.txt')
265 path = url_path_join(d, name + '.txt')
266 model = self.api.read(path).json()
266 model = self.api.read(path).json()
267 self.assertEqual(model['name'], u'%s.txt' % name)
267 self.assertEqual(model['name'], u'%s.txt' % name)
268 self.assertEqual(model['path'], path)
268 self.assertEqual(model['path'], path)
269 self.assertIn('content', model)
269 self.assertIn('content', model)
270 self.assertEqual(model['format'], 'text')
270 self.assertEqual(model['format'], 'text')
271 self.assertEqual(model['type'], 'file')
271 self.assertEqual(model['type'], 'file')
272 self.assertEqual(model['content'], self._txt_for_name(name))
272 self.assertEqual(model['content'], self._txt_for_name(name))
273
273
274 # Name that doesn't exist - should be a 404
274 # Name that doesn't exist - should be a 404
275 with assert_http_error(404):
275 with assert_http_error(404):
276 self.api.read('foo/q.txt')
276 self.api.read('foo/q.txt')
277
277
278 # Specifying format=text should fail on a non-UTF-8 file
278 # Specifying format=text should fail on a non-UTF-8 file
279 with assert_http_error(400):
279 with assert_http_error(400):
280 self.api.read('foo/bar/baz.blob', type='file', format='text')
280 self.api.read('foo/bar/baz.blob', type='file', format='text')
281
281
282 def test_get_binary_file_contents(self):
282 def test_get_binary_file_contents(self):
283 for d, name in self.dirs_nbs:
283 for d, name in self.dirs_nbs:
284 path = url_path_join(d, name + '.blob')
284 path = url_path_join(d, name + '.blob')
285 model = self.api.read(path).json()
285 model = self.api.read(path).json()
286 self.assertEqual(model['name'], u'%s.blob' % name)
286 self.assertEqual(model['name'], u'%s.blob' % name)
287 self.assertEqual(model['path'], path)
287 self.assertEqual(model['path'], path)
288 self.assertIn('content', model)
288 self.assertIn('content', model)
289 self.assertEqual(model['format'], 'base64')
289 self.assertEqual(model['format'], 'base64')
290 self.assertEqual(model['type'], 'file')
290 self.assertEqual(model['type'], 'file')
291 self.assertEqual(
291 self.assertEqual(
292 base64.decodestring(model['content'].encode('ascii')),
292 base64.decodestring(model['content'].encode('ascii')),
293 self._blob_for_name(name),
293 self._blob_for_name(name),
294 )
294 )
295
295
296 # Name that doesn't exist - should be a 404
296 # Name that doesn't exist - should be a 404
297 with assert_http_error(404):
297 with assert_http_error(404):
298 self.api.read('foo/q.txt')
298 self.api.read('foo/q.txt')
299
299
300 def test_get_bad_type(self):
300 def test_get_bad_type(self):
301 with assert_http_error(400):
301 with assert_http_error(400):
302 self.api.read(u'unicodΓ©', type='file') # this is a directory
302 self.api.read(u'unicodΓ©', type='file') # this is a directory
303
303
304 with assert_http_error(400):
304 with assert_http_error(400):
305 self.api.read(u'unicodΓ©/innonascii.ipynb', type='directory')
305 self.api.read(u'unicodΓ©/innonascii.ipynb', type='directory')
306
306
307 def _check_created(self, resp, path, type='notebook'):
307 def _check_created(self, resp, path, type='notebook'):
308 self.assertEqual(resp.status_code, 201)
308 self.assertEqual(resp.status_code, 201)
309 location_header = py3compat.str_to_unicode(resp.headers['Location'])
309 location_header = py3compat.str_to_unicode(resp.headers['Location'])
310 self.assertEqual(location_header, url_escape(url_path_join(u'/api/contents', path)))
310 self.assertEqual(location_header, url_escape(url_path_join(u'/api/contents', path)))
311 rjson = resp.json()
311 rjson = resp.json()
312 self.assertEqual(rjson['name'], path.rsplit('/', 1)[-1])
312 self.assertEqual(rjson['name'], path.rsplit('/', 1)[-1])
313 self.assertEqual(rjson['path'], path)
313 self.assertEqual(rjson['path'], path)
314 self.assertEqual(rjson['type'], type)
314 self.assertEqual(rjson['type'], type)
315 isright = self.isdir if type == 'directory' else self.isfile
315 isright = self.isdir if type == 'directory' else self.isfile
316 assert isright(path)
316 assert isright(path)
317
317
318 def test_create_untitled(self):
318 def test_create_untitled(self):
319 resp = self.api.create_untitled(path=u'Γ₯ b')
319 resp = self.api.create_untitled(path=u'Γ₯ b')
320 self._check_created(resp, u'Γ₯ b/Untitled.ipynb')
320 self._check_created(resp, u'Γ₯ b/Untitled.ipynb')
321
321
322 # Second time
322 # Second time
323 resp = self.api.create_untitled(path=u'Γ₯ b')
323 resp = self.api.create_untitled(path=u'Γ₯ b')
324 self._check_created(resp, u'Γ₯ b/Untitled1.ipynb')
324 self._check_created(resp, u'Γ₯ b/Untitled1.ipynb')
325
325
326 # And two directories down
326 # And two directories down
327 resp = self.api.create_untitled(path='foo/bar')
327 resp = self.api.create_untitled(path='foo/bar')
328 self._check_created(resp, 'foo/bar/Untitled.ipynb')
328 self._check_created(resp, 'foo/bar/Untitled.ipynb')
329
329
330 def test_create_untitled_txt(self):
330 def test_create_untitled_txt(self):
331 resp = self.api.create_untitled(path='foo/bar', ext='.txt')
331 resp = self.api.create_untitled(path='foo/bar', ext='.txt')
332 self._check_created(resp, 'foo/bar/untitled.txt', type='file')
332 self._check_created(resp, 'foo/bar/untitled.txt', type='file')
333
333
334 resp = self.api.read(path='foo/bar/untitled.txt')
334 resp = self.api.read(path='foo/bar/untitled.txt')
335 model = resp.json()
335 model = resp.json()
336 self.assertEqual(model['type'], 'file')
336 self.assertEqual(model['type'], 'file')
337 self.assertEqual(model['format'], 'text')
337 self.assertEqual(model['format'], 'text')
338 self.assertEqual(model['content'], '')
338 self.assertEqual(model['content'], '')
339
339
340 def test_upload(self):
340 def test_upload(self):
341 nb = new_notebook()
341 nb = new_notebook()
342 nbmodel = {'content': nb, 'type': 'notebook'}
342 nbmodel = {'content': nb, 'type': 'notebook'}
343 path = u'Γ₯ b/Upload tΓ©st.ipynb'
343 path = u'Γ₯ b/Upload tΓ©st.ipynb'
344 resp = self.api.upload(path, body=json.dumps(nbmodel))
344 resp = self.api.upload(path, body=json.dumps(nbmodel))
345 self._check_created(resp, path)
345 self._check_created(resp, path)
346
346
347 def test_mkdir_untitled(self):
347 def test_mkdir_untitled(self):
348 resp = self.api.mkdir_untitled(path=u'Γ₯ b')
348 resp = self.api.mkdir_untitled(path=u'Γ₯ b')
349 self._check_created(resp, u'Γ₯ b/Untitled Folder', type='directory')
349 self._check_created(resp, u'Γ₯ b/Untitled Folder', type='directory')
350
350
351 # Second time
351 # Second time
352 resp = self.api.mkdir_untitled(path=u'Γ₯ b')
352 resp = self.api.mkdir_untitled(path=u'Γ₯ b')
353 self._check_created(resp, u'Γ₯ b/Untitled Folder 1', type='directory')
353 self._check_created(resp, u'Γ₯ b/Untitled Folder 1', type='directory')
354
354
355 # And two directories down
355 # And two directories down
356 resp = self.api.mkdir_untitled(path='foo/bar')
356 resp = self.api.mkdir_untitled(path='foo/bar')
357 self._check_created(resp, 'foo/bar/Untitled Folder', type='directory')
357 self._check_created(resp, 'foo/bar/Untitled Folder', type='directory')
358
358
359 def test_mkdir(self):
359 def test_mkdir(self):
360 path = u'Γ₯ b/New βˆ‚ir'
360 path = u'Γ₯ b/New βˆ‚ir'
361 resp = self.api.mkdir(path)
361 resp = self.api.mkdir(path)
362 self._check_created(resp, path, type='directory')
362 self._check_created(resp, path, type='directory')
363
363
364 def test_mkdir_hidden_400(self):
364 def test_mkdir_hidden_400(self):
365 with assert_http_error(400):
365 with assert_http_error(400):
366 resp = self.api.mkdir(u'Γ₯ b/.hidden')
366 resp = self.api.mkdir(u'Γ₯ b/.hidden')
367
367
368 def test_upload_txt(self):
368 def test_upload_txt(self):
369 body = u'ΓΌnicode tΓ©xt'
369 body = u'ΓΌnicode tΓ©xt'
370 model = {
370 model = {
371 'content' : body,
371 'content' : body,
372 'format' : 'text',
372 'format' : 'text',
373 'type' : 'file',
373 'type' : 'file',
374 }
374 }
375 path = u'Γ₯ b/Upload tΓ©st.txt'
375 path = u'Γ₯ b/Upload tΓ©st.txt'
376 resp = self.api.upload(path, body=json.dumps(model))
376 resp = self.api.upload(path, body=json.dumps(model))
377
377
378 # check roundtrip
378 # check roundtrip
379 resp = self.api.read(path)
379 resp = self.api.read(path)
380 model = resp.json()
380 model = resp.json()
381 self.assertEqual(model['type'], 'file')
381 self.assertEqual(model['type'], 'file')
382 self.assertEqual(model['format'], 'text')
382 self.assertEqual(model['format'], 'text')
383 self.assertEqual(model['content'], body)
383 self.assertEqual(model['content'], body)
384
384
385 def test_upload_b64(self):
385 def test_upload_b64(self):
386 body = b'\xFFblob'
386 body = b'\xFFblob'
387 b64body = base64.encodestring(body).decode('ascii')
387 b64body = base64.encodestring(body).decode('ascii')
388 model = {
388 model = {
389 'content' : b64body,
389 'content' : b64body,
390 'format' : 'base64',
390 'format' : 'base64',
391 'type' : 'file',
391 'type' : 'file',
392 }
392 }
393 path = u'Γ₯ b/Upload tΓ©st.blob'
393 path = u'Γ₯ b/Upload tΓ©st.blob'
394 resp = self.api.upload(path, body=json.dumps(model))
394 resp = self.api.upload(path, body=json.dumps(model))
395
395
396 # check roundtrip
396 # check roundtrip
397 resp = self.api.read(path)
397 resp = self.api.read(path)
398 model = resp.json()
398 model = resp.json()
399 self.assertEqual(model['type'], 'file')
399 self.assertEqual(model['type'], 'file')
400 self.assertEqual(model['path'], path)
400 self.assertEqual(model['path'], path)
401 self.assertEqual(model['format'], 'base64')
401 self.assertEqual(model['format'], 'base64')
402 decoded = base64.decodestring(model['content'].encode('ascii'))
402 decoded = base64.decodestring(model['content'].encode('ascii'))
403 self.assertEqual(decoded, body)
403 self.assertEqual(decoded, body)
404
404
405 def test_upload_v2(self):
405 def test_upload_v2(self):
406 nb = v2.new_notebook()
406 nb = v2.new_notebook()
407 ws = v2.new_worksheet()
407 ws = v2.new_worksheet()
408 nb.worksheets.append(ws)
408 nb.worksheets.append(ws)
409 ws.cells.append(v2.new_code_cell(input='print("hi")'))
409 ws.cells.append(v2.new_code_cell(input='print("hi")'))
410 nbmodel = {'content': nb, 'type': 'notebook'}
410 nbmodel = {'content': nb, 'type': 'notebook'}
411 path = u'Γ₯ b/Upload tΓ©st.ipynb'
411 path = u'Γ₯ b/Upload tΓ©st.ipynb'
412 resp = self.api.upload(path, body=json.dumps(nbmodel))
412 resp = self.api.upload(path, body=json.dumps(nbmodel))
413 self._check_created(resp, path)
413 self._check_created(resp, path)
414 resp = self.api.read(path)
414 resp = self.api.read(path)
415 data = resp.json()
415 data = resp.json()
416 self.assertEqual(data['content']['nbformat'], 4)
416 self.assertEqual(data['content']['nbformat'], 4)
417
417
418 def test_copy(self):
418 def test_copy(self):
419 resp = self.api.copy(u'Γ₯ b/Γ§ d.ipynb', u'Γ₯ b')
419 resp = self.api.copy(u'Γ₯ b/Γ§ d.ipynb', u'Γ₯ b')
420 self._check_created(resp, u'Γ₯ b/Γ§ d-Copy1.ipynb')
420 self._check_created(resp, u'Γ₯ b/Γ§ d-Copy1.ipynb')
421
421
422 resp = self.api.copy(u'Γ₯ b/Γ§ d.ipynb', u'Γ₯ b')
422 resp = self.api.copy(u'Γ₯ b/Γ§ d.ipynb', u'Γ₯ b')
423 self._check_created(resp, u'Γ₯ b/Γ§ d-Copy2.ipynb')
423 self._check_created(resp, u'Γ₯ b/Γ§ d-Copy2.ipynb')
424
424
425 def test_copy_copy(self):
425 def test_copy_copy(self):
426 resp = self.api.copy(u'Γ₯ b/Γ§ d.ipynb', u'Γ₯ b')
426 resp = self.api.copy(u'Γ₯ b/Γ§ d.ipynb', u'Γ₯ b')
427 self._check_created(resp, u'Γ₯ b/Γ§ d-Copy1.ipynb')
427 self._check_created(resp, u'Γ₯ b/Γ§ d-Copy1.ipynb')
428
428
429 resp = self.api.copy(u'Γ₯ b/Γ§ d-Copy1.ipynb', u'Γ₯ b')
429 resp = self.api.copy(u'Γ₯ b/Γ§ d-Copy1.ipynb', u'Γ₯ b')
430 self._check_created(resp, u'Γ₯ b/Γ§ d-Copy2.ipynb')
430 self._check_created(resp, u'Γ₯ b/Γ§ d-Copy2.ipynb')
431
431
432 def test_copy_path(self):
432 def test_copy_path(self):
433 resp = self.api.copy(u'foo/a.ipynb', u'Γ₯ b')
433 resp = self.api.copy(u'foo/a.ipynb', u'Γ₯ b')
434 self._check_created(resp, u'Γ₯ b/a.ipynb')
434 self._check_created(resp, u'Γ₯ b/a.ipynb')
435
435
436 resp = self.api.copy(u'foo/a.ipynb', u'Γ₯ b')
436 resp = self.api.copy(u'foo/a.ipynb', u'Γ₯ b')
437 self._check_created(resp, u'Γ₯ b/a-Copy1.ipynb')
437 self._check_created(resp, u'Γ₯ b/a-Copy1.ipynb')
438
438
439 def test_copy_put_400(self):
439 def test_copy_put_400(self):
440 with assert_http_error(400):
440 with assert_http_error(400):
441 resp = self.api.copy_put(u'Γ₯ b/Γ§ d.ipynb', u'Γ₯ b/cΓΈpy.ipynb')
441 resp = self.api.copy_put(u'Γ₯ b/Γ§ d.ipynb', u'Γ₯ b/cΓΈpy.ipynb')
442
442
443 def test_copy_dir_400(self):
443 def test_copy_dir_400(self):
444 # can't copy directories
444 # can't copy directories
445 with assert_http_error(400):
445 with assert_http_error(400):
446 resp = self.api.copy(u'Γ₯ b', u'foo')
446 resp = self.api.copy(u'Γ₯ b', u'foo')
447
447
448 def test_delete(self):
448 def test_delete(self):
449 for d, name in self.dirs_nbs:
449 for d, name in self.dirs_nbs:
450 print('%r, %r' % (d, name))
450 print('%r, %r' % (d, name))
451 resp = self.api.delete(url_path_join(d, name + '.ipynb'))
451 resp = self.api.delete(url_path_join(d, name + '.ipynb'))
452 self.assertEqual(resp.status_code, 204)
452 self.assertEqual(resp.status_code, 204)
453
453
454 for d in self.dirs + ['/']:
454 for d in self.dirs + ['/']:
455 nbs = notebooks_only(self.api.list(d).json())
455 nbs = notebooks_only(self.api.list(d).json())
456 print('------')
456 print('------')
457 print(d)
457 print(d)
458 print(nbs)
458 print(nbs)
459 self.assertEqual(nbs, [])
459 self.assertEqual(nbs, [])
460
460
461 def test_delete_dirs(self):
461 def test_delete_dirs(self):
462 # depth-first delete everything, so we don't try to delete empty directories
462 # depth-first delete everything, so we don't try to delete empty directories
463 for name in sorted(self.dirs + ['/'], key=len, reverse=True):
463 for name in sorted(self.dirs + ['/'], key=len, reverse=True):
464 listing = self.api.list(name).json()['content']
464 listing = self.api.list(name).json()['content']
465 for model in listing:
465 for model in listing:
466 self.api.delete(model['path'])
466 self.api.delete(model['path'])
467 listing = self.api.list('/').json()['content']
467 listing = self.api.list('/').json()['content']
468 self.assertEqual(listing, [])
468 self.assertEqual(listing, [])
469
469
470 def test_delete_non_empty_dir(self):
470 def test_delete_non_empty_dir(self):
471 """delete non-empty dir raises 400"""
471 """delete non-empty dir raises 400"""
472 with assert_http_error(400):
472 with assert_http_error(400):
473 self.api.delete(u'Γ₯ b')
473 self.api.delete(u'Γ₯ b')
474
474
475 def test_rename(self):
475 def test_rename(self):
476 resp = self.api.rename('foo/a.ipynb', 'foo/z.ipynb')
476 resp = self.api.rename('foo/a.ipynb', 'foo/z.ipynb')
477 self.assertEqual(resp.headers['Location'].split('/')[-1], 'z.ipynb')
477 self.assertEqual(resp.headers['Location'].split('/')[-1], 'z.ipynb')
478 self.assertEqual(resp.json()['name'], 'z.ipynb')
478 self.assertEqual(resp.json()['name'], 'z.ipynb')
479 self.assertEqual(resp.json()['path'], 'foo/z.ipynb')
479 self.assertEqual(resp.json()['path'], 'foo/z.ipynb')
480 assert self.isfile('foo/z.ipynb')
480 assert self.isfile('foo/z.ipynb')
481
481
482 nbs = notebooks_only(self.api.list('foo').json())
482 nbs = notebooks_only(self.api.list('foo').json())
483 nbnames = set(n['name'] for n in nbs)
483 nbnames = set(n['name'] for n in nbs)
484 self.assertIn('z.ipynb', nbnames)
484 self.assertIn('z.ipynb', nbnames)
485 self.assertNotIn('a.ipynb', nbnames)
485 self.assertNotIn('a.ipynb', nbnames)
486
486
487 def test_rename_existing(self):
487 def test_rename_existing(self):
488 with assert_http_error(409):
488 with assert_http_error(409):
489 self.api.rename('foo/a.ipynb', 'foo/b.ipynb')
489 self.api.rename('foo/a.ipynb', 'foo/b.ipynb')
490
490
491 def test_save(self):
491 def test_save(self):
492 resp = self.api.read('foo/a.ipynb')
492 resp = self.api.read('foo/a.ipynb')
493 nbcontent = json.loads(resp.text)['content']
493 nbcontent = json.loads(resp.text)['content']
494 nb = from_dict(nbcontent)
494 nb = from_dict(nbcontent)
495 nb.cells.append(new_markdown_cell(u'Created by test Β³'))
495 nb.cells.append(new_markdown_cell(u'Created by test Β³'))
496
496
497 nbmodel= {'content': nb, 'type': 'notebook'}
497 nbmodel= {'content': nb, 'type': 'notebook'}
498 resp = self.api.save('foo/a.ipynb', body=json.dumps(nbmodel))
498 resp = self.api.save('foo/a.ipynb', body=json.dumps(nbmodel))
499
499
500 nbcontent = self.api.read('foo/a.ipynb').json()['content']
500 nbcontent = self.api.read('foo/a.ipynb').json()['content']
501 newnb = from_dict(nbcontent)
501 newnb = from_dict(nbcontent)
502 self.assertEqual(newnb.cells[0].source,
502 self.assertEqual(newnb.cells[0].source,
503 u'Created by test Β³')
503 u'Created by test Β³')
504
504
505
505
506 def test_checkpoints(self):
506 def test_checkpoints(self):
507 resp = self.api.read('foo/a.ipynb')
507 resp = self.api.read('foo/a.ipynb')
508 r = self.api.new_checkpoint('foo/a.ipynb')
508 r = self.api.new_checkpoint('foo/a.ipynb')
509 self.assertEqual(r.status_code, 201)
509 self.assertEqual(r.status_code, 201)
510 cp1 = r.json()
510 cp1 = r.json()
511 self.assertEqual(set(cp1), {'id', 'last_modified'})
511 self.assertEqual(set(cp1), {'id', 'last_modified'})
512 self.assertEqual(r.headers['Location'].split('/')[-1], cp1['id'])
512 self.assertEqual(r.headers['Location'].split('/')[-1], cp1['id'])
513
513
514 # Modify it
514 # Modify it
515 nbcontent = json.loads(resp.text)['content']
515 nbcontent = json.loads(resp.text)['content']
516 nb = from_dict(nbcontent)
516 nb = from_dict(nbcontent)
517 hcell = new_markdown_cell('Created by test')
517 hcell = new_markdown_cell('Created by test')
518 nb.cells.append(hcell)
518 nb.cells.append(hcell)
519 # Save
519 # Save
520 nbmodel= {'content': nb, 'type': 'notebook'}
520 nbmodel= {'content': nb, 'type': 'notebook'}
521 resp = self.api.save('foo/a.ipynb', body=json.dumps(nbmodel))
521 resp = self.api.save('foo/a.ipynb', body=json.dumps(nbmodel))
522
522
523 # List checkpoints
523 # List checkpoints
524 cps = self.api.get_checkpoints('foo/a.ipynb').json()
524 cps = self.api.get_checkpoints('foo/a.ipynb').json()
525 self.assertEqual(cps, [cp1])
525 self.assertEqual(cps, [cp1])
526
526
527 nbcontent = self.api.read('foo/a.ipynb').json()['content']
527 nbcontent = self.api.read('foo/a.ipynb').json()['content']
528 nb = from_dict(nbcontent)
528 nb = from_dict(nbcontent)
529 self.assertEqual(nb.cells[0].source, 'Created by test')
529 self.assertEqual(nb.cells[0].source, 'Created by test')
530
530
531 # Restore cp1
531 # Restore cp1
532 r = self.api.restore_checkpoint('foo/a.ipynb', cp1['id'])
532 r = self.api.restore_checkpoint('foo/a.ipynb', cp1['id'])
533 self.assertEqual(r.status_code, 204)
533 self.assertEqual(r.status_code, 204)
534 nbcontent = self.api.read('foo/a.ipynb').json()['content']
534 nbcontent = self.api.read('foo/a.ipynb').json()['content']
535 nb = from_dict(nbcontent)
535 nb = from_dict(nbcontent)
536 self.assertEqual(nb.cells, [])
536 self.assertEqual(nb.cells, [])
537
537
538 # Delete cp1
538 # Delete cp1
539 r = self.api.delete_checkpoint('foo/a.ipynb', cp1['id'])
539 r = self.api.delete_checkpoint('foo/a.ipynb', cp1['id'])
540 self.assertEqual(r.status_code, 204)
540 self.assertEqual(r.status_code, 204)
541 cps = self.api.get_checkpoints('foo/a.ipynb').json()
541 cps = self.api.get_checkpoints('foo/a.ipynb').json()
542 self.assertEqual(cps, [])
542 self.assertEqual(cps, [])
General Comments 0
You need to be logged in to leave comments. Login now