##// END OF EJS Templates
Add failing test for listing nonexistant directory
Thomas Kluyver -
Show More
@@ -1,215 +1,213
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 from IPython.html.tests.launchnotebook import NotebookTestBase
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 19 from IPython.utils.data import uniq_stable
20 20
21 21 class NBAPI(object):
22 22 """Wrapper for notebook API calls."""
23 23 def __init__(self, base_url):
24 24 self.base_url = base_url
25 25
26 26 @property
27 27 def nb_url(self):
28 28 return url_path_join(self.base_url, 'api/notebooks')
29 29
30 30 def _req(self, verb, path, body=None):
31 31 response = requests.request(verb,
32 32 url_path_join(self.base_url, 'api/notebooks', path), data=body)
33 33 response.raise_for_status()
34 34 return response
35 35
36 36 def list(self, path='/'):
37 37 return self._req('GET', path)
38 38
39 39 def read(self, name, path='/'):
40 40 return self._req('GET', url_path_join(path, name))
41 41
42 42 def create_untitled(self, path='/'):
43 43 return self._req('POST', path)
44 44
45 45 def upload(self, name, body, path='/'):
46 46 return self._req('POST', url_path_join(path, name), body)
47 47
48 48 def copy(self, name, path='/'):
49 49 return self._req('POST', url_path_join(path, name, 'copy'))
50 50
51 51 def save(self, name, body, path='/'):
52 52 return self._req('PUT', url_path_join(path, name), body)
53 53
54 54 def delete(self, name, path='/'):
55 55 return self._req('DELETE', url_path_join(path, name))
56 56
57 57 def rename(self, name, path, new_name):
58 58 body = jsonapi.dumps({'name': new_name})
59 59 return self._req('PATCH', url_path_join(path, name), body)
60 60
61 61 class APITest(NotebookTestBase):
62 62 """Test the kernels web service API"""
63 63 dirs_nbs = [('', 'inroot'),
64 64 ('Directory with spaces in', 'inspace'),
65 65 (u'unicodé', 'innonascii'),
66 66 ('foo', 'a'),
67 67 ('foo', 'b'),
68 68 ('foo', 'name with spaces'),
69 69 ('foo', u'unicodé'),
70 70 ('foo/bar', 'baz'),
71 71 ]
72 72
73 73 dirs = uniq_stable([d for (d,n) in dirs_nbs])
74 74 del dirs[0] # remove ''
75 75
76 76 def setUp(self):
77 77 nbdir = self.notebook_dir.name
78 78
79 79 for d in self.dirs:
80 80 os.mkdir(pjoin(nbdir, d))
81 81
82 82 for d, name in self.dirs_nbs:
83 83 with io.open(pjoin(nbdir, d, '%s.ipynb' % name), 'w') as f:
84 84 nb = new_notebook(name=name)
85 85 write(nb, f, format='ipynb')
86 86
87 87 self.nb_api = NBAPI(self.base_url())
88 88
89 89 def tearDown(self):
90 90 nbdir = self.notebook_dir.name
91 91
92 92 for dname in ['foo', 'Directory with spaces in', u'unicodé']:
93 93 shutil.rmtree(pjoin(nbdir, dname), ignore_errors=True)
94 94
95 95 if os.path.isfile(pjoin(nbdir, 'inroot.ipynb')):
96 96 os.unlink(pjoin(nbdir, 'inroot.ipynb'))
97 97
98 98 def test_list_notebooks(self):
99 99 nbs = self.nb_api.list().json()
100 100 self.assertEqual(len(nbs), 1)
101 101 self.assertEqual(nbs[0]['name'], 'inroot.ipynb')
102 102
103 103 nbs = self.nb_api.list('/Directory with spaces in/').json()
104 104 self.assertEqual(len(nbs), 1)
105 105 self.assertEqual(nbs[0]['name'], 'inspace.ipynb')
106 106
107 107 nbs = self.nb_api.list(u'/unicodé/').json()
108 108 self.assertEqual(len(nbs), 1)
109 109 self.assertEqual(nbs[0]['name'], 'innonascii.ipynb')
110 110
111 111 nbs = self.nb_api.list('/foo/bar/').json()
112 112 self.assertEqual(len(nbs), 1)
113 113 self.assertEqual(nbs[0]['name'], 'baz.ipynb')
114 114
115 115 nbs = self.nb_api.list('foo').json()
116 116 self.assertEqual(len(nbs), 4)
117 117 nbnames = { normalize('NFC', n['name']) for n in nbs }
118 118 expected = [ u'a.ipynb', u'b.ipynb', u'name with spaces.ipynb', u'unicodé.ipynb']
119 119 expected = { normalize('NFC', name) for name in expected }
120 120 self.assertEqual(nbnames, expected)
121 121
122 def assert_404(self, name, path):
123 try:
124 self.nb_api.read(name, path)
125 except requests.HTTPError as e:
126 self.assertEqual(e.response.status_code, 404)
127 else:
128 assert False, "Reading a non-existent notebook should fail"
122 def test_list_nonexistant_dir(self):
123 with assert_http_error(404):
124 self.nb_api.list('nonexistant')
129 125
130 126 def test_get_contents(self):
131 127 for d, name in self.dirs_nbs:
132 128 nb = self.nb_api.read('%s.ipynb' % name, d+'/').json()
133 129 self.assertEqual(nb['name'], '%s.ipynb' % name)
134 130 self.assertIn('content', nb)
135 131 self.assertIn('metadata', nb['content'])
136 132 self.assertIsInstance(nb['content']['metadata'], dict)
137 133
138 134 # Name that doesn't exist - should be a 404
139 self.assert_404('q.ipynb', 'foo')
135 with assert_http_error(404):
136 self.nb_api.read('q.ipynb', 'foo')
140 137
141 138 def _check_nb_created(self, resp, name, path):
142 139 self.assertEqual(resp.status_code, 201)
143 140 self.assertEqual(resp.headers['Location'].split('/')[-1], name)
144 141 self.assertEqual(resp.json()['name'], name)
145 142 assert os.path.isfile(pjoin(self.notebook_dir.name, path, name))
146 143
147 144 def test_create_untitled(self):
148 145 resp = self.nb_api.create_untitled(path='foo')
149 146 self._check_nb_created(resp, 'Untitled0.ipynb', 'foo')
150 147
151 148 # Second time
152 149 resp = self.nb_api.create_untitled(path='foo')
153 150 self._check_nb_created(resp, 'Untitled1.ipynb', 'foo')
154 151
155 152 # And two directories down
156 153 resp = self.nb_api.create_untitled(path='foo/bar')
157 154 self._check_nb_created(resp, 'Untitled0.ipynb', pjoin('foo', 'bar'))
158 155
159 156 def test_upload(self):
160 157 nb = new_notebook(name='Upload test')
161 158 nbmodel = {'content': nb}
162 159 resp = self.nb_api.upload('Upload test.ipynb', path='foo',
163 160 body=jsonapi.dumps(nbmodel))
164 161 self._check_nb_created(resp, 'Upload test.ipynb', 'foo')
165 162
166 163 def test_copy(self):
167 164 resp = self.nb_api.copy('a.ipynb', path='foo')
168 165 self._check_nb_created(resp, 'a-Copy0.ipynb', 'foo')
169 166
170 167 def test_delete(self):
171 168 for d, name in self.dirs_nbs:
172 169 resp = self.nb_api.delete('%s.ipynb' % name, d)
173 170 self.assertEqual(resp.status_code, 204)
174 171
175 172 for d in self.dirs + ['/']:
176 173 nbs = self.nb_api.list(d).json()
177 174 self.assertEqual(len(nbs), 0)
178 175
179 176 def test_rename(self):
180 177 resp = self.nb_api.rename('a.ipynb', 'foo', 'z.ipynb')
181 178 self.assertEqual(resp.headers['Location'].split('/')[-1], 'z.ipynb')
182 179 self.assertEqual(resp.json()['name'], 'z.ipynb')
183 180 assert os.path.isfile(pjoin(self.notebook_dir.name, 'foo', 'z.ipynb'))
184 181
185 182 nbs = self.nb_api.list('foo').json()
186 183 nbnames = set(n['name'] for n in nbs)
187 184 self.assertIn('z.ipynb', nbnames)
188 185 self.assertNotIn('a.ipynb', nbnames)
189 186
190 187 def test_save(self):
191 188 resp = self.nb_api.read('a.ipynb', 'foo')
192 189 nbcontent = jsonapi.loads(resp.text)['content']
193 190 nb = to_notebook_json(nbcontent)
194 191 ws = new_worksheet()
195 192 nb.worksheets = [ws]
196 193 ws.cells.append(new_heading_cell('Created by test'))
197 194
198 195 nbmodel= {'name': 'a.ipynb', 'path':'foo', 'content': nb}
199 196 resp = self.nb_api.save('a.ipynb', path='foo', body=jsonapi.dumps(nbmodel))
200 197
201 198 nbfile = pjoin(self.notebook_dir.name, 'foo', 'a.ipynb')
202 199 with open(nbfile, 'r') as f:
203 200 newnb = read(f, format='ipynb')
204 201 self.assertEqual(newnb.worksheets[0].cells[0].source,
205 202 'Created by test')
206 203
207 204 # Save and rename
208 205 nbmodel= {'name': 'a2.ipynb', 'path':'foo/bar', 'content': nb}
209 206 resp = self.nb_api.save('a.ipynb', path='foo', body=jsonapi.dumps(nbmodel))
210 207 saved = resp.json()
211 208 self.assertEqual(saved['name'], 'a2.ipynb')
212 209 self.assertEqual(saved['path'], 'foo/bar')
213 210 assert os.path.isfile(pjoin(self.notebook_dir.name,'foo','bar','a2.ipynb'))
214 211 assert not os.path.isfile(pjoin(self.notebook_dir.name, 'foo', 'a.ipynb'))
215 self.assert_404('a.ipynb', 'foo') No newline at end of file
212 with assert_http_error(404):
213 self.nb_api.read('a.ipynb', 'foo') No newline at end of file
@@ -1,113 +1,106
1 1 """Test the sessions web service API."""
2 2
3 3 import io
4 4 import os
5 5 import json
6 6 import requests
7 7 import shutil
8 8
9 9 pjoin = os.path.join
10 10
11 11 from IPython.html.utils import url_path_join
12 from IPython.html.tests.launchnotebook import NotebookTestBase
12 from IPython.html.tests.launchnotebook import NotebookTestBase, assert_http_error
13 13 from IPython.nbformat.current import new_notebook, write
14 14
15 15 class SessionAPI(object):
16 16 """Wrapper for notebook API calls."""
17 17 def __init__(self, base_url):
18 18 self.base_url = base_url
19 19
20 20 def _req(self, verb, path, body=None):
21 21 response = requests.request(verb,
22 22 url_path_join(self.base_url, 'api/sessions', path), data=body)
23 23
24 24 if 400 <= response.status_code < 600:
25 25 try:
26 26 response.reason = response.json()['message']
27 27 except:
28 28 pass
29 29 response.raise_for_status()
30 30
31 31 return response
32 32
33 33 def list(self):
34 34 return self._req('GET', '')
35 35
36 36 def get(self, id):
37 37 return self._req('GET', id)
38 38
39 39 def create(self, name, path):
40 40 body = json.dumps({'notebook': {'name':name, 'path':path}})
41 41 return self._req('POST', '', body)
42 42
43 43 def modify(self, id, name, path):
44 44 body = json.dumps({'notebook': {'name':name, 'path':path}})
45 45 return self._req('PATCH', id, body)
46 46
47 47 def delete(self, id):
48 48 return self._req('DELETE', id)
49 49
50 50 class SessionAPITest(NotebookTestBase):
51 51 """Test the sessions web service API"""
52 52 def setUp(self):
53 53 nbdir = self.notebook_dir.name
54 54 os.mkdir(pjoin(nbdir, 'foo'))
55 55
56 56 with io.open(pjoin(nbdir, 'foo', 'nb1.ipynb'), 'w') as f:
57 57 nb = new_notebook(name='nb1')
58 58 write(nb, f, format='ipynb')
59 59
60 60 self.sess_api = SessionAPI(self.base_url())
61 61
62 62 def tearDown(self):
63 63 for session in self.sess_api.list().json():
64 64 self.sess_api.delete(session['id'])
65 65 shutil.rmtree(pjoin(self.notebook_dir.name, 'foo'))
66 66
67 def assert_404(self, id):
68 try:
69 self.sess_api.get(id)
70 except requests.HTTPError as e:
71 self.assertEqual(e.response.status_code, 404)
72 else:
73 assert False, "Getting nonexistent session didn't give HTTP error"
74
75 67 def test_create(self):
76 68 sessions = self.sess_api.list().json()
77 69 self.assertEqual(len(sessions), 0)
78 70
79 71 resp = self.sess_api.create('nb1.ipynb', 'foo')
80 72 self.assertEqual(resp.status_code, 201)
81 73 newsession = resp.json()
82 74 self.assertIn('id', newsession)
83 75 self.assertEqual(newsession['notebook']['name'], 'nb1.ipynb')
84 76 self.assertEqual(newsession['notebook']['path'], 'foo')
85 77
86 78 sessions = self.sess_api.list().json()
87 79 self.assertEqual(sessions, [newsession])
88 80
89 81 # Retrieve it
90 82 sid = newsession['id']
91 83 got = self.sess_api.get(sid).json()
92 84 self.assertEqual(got, newsession)
93 85
94 86 def test_delete(self):
95 87 newsession = self.sess_api.create('nb1.ipynb', 'foo').json()
96 88 sid = newsession['id']
97 89
98 90 resp = self.sess_api.delete(sid)
99 91 self.assertEqual(resp.status_code, 204)
100 92
101 93 sessions = self.sess_api.list().json()
102 94 self.assertEqual(sessions, [])
103 95
104 self.assert_404(sid)
96 with assert_http_error(404):
97 self.sess_api.get(sid)
105 98
106 99 def test_modify(self):
107 100 newsession = self.sess_api.create('nb1.ipynb', 'foo').json()
108 101 sid = newsession['id']
109 102
110 103 changed = self.sess_api.modify(sid, 'nb2.ipynb', '').json()
111 104 self.assertEqual(changed['id'], sid)
112 105 self.assertEqual(changed['notebook']['name'], 'nb2.ipynb')
113 106 self.assertEqual(changed['notebook']['path'], '')
@@ -1,69 +1,81
1 1 """Base class for notebook tests."""
2 2
3 3 import sys
4 4 import time
5 5 import requests
6 from contextlib import contextmanager
6 7 from subprocess import Popen, PIPE
7 8 from unittest import TestCase
8 9
9 10 from IPython.utils.tempdir import TemporaryDirectory
10 11
11
12 12 class NotebookTestBase(TestCase):
13 13 """A base class for tests that need a running notebook.
14 14
15 15 This creates an empty profile in a temp ipython_dir
16 16 and then starts the notebook server with a separate temp notebook_dir.
17 17 """
18 18
19 19 port = 12341
20 20
21 21 @classmethod
22 22 def wait_until_alive(cls):
23 23 """Wait for the server to be alive"""
24 24 url = 'http://localhost:%i/api/notebooks' % cls.port
25 25 while True:
26 26 try:
27 27 requests.get(url)
28 28 except requests.exceptions.ConnectionError:
29 29 time.sleep(.1)
30 30 else:
31 31 break
32 32
33 33 @classmethod
34 34 def wait_until_dead(cls):
35 35 """Wait for the server to stop getting requests after shutdown"""
36 36 url = 'http://localhost:%i/api/notebooks' % cls.port
37 37 while True:
38 38 try:
39 39 requests.get(url)
40 40 except requests.exceptions.ConnectionError:
41 41 break
42 42 else:
43 43 time.sleep(.1)
44 44
45 45 @classmethod
46 46 def setup_class(cls):
47 47 cls.ipython_dir = TemporaryDirectory()
48 48 cls.notebook_dir = TemporaryDirectory()
49 49 notebook_args = [
50 50 sys.executable, '-c',
51 51 'from IPython.html.notebookapp import launch_new_instance; launch_new_instance()',
52 52 '--port=%d' % cls.port,
53 53 '--no-browser',
54 54 '--ipython-dir=%s' % cls.ipython_dir.name,
55 55 '--notebook-dir=%s' % cls.notebook_dir.name
56 56 ]
57 57 cls.notebook = Popen(notebook_args, stdout=PIPE, stderr=PIPE)
58 58 cls.wait_until_alive()
59 59
60 60 @classmethod
61 61 def teardown_class(cls):
62 62 cls.notebook.terminate()
63 63 cls.ipython_dir.cleanup()
64 64 cls.notebook_dir.cleanup()
65 65 cls.wait_until_dead()
66 66
67 67 @classmethod
68 68 def base_url(cls):
69 69 return 'http://localhost:%i/' % cls.port
70
71
72 @contextmanager
73 def assert_http_error(status):
74 try:
75 yield
76 except requests.HTTPError as e:
77 real_status = e.response.status_code
78 assert real_status == status, \
79 "Expected status %d, got %d" % (real_status, status)
80 else:
81 assert False, "Expected HTTP error status" No newline at end of file
General Comments 0
You need to be logged in to leave comments. Login now