test_contents_api.py 25 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723
  1. # coding: utf-8
  2. """Test the contents webservice API."""
  3. from contextlib import contextmanager
  4. from functools import partial
  5. import io
  6. import json
  7. import os
  8. import shutil
  9. import sys
  10. from unicodedata import normalize
  11. pjoin = os.path.join
  12. import requests
  13. from ..filecheckpoints import GenericFileCheckpoints
  14. from traitlets.config import Config
  15. from notebook.utils import url_path_join, url_escape, to_os_path
  16. from notebook.tests.launchnotebook import NotebookTestBase, assert_http_error
  17. from nbformat import write, from_dict
  18. from nbformat.v4 import (
  19. new_notebook, new_markdown_cell,
  20. )
  21. from nbformat import v2
  22. from ipython_genutils import py3compat
  23. from ipython_genutils.tempdir import TemporaryDirectory
  24. try: #PY3
  25. from base64 import encodebytes, decodebytes
  26. except ImportError: #PY2
  27. from base64 import encodestring as encodebytes, decodestring as decodebytes
  28. def uniq_stable(elems):
  29. """uniq_stable(elems) -> list
  30. Return from an iterable, a list of all the unique elements in the input,
  31. maintaining the order in which they first appear.
  32. """
  33. seen = set()
  34. return [x for x in elems if x not in seen and not seen.add(x)]
  35. def notebooks_only(dir_model):
  36. return [nb for nb in dir_model['content'] if nb['type']=='notebook']
  37. def dirs_only(dir_model):
  38. return [x for x in dir_model['content'] if x['type']=='directory']
  39. class API(object):
  40. """Wrapper for contents API calls."""
  41. def __init__(self, request):
  42. self.request = request
  43. def _req(self, verb, path, body=None, params=None):
  44. response = self.request(verb,
  45. url_path_join('api/contents', path),
  46. data=body, params=params,
  47. )
  48. response.raise_for_status()
  49. return response
  50. def list(self, path='/'):
  51. return self._req('GET', path)
  52. def read(self, path, type=None, format=None, content=None):
  53. params = {}
  54. if type is not None:
  55. params['type'] = type
  56. if format is not None:
  57. params['format'] = format
  58. if content == False:
  59. params['content'] = '0'
  60. return self._req('GET', path, params=params)
  61. def create_untitled(self, path='/', ext='.ipynb'):
  62. body = None
  63. if ext:
  64. body = json.dumps({'ext': ext})
  65. return self._req('POST', path, body)
  66. def mkdir_untitled(self, path='/'):
  67. return self._req('POST', path, json.dumps({'type': 'directory'}))
  68. def copy(self, copy_from, path='/'):
  69. body = json.dumps({'copy_from':copy_from})
  70. return self._req('POST', path, body)
  71. def create(self, path='/'):
  72. return self._req('PUT', path)
  73. def upload(self, path, body):
  74. return self._req('PUT', path, body)
  75. def mkdir(self, path='/'):
  76. return self._req('PUT', path, json.dumps({'type': 'directory'}))
  77. def copy_put(self, copy_from, path='/'):
  78. body = json.dumps({'copy_from':copy_from})
  79. return self._req('PUT', path, body)
  80. def save(self, path, body):
  81. return self._req('PUT', path, body)
  82. def delete(self, path='/'):
  83. return self._req('DELETE', path)
  84. def rename(self, path, new_path):
  85. body = json.dumps({'path': new_path})
  86. return self._req('PATCH', path, body)
  87. def get_checkpoints(self, path):
  88. return self._req('GET', url_path_join(path, 'checkpoints'))
  89. def new_checkpoint(self, path):
  90. return self._req('POST', url_path_join(path, 'checkpoints'))
  91. def restore_checkpoint(self, path, checkpoint_id):
  92. return self._req('POST', url_path_join(path, 'checkpoints', checkpoint_id))
  93. def delete_checkpoint(self, path, checkpoint_id):
  94. return self._req('DELETE', url_path_join(path, 'checkpoints', checkpoint_id))
  95. class APITest(NotebookTestBase):
  96. """Test the kernels web service API"""
  97. dirs_nbs = [('', 'inroot'),
  98. ('Directory with spaces in', 'inspace'),
  99. (u'unicodé', 'innonascii'),
  100. ('foo', 'a'),
  101. ('foo', 'b'),
  102. ('foo', 'name with spaces'),
  103. ('foo', u'unicodé'),
  104. ('foo/bar', 'baz'),
  105. ('ordering', 'A'),
  106. ('ordering', 'b'),
  107. ('ordering', 'C'),
  108. (u'å b', u'ç d'),
  109. ]
  110. hidden_dirs = ['.hidden', '__pycache__']
  111. # Don't include root dir.
  112. dirs = uniq_stable([py3compat.cast_unicode(d) for (d,n) in dirs_nbs[1:]])
  113. top_level_dirs = {normalize('NFC', d.split('/')[0]) for d in dirs}
  114. @staticmethod
  115. def _blob_for_name(name):
  116. return name.encode('utf-8') + b'\xFF'
  117. @staticmethod
  118. def _txt_for_name(name):
  119. return u'%s text file' % name
  120. def to_os_path(self, api_path):
  121. return to_os_path(api_path, root=self.notebook_dir)
  122. def make_dir(self, api_path):
  123. """Create a directory at api_path"""
  124. os_path = self.to_os_path(api_path)
  125. try:
  126. os.makedirs(os_path)
  127. except OSError:
  128. print("Directory already exists: %r" % os_path)
  129. def make_txt(self, api_path, txt):
  130. """Make a text file at a given api_path"""
  131. os_path = self.to_os_path(api_path)
  132. with io.open(os_path, 'w', encoding='utf-8') as f:
  133. f.write(txt)
  134. def make_blob(self, api_path, blob):
  135. """Make a binary file at a given api_path"""
  136. os_path = self.to_os_path(api_path)
  137. with io.open(os_path, 'wb') as f:
  138. f.write(blob)
  139. def make_nb(self, api_path, nb):
  140. """Make a notebook file at a given api_path"""
  141. os_path = self.to_os_path(api_path)
  142. with io.open(os_path, 'w', encoding='utf-8') as f:
  143. write(nb, f, version=4)
  144. def delete_dir(self, api_path):
  145. """Delete a directory at api_path, removing any contents."""
  146. os_path = self.to_os_path(api_path)
  147. shutil.rmtree(os_path, ignore_errors=True)
  148. def delete_file(self, api_path):
  149. """Delete a file at the given path if it exists."""
  150. if self.isfile(api_path):
  151. os.unlink(self.to_os_path(api_path))
  152. def isfile(self, api_path):
  153. return os.path.isfile(self.to_os_path(api_path))
  154. def isdir(self, api_path):
  155. return os.path.isdir(self.to_os_path(api_path))
  156. def setUp(self):
  157. for d in (self.dirs + self.hidden_dirs):
  158. self.make_dir(d)
  159. self.addCleanup(partial(self.delete_dir, d))
  160. for d, name in self.dirs_nbs:
  161. # create a notebook
  162. nb = new_notebook()
  163. nbname = u'{}/{}.ipynb'.format(d, name)
  164. self.make_nb(nbname, nb)
  165. self.addCleanup(partial(self.delete_file, nbname))
  166. # create a text file
  167. txt = self._txt_for_name(name)
  168. txtname = u'{}/{}.txt'.format(d, name)
  169. self.make_txt(txtname, txt)
  170. self.addCleanup(partial(self.delete_file, txtname))
  171. blob = self._blob_for_name(name)
  172. blobname = u'{}/{}.blob'.format(d, name)
  173. self.make_blob(blobname, blob)
  174. self.addCleanup(partial(self.delete_file, blobname))
  175. self.api = API(self.request)
  176. def test_list_notebooks(self):
  177. nbs = notebooks_only(self.api.list().json())
  178. self.assertEqual(len(nbs), 1)
  179. self.assertEqual(nbs[0]['name'], 'inroot.ipynb')
  180. nbs = notebooks_only(self.api.list('/Directory with spaces in/').json())
  181. self.assertEqual(len(nbs), 1)
  182. self.assertEqual(nbs[0]['name'], 'inspace.ipynb')
  183. nbs = notebooks_only(self.api.list(u'/unicodé/').json())
  184. self.assertEqual(len(nbs), 1)
  185. self.assertEqual(nbs[0]['name'], 'innonascii.ipynb')
  186. self.assertEqual(nbs[0]['path'], u'unicodé/innonascii.ipynb')
  187. nbs = notebooks_only(self.api.list('/foo/bar/').json())
  188. self.assertEqual(len(nbs), 1)
  189. self.assertEqual(nbs[0]['name'], 'baz.ipynb')
  190. self.assertEqual(nbs[0]['path'], 'foo/bar/baz.ipynb')
  191. nbs = notebooks_only(self.api.list('foo').json())
  192. self.assertEqual(len(nbs), 4)
  193. nbnames = { normalize('NFC', n['name']) for n in nbs }
  194. expected = [ u'a.ipynb', u'b.ipynb', u'name with spaces.ipynb', u'unicodé.ipynb']
  195. expected = { normalize('NFC', name) for name in expected }
  196. self.assertEqual(nbnames, expected)
  197. nbs = notebooks_only(self.api.list('ordering').json())
  198. nbnames = {n['name'] for n in nbs}
  199. expected = {'A.ipynb', 'b.ipynb', 'C.ipynb'}
  200. self.assertEqual(nbnames, expected)
  201. def test_list_dirs(self):
  202. dirs = dirs_only(self.api.list().json())
  203. dir_names = {normalize('NFC', d['name']) for d in dirs}
  204. self.assertEqual(dir_names, self.top_level_dirs) # Excluding hidden dirs
  205. def test_get_dir_no_content(self):
  206. for d in self.dirs:
  207. model = self.api.read(d, content=False).json()
  208. self.assertEqual(model['path'], d)
  209. self.assertEqual(model['type'], 'directory')
  210. self.assertIn('content', model)
  211. self.assertEqual(model['content'], None)
  212. def test_list_nonexistant_dir(self):
  213. with assert_http_error(404):
  214. self.api.list('nonexistant')
  215. def test_get_nb_contents(self):
  216. for d, name in self.dirs_nbs:
  217. path = url_path_join(d, name + '.ipynb')
  218. nb = self.api.read(path).json()
  219. self.assertEqual(nb['name'], u'%s.ipynb' % name)
  220. self.assertEqual(nb['path'], path)
  221. self.assertEqual(nb['type'], 'notebook')
  222. self.assertIn('content', nb)
  223. self.assertEqual(nb['format'], 'json')
  224. self.assertIn('metadata', nb['content'])
  225. self.assertIsInstance(nb['content']['metadata'], dict)
  226. def test_get_nb_no_content(self):
  227. for d, name in self.dirs_nbs:
  228. path = url_path_join(d, name + '.ipynb')
  229. nb = self.api.read(path, content=False).json()
  230. self.assertEqual(nb['name'], u'%s.ipynb' % name)
  231. self.assertEqual(nb['path'], path)
  232. self.assertEqual(nb['type'], 'notebook')
  233. self.assertIn('content', nb)
  234. self.assertEqual(nb['content'], None)
  235. def test_get_nb_invalid(self):
  236. nb = {
  237. 'nbformat': 4,
  238. 'metadata': {},
  239. 'cells': [{
  240. 'cell_type': 'wrong',
  241. 'metadata': {},
  242. }],
  243. }
  244. path = u'å b/Validate tést.ipynb'
  245. self.make_txt(path, py3compat.cast_unicode(json.dumps(nb)))
  246. model = self.api.read(path).json()
  247. self.assertEqual(model['path'], path)
  248. self.assertEqual(model['type'], 'notebook')
  249. self.assertIn('content', model)
  250. self.assertIn('message', model)
  251. self.assertIn("validation failed", model['message'].lower())
  252. def test_get_contents_no_such_file(self):
  253. # Name that doesn't exist - should be a 404
  254. with assert_http_error(404):
  255. self.api.read('foo/q.ipynb')
  256. def test_get_text_file_contents(self):
  257. for d, name in self.dirs_nbs:
  258. path = url_path_join(d, name + '.txt')
  259. model = self.api.read(path).json()
  260. self.assertEqual(model['name'], u'%s.txt' % name)
  261. self.assertEqual(model['path'], path)
  262. self.assertIn('content', model)
  263. self.assertEqual(model['format'], 'text')
  264. self.assertEqual(model['type'], 'file')
  265. self.assertEqual(model['content'], self._txt_for_name(name))
  266. # Name that doesn't exist - should be a 404
  267. with assert_http_error(404):
  268. self.api.read('foo/q.txt')
  269. # Specifying format=text should fail on a non-UTF-8 file
  270. with assert_http_error(400):
  271. self.api.read('foo/bar/baz.blob', type='file', format='text')
  272. def test_get_binary_file_contents(self):
  273. for d, name in self.dirs_nbs:
  274. path = url_path_join(d, name + '.blob')
  275. model = self.api.read(path).json()
  276. self.assertEqual(model['name'], u'%s.blob' % name)
  277. self.assertEqual(model['path'], path)
  278. self.assertIn('content', model)
  279. self.assertEqual(model['format'], 'base64')
  280. self.assertEqual(model['type'], 'file')
  281. self.assertEqual(
  282. decodebytes(model['content'].encode('ascii')),
  283. self._blob_for_name(name),
  284. )
  285. # Name that doesn't exist - should be a 404
  286. with assert_http_error(404):
  287. self.api.read('foo/q.txt')
  288. def test_get_bad_type(self):
  289. with assert_http_error(400):
  290. self.api.read(u'unicodé', type='file') # this is a directory
  291. with assert_http_error(400):
  292. self.api.read(u'unicodé/innonascii.ipynb', type='directory')
  293. def _check_created(self, resp, path, type='notebook'):
  294. self.assertEqual(resp.status_code, 201)
  295. location_header = py3compat.str_to_unicode(resp.headers['Location'])
  296. self.assertEqual(location_header, url_path_join(self.url_prefix, u'api/contents', url_escape(path)))
  297. rjson = resp.json()
  298. self.assertEqual(rjson['name'], path.rsplit('/', 1)[-1])
  299. self.assertEqual(rjson['path'], path)
  300. self.assertEqual(rjson['type'], type)
  301. isright = self.isdir if type == 'directory' else self.isfile
  302. assert isright(path)
  303. def test_create_untitled(self):
  304. resp = self.api.create_untitled(path=u'å b')
  305. self._check_created(resp, u'å b/Untitled.ipynb')
  306. # Second time
  307. resp = self.api.create_untitled(path=u'å b')
  308. self._check_created(resp, u'å b/Untitled1.ipynb')
  309. # And two directories down
  310. resp = self.api.create_untitled(path='foo/bar')
  311. self._check_created(resp, 'foo/bar/Untitled.ipynb')
  312. def test_create_untitled_txt(self):
  313. resp = self.api.create_untitled(path='foo/bar', ext='.txt')
  314. self._check_created(resp, 'foo/bar/untitled.txt', type='file')
  315. resp = self.api.read(path='foo/bar/untitled.txt')
  316. model = resp.json()
  317. self.assertEqual(model['type'], 'file')
  318. self.assertEqual(model['format'], 'text')
  319. self.assertEqual(model['content'], '')
  320. def test_upload(self):
  321. nb = new_notebook()
  322. nbmodel = {'content': nb, 'type': 'notebook'}
  323. path = u'å b/Upload tést.ipynb'
  324. resp = self.api.upload(path, body=json.dumps(nbmodel))
  325. self._check_created(resp, path)
  326. def test_mkdir_untitled(self):
  327. resp = self.api.mkdir_untitled(path=u'å b')
  328. self._check_created(resp, u'å b/Untitled Folder', type='directory')
  329. # Second time
  330. resp = self.api.mkdir_untitled(path=u'å b')
  331. self._check_created(resp, u'å b/Untitled Folder 1', type='directory')
  332. # And two directories down
  333. resp = self.api.mkdir_untitled(path='foo/bar')
  334. self._check_created(resp, 'foo/bar/Untitled Folder', type='directory')
  335. def test_mkdir(self):
  336. path = u'å b/New ∂ir'
  337. resp = self.api.mkdir(path)
  338. self._check_created(resp, path, type='directory')
  339. def test_mkdir_hidden_400(self):
  340. with assert_http_error(400):
  341. resp = self.api.mkdir(u'å b/.hidden')
  342. def test_upload_txt(self):
  343. body = u'ünicode téxt'
  344. model = {
  345. 'content' : body,
  346. 'format' : 'text',
  347. 'type' : 'file',
  348. }
  349. path = u'å b/Upload tést.txt'
  350. resp = self.api.upload(path, body=json.dumps(model))
  351. # check roundtrip
  352. resp = self.api.read(path)
  353. model = resp.json()
  354. self.assertEqual(model['type'], 'file')
  355. self.assertEqual(model['format'], 'text')
  356. self.assertEqual(model['content'], body)
  357. def test_upload_b64(self):
  358. body = b'\xFFblob'
  359. b64body = encodebytes(body).decode('ascii')
  360. model = {
  361. 'content' : b64body,
  362. 'format' : 'base64',
  363. 'type' : 'file',
  364. }
  365. path = u'å b/Upload tést.blob'
  366. resp = self.api.upload(path, body=json.dumps(model))
  367. # check roundtrip
  368. resp = self.api.read(path)
  369. model = resp.json()
  370. self.assertEqual(model['type'], 'file')
  371. self.assertEqual(model['path'], path)
  372. self.assertEqual(model['format'], 'base64')
  373. decoded = decodebytes(model['content'].encode('ascii'))
  374. self.assertEqual(decoded, body)
  375. def test_upload_v2(self):
  376. nb = v2.new_notebook()
  377. ws = v2.new_worksheet()
  378. nb.worksheets.append(ws)
  379. ws.cells.append(v2.new_code_cell(input='print("hi")'))
  380. nbmodel = {'content': nb, 'type': 'notebook'}
  381. path = u'å b/Upload tést.ipynb'
  382. resp = self.api.upload(path, body=json.dumps(nbmodel))
  383. self._check_created(resp, path)
  384. resp = self.api.read(path)
  385. data = resp.json()
  386. self.assertEqual(data['content']['nbformat'], 4)
  387. def test_copy(self):
  388. resp = self.api.copy(u'å b/ç d.ipynb', u'å b')
  389. self._check_created(resp, u'å b/ç d-Copy1.ipynb')
  390. resp = self.api.copy(u'å b/ç d.ipynb', u'å b')
  391. self._check_created(resp, u'å b/ç d-Copy2.ipynb')
  392. def test_copy_copy(self):
  393. resp = self.api.copy(u'å b/ç d.ipynb', u'å b')
  394. self._check_created(resp, u'å b/ç d-Copy1.ipynb')
  395. resp = self.api.copy(u'å b/ç d-Copy1.ipynb', u'å b')
  396. self._check_created(resp, u'å b/ç d-Copy2.ipynb')
  397. def test_copy_path(self):
  398. resp = self.api.copy(u'foo/a.ipynb', u'å b')
  399. self._check_created(resp, u'å b/a.ipynb')
  400. resp = self.api.copy(u'foo/a.ipynb', u'å b')
  401. self._check_created(resp, u'å b/a-Copy1.ipynb')
  402. def test_copy_put_400(self):
  403. with assert_http_error(400):
  404. resp = self.api.copy_put(u'å b/ç d.ipynb', u'å b/cøpy.ipynb')
  405. def test_copy_dir_400(self):
  406. # can't copy directories
  407. with assert_http_error(400):
  408. resp = self.api.copy(u'å b', u'foo')
  409. def test_delete(self):
  410. for d, name in self.dirs_nbs:
  411. print('%r, %r' % (d, name))
  412. resp = self.api.delete(url_path_join(d, name + '.ipynb'))
  413. self.assertEqual(resp.status_code, 204)
  414. for d in self.dirs + ['/']:
  415. nbs = notebooks_only(self.api.list(d).json())
  416. print('------')
  417. print(d)
  418. print(nbs)
  419. self.assertEqual(nbs, [])
  420. def test_delete_dirs(self):
  421. # depth-first delete everything, so we don't try to delete empty directories
  422. for name in sorted(self.dirs + ['/'], key=len, reverse=True):
  423. listing = self.api.list(name).json()['content']
  424. for model in listing:
  425. self.api.delete(model['path'])
  426. listing = self.api.list('/').json()['content']
  427. self.assertEqual(listing, [])
  428. def test_delete_non_empty_dir(self):
  429. if sys.platform == 'win32':
  430. self.skipTest("Disabled deleting non-empty dirs on Windows")
  431. # Test that non empty directory can be deleted
  432. self.api.delete(u'å b')
  433. # Check if directory has actually been deleted
  434. with assert_http_error(404):
  435. self.api.list(u'å b')
  436. def test_rename(self):
  437. resp = self.api.rename('foo/a.ipynb', 'foo/z.ipynb')
  438. self.assertEqual(resp.headers['Location'].split('/')[-1], 'z.ipynb')
  439. self.assertEqual(resp.json()['name'], 'z.ipynb')
  440. self.assertEqual(resp.json()['path'], 'foo/z.ipynb')
  441. assert self.isfile('foo/z.ipynb')
  442. nbs = notebooks_only(self.api.list('foo').json())
  443. nbnames = set(n['name'] for n in nbs)
  444. self.assertIn('z.ipynb', nbnames)
  445. self.assertNotIn('a.ipynb', nbnames)
  446. def test_checkpoints_follow_file(self):
  447. # Read initial file state
  448. orig = self.api.read('foo/a.ipynb')
  449. # Create a checkpoint of initial state
  450. r = self.api.new_checkpoint('foo/a.ipynb')
  451. cp1 = r.json()
  452. # Modify file and save
  453. nbcontent = json.loads(orig.text)['content']
  454. nb = from_dict(nbcontent)
  455. hcell = new_markdown_cell('Created by test')
  456. nb.cells.append(hcell)
  457. nbmodel = {'content': nb, 'type': 'notebook'}
  458. self.api.save('foo/a.ipynb', body=json.dumps(nbmodel))
  459. # Rename the file.
  460. self.api.rename('foo/a.ipynb', 'foo/z.ipynb')
  461. # Looking for checkpoints in the old location should yield no results.
  462. self.assertEqual(self.api.get_checkpoints('foo/a.ipynb').json(), [])
  463. # Looking for checkpoints in the new location should work.
  464. cps = self.api.get_checkpoints('foo/z.ipynb').json()
  465. self.assertEqual(cps, [cp1])
  466. # Delete the file. The checkpoint should be deleted as well.
  467. self.api.delete('foo/z.ipynb')
  468. cps = self.api.get_checkpoints('foo/z.ipynb').json()
  469. self.assertEqual(cps, [])
  470. def test_rename_existing(self):
  471. with assert_http_error(409):
  472. self.api.rename('foo/a.ipynb', 'foo/b.ipynb')
  473. def test_save(self):
  474. resp = self.api.read('foo/a.ipynb')
  475. nbcontent = json.loads(resp.text)['content']
  476. nb = from_dict(nbcontent)
  477. nb.cells.append(new_markdown_cell(u'Created by test ³'))
  478. nbmodel = {'content': nb, 'type': 'notebook'}
  479. resp = self.api.save('foo/a.ipynb', body=json.dumps(nbmodel))
  480. nbcontent = self.api.read('foo/a.ipynb').json()['content']
  481. newnb = from_dict(nbcontent)
  482. self.assertEqual(newnb.cells[0].source,
  483. u'Created by test ³')
  484. def test_checkpoints(self):
  485. resp = self.api.read('foo/a.ipynb')
  486. r = self.api.new_checkpoint('foo/a.ipynb')
  487. self.assertEqual(r.status_code, 201)
  488. cp1 = r.json()
  489. self.assertEqual(set(cp1), {'id', 'last_modified'})
  490. self.assertEqual(r.headers['Location'].split('/')[-1], cp1['id'])
  491. # Modify it
  492. nbcontent = json.loads(resp.text)['content']
  493. nb = from_dict(nbcontent)
  494. hcell = new_markdown_cell('Created by test')
  495. nb.cells.append(hcell)
  496. # Save
  497. nbmodel= {'content': nb, 'type': 'notebook'}
  498. resp = self.api.save('foo/a.ipynb', body=json.dumps(nbmodel))
  499. # List checkpoints
  500. cps = self.api.get_checkpoints('foo/a.ipynb').json()
  501. self.assertEqual(cps, [cp1])
  502. nbcontent = self.api.read('foo/a.ipynb').json()['content']
  503. nb = from_dict(nbcontent)
  504. self.assertEqual(nb.cells[0].source, 'Created by test')
  505. # Restore cp1
  506. r = self.api.restore_checkpoint('foo/a.ipynb', cp1['id'])
  507. self.assertEqual(r.status_code, 204)
  508. nbcontent = self.api.read('foo/a.ipynb').json()['content']
  509. nb = from_dict(nbcontent)
  510. self.assertEqual(nb.cells, [])
  511. # Delete cp1
  512. r = self.api.delete_checkpoint('foo/a.ipynb', cp1['id'])
  513. self.assertEqual(r.status_code, 204)
  514. cps = self.api.get_checkpoints('foo/a.ipynb').json()
  515. self.assertEqual(cps, [])
  516. def test_file_checkpoints(self):
  517. """
  518. Test checkpointing of non-notebook files.
  519. """
  520. filename = 'foo/a.txt'
  521. resp = self.api.read(filename)
  522. orig_content = json.loads(resp.text)['content']
  523. # Create a checkpoint.
  524. r = self.api.new_checkpoint(filename)
  525. self.assertEqual(r.status_code, 201)
  526. cp1 = r.json()
  527. self.assertEqual(set(cp1), {'id', 'last_modified'})
  528. self.assertEqual(r.headers['Location'].split('/')[-1], cp1['id'])
  529. # Modify the file and save.
  530. new_content = orig_content + '\nsecond line'
  531. model = {
  532. 'content': new_content,
  533. 'type': 'file',
  534. 'format': 'text',
  535. }
  536. resp = self.api.save(filename, body=json.dumps(model))
  537. # List checkpoints
  538. cps = self.api.get_checkpoints(filename).json()
  539. self.assertEqual(cps, [cp1])
  540. content = self.api.read(filename).json()['content']
  541. self.assertEqual(content, new_content)
  542. # Restore cp1
  543. r = self.api.restore_checkpoint(filename, cp1['id'])
  544. self.assertEqual(r.status_code, 204)
  545. restored_content = self.api.read(filename).json()['content']
  546. self.assertEqual(restored_content, orig_content)
  547. # Delete cp1
  548. r = self.api.delete_checkpoint(filename, cp1['id'])
  549. self.assertEqual(r.status_code, 204)
  550. cps = self.api.get_checkpoints(filename).json()
  551. self.assertEqual(cps, [])
  552. @contextmanager
  553. def patch_cp_root(self, dirname):
  554. """
  555. Temporarily patch the root dir of our checkpoint manager.
  556. """
  557. cpm = self.notebook.contents_manager.checkpoints
  558. old_dirname = cpm.root_dir
  559. cpm.root_dir = dirname
  560. try:
  561. yield
  562. finally:
  563. cpm.root_dir = old_dirname
  564. def test_checkpoints_separate_root(self):
  565. """
  566. Test that FileCheckpoints functions correctly even when it's
  567. using a different root dir from FileContentsManager. This also keeps
  568. the implementation honest for use with ContentsManagers that don't map
  569. models to the filesystem
  570. Override this method to a no-op when testing other managers.
  571. """
  572. with TemporaryDirectory() as td:
  573. with self.patch_cp_root(td):
  574. self.test_checkpoints()
  575. with TemporaryDirectory() as td:
  576. with self.patch_cp_root(td):
  577. self.test_file_checkpoints()
  578. class GenericFileCheckpointsAPITest(APITest):
  579. """
  580. Run the tests from APITest with GenericFileCheckpoints.
  581. """
  582. config = Config()
  583. config.FileContentsManager.checkpoints_class = GenericFileCheckpoints
  584. def test_config_did_something(self):
  585. self.assertIsInstance(
  586. self.notebook.contents_manager.checkpoints,
  587. GenericFileCheckpoints,
  588. )