From b535a14ce4f3707b66b64c5de3b6458a42f061d3 2013-10-17 21:09:17 From: Thomas Kluyver Date: 2013-10-17 21:09:17 Subject: [PATCH] Improve Session REST API tests --- diff --git a/IPython/html/services/sessions/tests/test_sessions_api.py b/IPython/html/services/sessions/tests/test_sessions_api.py index 3b88be2..991f32e 100644 --- a/IPython/html/services/sessions/tests/test_sessions_api.py +++ b/IPython/html/services/sessions/tests/test_sessions_api.py @@ -1,96 +1,114 @@ """Test the sessions web service API.""" - +import io import os -import sys import json import requests +import shutil + +pjoin = os.path.join from IPython.utils.jsonutil import date_default from IPython.html.utils import url_path_join from IPython.html.tests.launchnotebook import NotebookTestBase +from IPython.nbformat.current import new_notebook, write + +class SessionAPI(object): + """Wrapper for notebook API calls.""" + def __init__(self, base_url): + self.base_url = base_url + + def _req(self, verb, path, body=None): + response = requests.request(verb, + url_path_join(self.base_url, 'api/sessions', path), data=body) + + if 400 <= response.status_code < 600: + try: + response.reason = response.json()['message'] + except: + pass + response.raise_for_status() + + return response + + def list(self): + return self._req('GET', '') + + def get(self, id): + return self._req('GET', id) + + def create(self, name, path): + body = json.dumps({'notebook': {'name':name, 'path':path}}) + return self._req('POST', '', body) + + def modify(self, id, name, path): + body = json.dumps({'name':name, 'path':path}) + return self._req('PATCH', id, body) + + def delete(self, id): + return self._req('DELETE', id) class SessionAPITest(NotebookTestBase): """Test the sessions web service API""" - - def notebook_url(self): - return url_path_join(super(SessionAPITest,self).base_url(), 'api/notebooks') - - def session_url(self): - return super(SessionAPITest,self).base_url() + 'api/sessions' - - def mknb(self, name='', path='/'): - url = self.notebook_url() + path - return url, requests.post(url) - - def delnb(self, name, path='/'): - url = self.notebook_url() + path + name - r = requests.delete(url) - return r.status_code - - def test_no_sessions(self): - """Make sure there are no sessions running at the start""" - url = self.session_url() - r = requests.get(url) - self.assertEqual(r.json(), []) - - def test_session_root_handler(self): - # POST a session - url, nb = self.mknb() - notebook = nb.json() - model = {'notebook': {'name':notebook['name'], 'path': notebook['path']}} - r = requests.post(self.session_url(), data=json.dumps(model, default=date_default)) - data = r.json() - assert isinstance(data, dict) - self.assertIn('name', data['notebook']) - self.assertEqual(data['notebook']['name'], notebook['name']) - - # GET sessions - r = requests.get(self.session_url()) - assert isinstance(r.json(), list) - assert isinstance(r.json()[0], dict) - self.assertEqual(r.json()[0]['id'], data['id']) - - # Clean up - self.delnb('Untitled0.ipynb') - sess_url = self.session_url() +'/'+data['id'] - r = requests.delete(sess_url) - self.assertEqual(r.status_code, 204) - - def test_session_handler(self): - # Create a session - url, nb = self.mknb() - notebook = nb.json() - model = {'notebook': {'name':notebook['name'], 'path': notebook['path']}} - r = requests.post(self.session_url(), data=json.dumps(model, default=date_default)) - session = r.json() - - # GET a session - sess_url = self.session_url() + '/' + session['id'] - r = requests.get(sess_url) - assert isinstance(r.json(), dict) - self.assertEqual(r.json(), session) - - # PATCH a session - model = {'notebook': {'name':'test.ipynb', 'path': '/'}} - r = requests.patch(sess_url, data=json.dumps(model, default=date_default)) - - # Patching the notebook webservice too (just for consistency) - requests.patch(self.notebook_url() + '/Untitled0.ipynb', - data=json.dumps({'name':'test.ipynb'})) - print r.json() - assert isinstance(r.json(), dict) - self.assertIn('name', r.json()['notebook']) - self.assertIn('id', r.json()) - self.assertEqual(r.json()['notebook']['name'], 'test.ipynb') - self.assertEqual(r.json()['id'], session['id']) - - # DELETE a session - r = requests.delete(sess_url) - self.assertEqual(r.status_code, 204) - r = requests.get(self.session_url()) - self.assertEqual(r.json(), []) - - # Clean up - r = self.delnb('test.ipynb') - self.assertEqual(r, 204) \ No newline at end of file + def setUp(self): + nbdir = self.notebook_dir.name + os.mkdir(pjoin(nbdir, 'foo')) + + with io.open(pjoin(nbdir, 'foo', 'nb1.ipynb'), 'w') as f: + nb = new_notebook(name='nb1') + write(nb, f, format='ipynb') + + self.sess_api = SessionAPI(self.base_url()) + + def tearDown(self): + for session in self.sess_api.list().json(): + self.sess_api.delete(session['id']) + shutil.rmtree(pjoin(self.notebook_dir.name, 'foo')) + + def assert_404(self, id): + try: + self.sess_api.get(id) + except requests.HTTPError as e: + self.assertEqual(e.response.status_code, 404) + else: + assert False, "Getting nonexistent session didn't give HTTP error" + + def test_create(self): + sessions = self.sess_api.list().json() + self.assertEqual(len(sessions), 0) + + resp = self.sess_api.create('nb1.ipynb', 'foo') + self.assertEqual(resp.status_code, 201) + newsession = resp.json() + self.assertIn('id', newsession) + self.assertEqual(newsession['notebook']['name'], 'nb1.ipynb') + self.assertEqual(newsession['notebook']['path'], 'foo') + + sessions = self.sess_api.list().json() + self.assertEqual(sessions, [newsession]) + + # Retrieve it + sid = newsession['id'] + got = self.sess_api.get(sid) + self.assertEqual(got, newsession) + + def test_delete(self): + newsession = self.sess_api.create('nb1.ipynb', 'foo').json() + sid = newsession['id'] + + resp = self.sess_api.delete(sid) + self.assertEqual(resp.status_code, 204) + + sessions = self.sess_api.list().json() + self.assertEqual(sessions, []) + + self.assert_404(sid) + + def test_modify(self): + newsession = self.sess_api.create('nb1.ipynb', 'foo').json() + sid = newsession['id'] + + changed = self.sess_api.modify(sid, 'nb2.ipynb', '').json() + self.assertEqual(changed['id'], sid) + self.assertEqual(newsession['notebook']['name'], 'nb2.ipynb') + self.assertEqual(newsession['notebook']['path'], '')