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