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