##// END OF EJS Templates
Add some unicode testing for saving notebooks
Thomas Kluyver -
Show More
@@ -1,261 +1,265 b''
1 # coding: utf-8
1 # coding: utf-8
2 """Test the notebooks webservice API."""
2 """Test the notebooks webservice API."""
3
3
4 import io
4 import io
5 import os
5 import os
6 import shutil
6 import shutil
7 from unicodedata import normalize
7 from unicodedata import normalize
8
8
9 from zmq.utils import jsonapi
9 from zmq.utils import jsonapi
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
15 from IPython.html.utils import url_path_join
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.current import (new_notebook, write, read, new_worksheet,
17 from IPython.nbformat.current import (new_notebook, write, read, new_worksheet,
18 new_heading_cell, to_notebook_json)
18 new_heading_cell, to_notebook_json)
19 from IPython.utils.data import uniq_stable
19 from IPython.utils.data import uniq_stable
20
20
21 class NBAPI(object):
21 class NBAPI(object):
22 """Wrapper for notebook API calls."""
22 """Wrapper for notebook API calls."""
23 def __init__(self, base_url):
23 def __init__(self, base_url):
24 self.base_url = base_url
24 self.base_url = base_url
25
25
26 def _req(self, verb, path, body=None):
26 def _req(self, verb, path, body=None):
27 response = requests.request(verb,
27 response = requests.request(verb,
28 url_path_join(self.base_url, 'api/notebooks', path), data=body)
28 url_path_join(self.base_url, 'api/notebooks', path), data=body)
29 response.raise_for_status()
29 response.raise_for_status()
30 return response
30 return response
31
31
32 def list(self, path='/'):
32 def list(self, path='/'):
33 return self._req('GET', path)
33 return self._req('GET', path)
34
34
35 def read(self, name, path='/'):
35 def read(self, name, path='/'):
36 return self._req('GET', url_path_join(path, name))
36 return self._req('GET', url_path_join(path, name))
37
37
38 def create_untitled(self, path='/'):
38 def create_untitled(self, path='/'):
39 return self._req('POST', path)
39 return self._req('POST', path)
40
40
41 def upload(self, name, body, path='/'):
41 def upload(self, name, body, path='/'):
42 return self._req('POST', url_path_join(path, name), body)
42 return self._req('POST', url_path_join(path, name), body)
43
43
44 def copy(self, name, path='/'):
44 def copy(self, name, path='/'):
45 return self._req('POST', url_path_join(path, name, 'copy'))
45 return self._req('POST', url_path_join(path, name, 'copy'))
46
46
47 def save(self, name, body, path='/'):
47 def save(self, name, body, path='/'):
48 return self._req('PUT', url_path_join(path, name), body)
48 return self._req('PUT', url_path_join(path, name), body)
49
49
50 def delete(self, name, path='/'):
50 def delete(self, name, path='/'):
51 return self._req('DELETE', url_path_join(path, name))
51 return self._req('DELETE', url_path_join(path, name))
52
52
53 def rename(self, name, path, new_name):
53 def rename(self, name, path, new_name):
54 body = jsonapi.dumps({'name': new_name})
54 body = jsonapi.dumps({'name': new_name})
55 return self._req('PATCH', url_path_join(path, name), body)
55 return self._req('PATCH', url_path_join(path, name), body)
56
56
57 def get_checkpoints(self, name, path):
57 def get_checkpoints(self, name, path):
58 return self._req('GET', url_path_join(path, name, 'checkpoints'))
58 return self._req('GET', url_path_join(path, name, 'checkpoints'))
59
59
60 def new_checkpoint(self, name, path):
60 def new_checkpoint(self, name, path):
61 return self._req('POST', url_path_join(path, name, 'checkpoints'))
61 return self._req('POST', url_path_join(path, name, 'checkpoints'))
62
62
63 def restore_checkpoint(self, name, path, checkpoint_id):
63 def restore_checkpoint(self, name, path, checkpoint_id):
64 return self._req('POST', url_path_join(path, name, 'checkpoints', checkpoint_id))
64 return self._req('POST', url_path_join(path, name, 'checkpoints', checkpoint_id))
65
65
66 def delete_checkpoint(self, name, path, checkpoint_id):
66 def delete_checkpoint(self, name, path, checkpoint_id):
67 return self._req('DELETE', url_path_join(path, name, 'checkpoints', checkpoint_id))
67 return self._req('DELETE', url_path_join(path, name, 'checkpoints', checkpoint_id))
68
68
69 class APITest(NotebookTestBase):
69 class APITest(NotebookTestBase):
70 """Test the kernels web service API"""
70 """Test the kernels web service API"""
71 dirs_nbs = [('', 'inroot'),
71 dirs_nbs = [('', 'inroot'),
72 ('Directory with spaces in', 'inspace'),
72 ('Directory with spaces in', 'inspace'),
73 (u'unicodΓ©', 'innonascii'),
73 (u'unicodΓ©', 'innonascii'),
74 ('foo', 'a'),
74 ('foo', 'a'),
75 ('foo', 'b'),
75 ('foo', 'b'),
76 ('foo', 'name with spaces'),
76 ('foo', 'name with spaces'),
77 ('foo', u'unicodΓ©'),
77 ('foo', u'unicodΓ©'),
78 ('foo/bar', 'baz'),
78 ('foo/bar', 'baz'),
79 ]
79 ]
80
80
81 dirs = uniq_stable([d for (d,n) in dirs_nbs])
81 dirs = uniq_stable([d for (d,n) in dirs_nbs])
82 del dirs[0] # remove ''
82 del dirs[0] # remove ''
83
83
84 def setUp(self):
84 def setUp(self):
85 nbdir = self.notebook_dir.name
85 nbdir = self.notebook_dir.name
86
86
87 for d in self.dirs:
87 for d in self.dirs:
88 os.mkdir(pjoin(nbdir, d))
88 os.mkdir(pjoin(nbdir, d))
89
89
90 for d, name in self.dirs_nbs:
90 for d, name in self.dirs_nbs:
91 with io.open(pjoin(nbdir, d, '%s.ipynb' % name), 'w') as f:
91 with io.open(pjoin(nbdir, d, '%s.ipynb' % name), 'w') as f:
92 nb = new_notebook(name=name)
92 nb = new_notebook(name=name)
93 write(nb, f, format='ipynb')
93 write(nb, f, format='ipynb')
94
94
95 self.nb_api = NBAPI(self.base_url())
95 self.nb_api = NBAPI(self.base_url())
96
96
97 def tearDown(self):
97 def tearDown(self):
98 nbdir = self.notebook_dir.name
98 nbdir = self.notebook_dir.name
99
99
100 for dname in ['foo', 'Directory with spaces in', u'unicodΓ©']:
100 for dname in ['foo', 'Directory with spaces in', u'unicodΓ©']:
101 shutil.rmtree(pjoin(nbdir, dname), ignore_errors=True)
101 shutil.rmtree(pjoin(nbdir, dname), ignore_errors=True)
102
102
103 if os.path.isfile(pjoin(nbdir, 'inroot.ipynb')):
103 if os.path.isfile(pjoin(nbdir, 'inroot.ipynb')):
104 os.unlink(pjoin(nbdir, 'inroot.ipynb'))
104 os.unlink(pjoin(nbdir, 'inroot.ipynb'))
105
105
106 def test_list_notebooks(self):
106 def test_list_notebooks(self):
107 nbs = self.nb_api.list().json()
107 nbs = self.nb_api.list().json()
108 self.assertEqual(len(nbs), 1)
108 self.assertEqual(len(nbs), 1)
109 self.assertEqual(nbs[0]['name'], 'inroot.ipynb')
109 self.assertEqual(nbs[0]['name'], 'inroot.ipynb')
110
110
111 nbs = self.nb_api.list('/Directory with spaces in/').json()
111 nbs = self.nb_api.list('/Directory with spaces in/').json()
112 self.assertEqual(len(nbs), 1)
112 self.assertEqual(len(nbs), 1)
113 self.assertEqual(nbs[0]['name'], 'inspace.ipynb')
113 self.assertEqual(nbs[0]['name'], 'inspace.ipynb')
114
114
115 nbs = self.nb_api.list(u'/unicodΓ©/').json()
115 nbs = self.nb_api.list(u'/unicodΓ©/').json()
116 self.assertEqual(len(nbs), 1)
116 self.assertEqual(len(nbs), 1)
117 self.assertEqual(nbs[0]['name'], 'innonascii.ipynb')
117 self.assertEqual(nbs[0]['name'], 'innonascii.ipynb')
118
118
119 nbs = self.nb_api.list('/foo/bar/').json()
119 nbs = self.nb_api.list('/foo/bar/').json()
120 self.assertEqual(len(nbs), 1)
120 self.assertEqual(len(nbs), 1)
121 self.assertEqual(nbs[0]['name'], 'baz.ipynb')
121 self.assertEqual(nbs[0]['name'], 'baz.ipynb')
122
122
123 nbs = self.nb_api.list('foo').json()
123 nbs = self.nb_api.list('foo').json()
124 self.assertEqual(len(nbs), 4)
124 self.assertEqual(len(nbs), 4)
125 nbnames = { normalize('NFC', n['name']) for n in nbs }
125 nbnames = { normalize('NFC', n['name']) for n in nbs }
126 expected = [ u'a.ipynb', u'b.ipynb', u'name with spaces.ipynb', u'unicodΓ©.ipynb']
126 expected = [ u'a.ipynb', u'b.ipynb', u'name with spaces.ipynb', u'unicodΓ©.ipynb']
127 expected = { normalize('NFC', name) for name in expected }
127 expected = { normalize('NFC', name) for name in expected }
128 self.assertEqual(nbnames, expected)
128 self.assertEqual(nbnames, expected)
129
129
130 def test_list_nonexistant_dir(self):
130 def test_list_nonexistant_dir(self):
131 with assert_http_error(404):
131 with assert_http_error(404):
132 self.nb_api.list('nonexistant')
132 self.nb_api.list('nonexistant')
133
133
134 def test_get_contents(self):
134 def test_get_contents(self):
135 for d, name in self.dirs_nbs:
135 for d, name in self.dirs_nbs:
136 nb = self.nb_api.read('%s.ipynb' % name, d+'/').json()
136 nb = self.nb_api.read('%s.ipynb' % name, d+'/').json()
137 self.assertEqual(nb['name'], '%s.ipynb' % name)
137 self.assertEqual(nb['name'], '%s.ipynb' % name)
138 self.assertIn('content', nb)
138 self.assertIn('content', nb)
139 self.assertIn('metadata', nb['content'])
139 self.assertIn('metadata', nb['content'])
140 self.assertIsInstance(nb['content']['metadata'], dict)
140 self.assertIsInstance(nb['content']['metadata'], dict)
141
141
142 # Name that doesn't exist - should be a 404
142 # Name that doesn't exist - should be a 404
143 with assert_http_error(404):
143 with assert_http_error(404):
144 self.nb_api.read('q.ipynb', 'foo')
144 self.nb_api.read('q.ipynb', 'foo')
145
145
146 def _check_nb_created(self, resp, name, path):
146 def _check_nb_created(self, resp, name, path):
147 self.assertEqual(resp.status_code, 201)
147 self.assertEqual(resp.status_code, 201)
148 self.assertEqual(resp.headers['Location'].split('/')[-1], name)
148 self.assertEqual(resp.headers['Location'].split('/')[-1], name)
149 self.assertEqual(resp.json()['name'], name)
149 self.assertEqual(resp.json()['name'], name)
150 assert os.path.isfile(pjoin(self.notebook_dir.name, path, name))
150 assert os.path.isfile(pjoin(self.notebook_dir.name, path, name))
151
151
152 def test_create_untitled(self):
152 def test_create_untitled(self):
153 resp = self.nb_api.create_untitled(path='foo')
153 resp = self.nb_api.create_untitled(path='foo')
154 self._check_nb_created(resp, 'Untitled0.ipynb', 'foo')
154 self._check_nb_created(resp, 'Untitled0.ipynb', 'foo')
155
155
156 # Second time
156 # Second time
157 resp = self.nb_api.create_untitled(path='foo')
157 resp = self.nb_api.create_untitled(path='foo')
158 self._check_nb_created(resp, 'Untitled1.ipynb', 'foo')
158 self._check_nb_created(resp, 'Untitled1.ipynb', 'foo')
159
159
160 # And two directories down
160 # And two directories down
161 resp = self.nb_api.create_untitled(path='foo/bar')
161 resp = self.nb_api.create_untitled(path='foo/bar')
162 self._check_nb_created(resp, 'Untitled0.ipynb', pjoin('foo', 'bar'))
162 self._check_nb_created(resp, 'Untitled0.ipynb', pjoin('foo', 'bar'))
163
163
164 def test_upload(self):
164 def test_upload(self):
165 nb = new_notebook(name='Upload test')
165 nb = new_notebook(name='Upload test')
166 nbmodel = {'content': nb}
166 nbmodel = {'content': nb}
167 resp = self.nb_api.upload('Upload test.ipynb', path='foo',
167 resp = self.nb_api.upload('Upload test.ipynb', path='foo',
168 body=jsonapi.dumps(nbmodel))
168 body=jsonapi.dumps(nbmodel))
169 self._check_nb_created(resp, 'Upload test.ipynb', 'foo')
169 self._check_nb_created(resp, 'Upload test.ipynb', 'foo')
170
170
171 def test_copy(self):
171 def test_copy(self):
172 resp = self.nb_api.copy('a.ipynb', path='foo')
172 resp = self.nb_api.copy('a.ipynb', path='foo')
173 self._check_nb_created(resp, 'a-Copy0.ipynb', 'foo')
173 self._check_nb_created(resp, 'a-Copy0.ipynb', 'foo')
174
174
175 def test_delete(self):
175 def test_delete(self):
176 for d, name in self.dirs_nbs:
176 for d, name in self.dirs_nbs:
177 resp = self.nb_api.delete('%s.ipynb' % name, d)
177 resp = self.nb_api.delete('%s.ipynb' % name, d)
178 self.assertEqual(resp.status_code, 204)
178 self.assertEqual(resp.status_code, 204)
179
179
180 for d in self.dirs + ['/']:
180 for d in self.dirs + ['/']:
181 nbs = self.nb_api.list(d).json()
181 nbs = self.nb_api.list(d).json()
182 self.assertEqual(len(nbs), 0)
182 self.assertEqual(len(nbs), 0)
183
183
184 def test_rename(self):
184 def test_rename(self):
185 resp = self.nb_api.rename('a.ipynb', 'foo', 'z.ipynb')
185 resp = self.nb_api.rename('a.ipynb', 'foo', 'z.ipynb')
186 self.assertEqual(resp.headers['Location'].split('/')[-1], 'z.ipynb')
186 self.assertEqual(resp.headers['Location'].split('/')[-1], 'z.ipynb')
187 self.assertEqual(resp.json()['name'], 'z.ipynb')
187 self.assertEqual(resp.json()['name'], 'z.ipynb')
188 assert os.path.isfile(pjoin(self.notebook_dir.name, 'foo', 'z.ipynb'))
188 assert os.path.isfile(pjoin(self.notebook_dir.name, 'foo', 'z.ipynb'))
189
189
190 nbs = self.nb_api.list('foo').json()
190 nbs = self.nb_api.list('foo').json()
191 nbnames = set(n['name'] for n in nbs)
191 nbnames = set(n['name'] for n in nbs)
192 self.assertIn('z.ipynb', nbnames)
192 self.assertIn('z.ipynb', nbnames)
193 self.assertNotIn('a.ipynb', nbnames)
193 self.assertNotIn('a.ipynb', nbnames)
194
194
195 def test_save(self):
195 def test_save(self):
196 resp = self.nb_api.read('a.ipynb', 'foo')
196 resp = self.nb_api.read('a.ipynb', 'foo')
197 nbcontent = jsonapi.loads(resp.text)['content']
197 nbcontent = jsonapi.loads(resp.text)['content']
198 nb = to_notebook_json(nbcontent)
198 nb = to_notebook_json(nbcontent)
199 ws = new_worksheet()
199 ws = new_worksheet()
200 nb.worksheets = [ws]
200 nb.worksheets = [ws]
201 ws.cells.append(new_heading_cell('Created by test'))
201 ws.cells.append(new_heading_cell(u'Created by test Β³'))
202
202
203 nbmodel= {'name': 'a.ipynb', 'path':'foo', 'content': nb}
203 nbmodel= {'name': 'a.ipynb', 'path':'foo', 'content': nb}
204 resp = self.nb_api.save('a.ipynb', path='foo', body=jsonapi.dumps(nbmodel))
204 resp = self.nb_api.save('a.ipynb', path='foo', body=jsonapi.dumps(nbmodel))
205
205
206 nbfile = pjoin(self.notebook_dir.name, 'foo', 'a.ipynb')
206 nbfile = pjoin(self.notebook_dir.name, 'foo', 'a.ipynb')
207 with open(nbfile, 'r') as f:
207 with io.open(nbfile, 'r', encoding='utf-8') as f:
208 newnb = read(f, format='ipynb')
208 newnb = read(f, format='ipynb')
209 self.assertEqual(newnb.worksheets[0].cells[0].source,
209 self.assertEqual(newnb.worksheets[0].cells[0].source,
210 'Created by test')
210 u'Created by test Β³')
211 nbcontent = self.nb_api.read('a.ipynb', 'foo').json()['content']
212 newnb = to_notebook_json(nbcontent)
213 self.assertEqual(newnb.worksheets[0].cells[0].source,
214 u'Created by test Β³')
211
215
212 # Save and rename
216 # Save and rename
213 nbmodel= {'name': 'a2.ipynb', 'path':'foo/bar', 'content': nb}
217 nbmodel= {'name': 'a2.ipynb', 'path':'foo/bar', 'content': nb}
214 resp = self.nb_api.save('a.ipynb', path='foo', body=jsonapi.dumps(nbmodel))
218 resp = self.nb_api.save('a.ipynb', path='foo', body=jsonapi.dumps(nbmodel))
215 saved = resp.json()
219 saved = resp.json()
216 self.assertEqual(saved['name'], 'a2.ipynb')
220 self.assertEqual(saved['name'], 'a2.ipynb')
217 self.assertEqual(saved['path'], 'foo/bar')
221 self.assertEqual(saved['path'], 'foo/bar')
218 assert os.path.isfile(pjoin(self.notebook_dir.name,'foo','bar','a2.ipynb'))
222 assert os.path.isfile(pjoin(self.notebook_dir.name,'foo','bar','a2.ipynb'))
219 assert not os.path.isfile(pjoin(self.notebook_dir.name, 'foo', 'a.ipynb'))
223 assert not os.path.isfile(pjoin(self.notebook_dir.name, 'foo', 'a.ipynb'))
220 with assert_http_error(404):
224 with assert_http_error(404):
221 self.nb_api.read('a.ipynb', 'foo')
225 self.nb_api.read('a.ipynb', 'foo')
222
226
223 def test_checkpoints(self):
227 def test_checkpoints(self):
224 resp = self.nb_api.read('a.ipynb', 'foo')
228 resp = self.nb_api.read('a.ipynb', 'foo')
225 r = self.nb_api.new_checkpoint('a.ipynb', 'foo')
229 r = self.nb_api.new_checkpoint('a.ipynb', 'foo')
226 self.assertEqual(r.status_code, 201)
230 self.assertEqual(r.status_code, 201)
227 cp1 = r.json()
231 cp1 = r.json()
228 self.assertEqual(set(cp1), {'checkpoint_id', 'last_modified'})
232 self.assertEqual(set(cp1), {'checkpoint_id', 'last_modified'})
229 self.assertEqual(r.headers['Location'].split('/')[-1], cp1['checkpoint_id'])
233 self.assertEqual(r.headers['Location'].split('/')[-1], cp1['checkpoint_id'])
230
234
231 # Modify it
235 # Modify it
232 nbcontent = jsonapi.loads(resp.text)['content']
236 nbcontent = jsonapi.loads(resp.text)['content']
233 nb = to_notebook_json(nbcontent)
237 nb = to_notebook_json(nbcontent)
234 ws = new_worksheet()
238 ws = new_worksheet()
235 nb.worksheets = [ws]
239 nb.worksheets = [ws]
236 hcell = new_heading_cell('Created by test')
240 hcell = new_heading_cell('Created by test')
237 ws.cells.append(hcell)
241 ws.cells.append(hcell)
238 # Save
242 # Save
239 nbmodel= {'name': 'a.ipynb', 'path':'foo', 'content': nb}
243 nbmodel= {'name': 'a.ipynb', 'path':'foo', 'content': nb}
240 resp = self.nb_api.save('a.ipynb', path='foo', body=jsonapi.dumps(nbmodel))
244 resp = self.nb_api.save('a.ipynb', path='foo', body=jsonapi.dumps(nbmodel))
241
245
242 # List checkpoints
246 # List checkpoints
243 cps = self.nb_api.get_checkpoints('a.ipynb', 'foo').json()
247 cps = self.nb_api.get_checkpoints('a.ipynb', 'foo').json()
244 self.assertEqual(cps, [cp1])
248 self.assertEqual(cps, [cp1])
245
249
246 nbcontent = self.nb_api.read('a.ipynb', 'foo').json()['content']
250 nbcontent = self.nb_api.read('a.ipynb', 'foo').json()['content']
247 nb = to_notebook_json(nbcontent)
251 nb = to_notebook_json(nbcontent)
248 self.assertEqual(nb.worksheets[0].cells[0].source, 'Created by test')
252 self.assertEqual(nb.worksheets[0].cells[0].source, 'Created by test')
249
253
250 # Restore cp1
254 # Restore cp1
251 r = self.nb_api.restore_checkpoint('a.ipynb', 'foo', cp1['checkpoint_id'])
255 r = self.nb_api.restore_checkpoint('a.ipynb', 'foo', cp1['checkpoint_id'])
252 self.assertEqual(r.status_code, 204)
256 self.assertEqual(r.status_code, 204)
253 nbcontent = self.nb_api.read('a.ipynb', 'foo').json()['content']
257 nbcontent = self.nb_api.read('a.ipynb', 'foo').json()['content']
254 nb = to_notebook_json(nbcontent)
258 nb = to_notebook_json(nbcontent)
255 self.assertEqual(nb.worksheets, [])
259 self.assertEqual(nb.worksheets, [])
256
260
257 # Delete cp1
261 # Delete cp1
258 r = self.nb_api.delete_checkpoint('a.ipynb', 'foo', cp1['checkpoint_id'])
262 r = self.nb_api.delete_checkpoint('a.ipynb', 'foo', cp1['checkpoint_id'])
259 self.assertEqual(r.status_code, 204)
263 self.assertEqual(r.status_code, 204)
260 cps = self.nb_api.get_checkpoints('a.ipynb', 'foo').json()
264 cps = self.nb_api.get_checkpoints('a.ipynb', 'foo').json()
261 self.assertEqual(cps, [])
265 self.assertEqual(cps, [])
General Comments 0
You need to be logged in to leave comments. Login now