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