c37fef33b46540db566d8a98a690e8da4ccda9ff
[wolnelektury.git] / src / api / tests / tests.py
1 # This file is part of Wolnelektury, licensed under GNU Affero GPLv3 or later.
2 # Copyright © Fundacja Nowoczesna Polska. See NOTICE for more information.
3 #
4 from base64 import b64encode
5 from os import path
6 import hashlib
7 import hmac
8 import json
9 from io import BytesIO
10 from time import time
11 from urllib.parse import quote, urlencode, parse_qs
12
13 from django.contrib.auth.models import User
14 from django.core.files.uploadedfile import SimpleUploadedFile
15 from django.test import TestCase
16 from django.test.utils import override_settings
17 from unittest.mock import patch
18 from api.models import Consumer, Token
19
20 from catalogue.models import Book, Tag
21 from picture.forms import PictureImportForm
22 from picture.models import Picture
23 import picture.tests
24
25
26 @override_settings(
27     NO_SEARCH_INDEX=True,
28     CACHES={'default': {
29         'BACKEND': 'django.core.cache.backends.dummy.DummyCache'}},
30 )
31 class ApiTest(TestCase):
32     def load_json(self, url):
33         content = self.client.get(url).content
34         try:
35             data = json.loads(content)
36         except ValueError:
37             self.fail('No JSON could be decoded: %s' % content)
38         return data
39
40     def assert_response(self, url, name):
41         content = self.client.get(url).content.decode('utf-8').rstrip()
42         filename = path.join(path.dirname(__file__), 'res', 'responses', name)
43         with open(filename) as f:
44             good_content = f.read().rstrip()
45         self.assertEqual(content, good_content, content)
46
47     def assert_json_response(self, url, name):
48         data = self.load_json(url)
49         filename = path.join(path.dirname(__file__), 'res', 'responses', name)
50         with open(filename) as f:
51             good_data = json.load(f)
52         self.assertEqual(data, good_data, json.dumps(data, indent=4))
53
54     def assert_slugs(self, url, slugs):
55         have_slugs = [x['slug'] for x in self.load_json(url)]
56         self.assertEqual(have_slugs, slugs, have_slugs)
57
58
59 class BookTests(ApiTest):
60
61     def setUp(self):
62         self.tag = Tag.objects.create(category='author', slug='joe')
63         self.book = Book.objects.create(title='A Book', slug='a-book')
64         self.book_tagged = Book.objects.create(
65             title='Tagged Book', slug='tagged-book')
66         self.book_tagged.tags = [self.tag]
67         self.book_tagged.save()
68
69     def test_book_list(self):
70         books = self.load_json('/api/books/')
71         self.assertEqual(len(books), 2,
72                          'Wrong book list.')
73
74     def test_tagged_books(self):
75         books = self.load_json('/api/authors/joe/books/')
76
77         self.assertEqual([b['title'] for b in books], [self.book_tagged.title],
78                          'Wrong tagged book list.')
79
80     def test_detail(self):
81         book = self.load_json('/api/books/a-book/')
82         self.assertEqual(book['title'], self.book.title,
83                          'Wrong book details.')
84
85
86 class TagTests(ApiTest):
87
88     def setUp(self):
89         self.tag = Tag.objects.create(
90             category='author', slug='joe', name='Joe')
91         self.book = Book.objects.create(title='A Book', slug='a-book')
92         self.book.tags = [self.tag]
93         self.book.save()
94
95     def test_tag_list(self):
96         tags = self.load_json('/api/authors/')
97         self.assertEqual(len(tags), 1,
98                          'Wrong tag list.')
99
100     def test_tag_detail(self):
101         tag = self.load_json('/api/authors/joe/')
102         self.assertEqual(tag['name'], self.tag.name,
103                          'Wrong tag details.')
104
105
106 class PictureTests(ApiTest):
107     def test_publish(self):
108         slug = "kandinsky-composition-viii"
109         xml = SimpleUploadedFile(
110             'composition8.xml',
111             open(path.join(
112                 picture.tests.__path__[0], "files", slug + ".xml"
113             ), 'rb').read())
114         img = SimpleUploadedFile(
115             'kompozycja-8.png',
116             open(path.join(
117                 picture.tests.__path__[0], "files", slug + ".png"
118             ), 'rb').read())
119
120         import_form = PictureImportForm({}, {
121             'picture_xml_file': xml,
122             'picture_image_file': img
123             })
124
125         assert import_form.is_valid()
126         if import_form.is_valid():
127             import_form.save()
128
129         Picture.objects.get(slug=slug)
130
131
132 class BooksTests(ApiTest):
133     fixtures = ['test-books.yaml']
134
135     def test_books(self):
136         self.assert_json_response('/api/books/', 'books.json')
137         self.assert_json_response('/api/books/?new_api=true', 'books.json')
138         self.assert_response('/api/books/?format=xml', 'books.xml')
139
140         self.assert_slugs('/api/audiobooks/', ['parent'])
141         self.assert_slugs('/api/daisy/', ['parent'])
142         self.assert_slugs('/api/newest/', ['parent'])
143         self.assert_slugs('/api/parent_books/', ['parent'])
144         self.assert_slugs('/api/recommended/', ['parent'])
145
146         # Book paging.
147         self.assert_slugs('/api/books/after/grandchild/count/1/', ['parent'])
148         self.assert_slugs(
149             '/api/books/?new_api=true&after=$grandchild$3&count=1', ['parent'])
150
151         # By tag.
152         self.assert_slugs('/api/authors/john-doe/books/', ['parent'])
153         self.assert_slugs(
154             '/api/genres/sonet/books/?authors=john-doe',
155             ['parent'])
156         # It is probably a mistake that this doesn't filter:
157         self.assert_slugs(
158             '/api/books/?authors=john-doe',
159             ['child', 'grandchild', 'parent'])
160
161         # Parent books by tag.
162         # Notice this contains a grandchild, if a child doesn't have the tag.
163         # This probably isn't really intended behavior and should be redefined.
164         self.assert_slugs(
165             '/api/genres/sonet/parent_books/',
166             ['grandchild', 'parent'])
167
168     def test_ebooks(self):
169         self.assert_json_response('/api/ebooks/', 'ebooks.json')
170
171     def test_filter_books(self):
172         self.assert_json_response('/api/filter-books/', 'filter-books.json')
173         self.assert_slugs(
174             '/api/filter-books/?lektura=false',
175             ['child', 'grandchild', 'parent'])
176         self.assert_slugs(
177             '/api/filter-books/?lektura=true',
178             [])
179
180         self.assert_slugs(
181             '/api/filter-books/?preview=true',
182             ['grandchild'])
183         self.assert_slugs(
184             '/api/filter-books/?preview=false',
185             ['child', 'parent'])
186
187         self.assert_slugs(
188             '/api/filter-books/?audiobook=true',
189             ['parent'])
190         self.assert_slugs(
191             '/api/filter-books/?audiobook=false',
192             ['child', 'grandchild'])
193
194         self.assert_slugs('/api/filter-books/?genres=wiersz', ['child'])
195
196         self.assert_slugs('/api/filter-books/?search=parent', ['parent'])
197
198     def test_collections(self):
199         self.assert_json_response('/api/collections/', 'collections.json')
200         self.assert_json_response(
201             '/api/collections/a-collection/',
202             'collection.json')
203
204     def test_book(self):
205         self.assert_json_response('/api/books/parent/', 'books-parent.json')
206         self.assert_json_response('/api/books/child/', 'books-child.json')
207         self.assert_json_response(
208             '/api/books/grandchild/',
209             'books-grandchild.json')
210
211     def test_tags(self):
212         # List of tags by category.
213         self.assert_json_response('/api/genres/', 'tags.json')
214
215     def test_fragments(self):
216         # This is not supported, though it probably should be.
217         # self.assert_json_response(
218         #     '/api/books/child/fragments/',
219         #     'fragments.json')
220
221         self.assert_json_response(
222             '/api/genres/wiersz/fragments/',
223             'fragments.json')
224         self.assert_json_response(
225             '/api/books/child/fragments/an-anchor/',
226             'fragment.json')
227
228
229 class BlogTests(ApiTest):
230     def test_get(self):
231         self.assertEqual(self.load_json('/api/blog'), [])
232
233
234 class PreviewTests(ApiTest):
235     def unauth(self):
236         self.assert_json_response('/api/preview/', 'preview.json')
237
238
239 class OAuth1Tests(ApiTest):
240     @classmethod
241     def setUpClass(cls):
242         cls.user = User.objects.create(username='test')
243         cls.user.set_password('test')
244         cls.user.save()
245         cls.consumer_secret = 'len(quote(consumer secret))>=32'
246         Consumer.objects.create(
247             key='client',
248             secret=cls.consumer_secret
249         )
250
251     @classmethod
252     def tearDownClass(cls):
253         User.objects.all().delete()
254
255     def test_create_token(self):
256         # Fetch request token.
257         base_query = ("oauth_consumer_key=client&oauth_nonce=12345678&"
258                       "oauth_signature_method=HMAC-SHA1&oauth_timestamp={}&"
259                       "oauth_version=1.0".format(int(time())))
260         raw = '&'.join([
261             'GET',
262             quote('http://testserver/api/oauth/request_token/', safe=''),
263             quote(base_query, safe='')
264         ])
265         h = hmac.new(
266             (quote(self.consumer_secret) + '&').encode('latin1'),
267             raw.encode('latin1'),
268             hashlib.sha1
269         ).digest()
270         h = b64encode(h).rstrip(b'\n')
271         sign = quote(h)
272         query = "{}&oauth_signature={}".format(base_query, sign)
273         response = self.client.get('/api/oauth/request_token/?' + query)
274         request_token_data = parse_qs(response.content.decode('latin1'))
275         request_token = request_token_data['oauth_token'][0]
276         request_token_secret = request_token_data['oauth_token_secret'][0]
277
278         # Request token authorization.
279         self.client.login(username='test', password='test')
280         response = self.client.get('/api/oauth/authorize/?oauth_token=%s&oauth_callback=test://oauth.callback/' % request_token)
281         post_data = response.context['form'].initial
282
283         response = self.client.post('/api/oauth/authorize/?' + urlencode(post_data))
284         self.assertEqual(
285             response['Location'],
286             'test://oauth.callback/?oauth_token=' + request_token
287         )
288
289         # Fetch access token.
290         base_query = ("oauth_consumer_key=client&oauth_nonce=12345678&"
291                       "oauth_signature_method=HMAC-SHA1&oauth_timestamp={}&"
292                       "oauth_token={}&oauth_version=1.0".format(
293                           int(time()), request_token))
294         raw = '&'.join([
295             'GET',
296             quote('http://testserver/api/oauth/access_token/', safe=''),
297             quote(base_query, safe='')
298         ])
299         h = hmac.new(
300             (quote(self.consumer_secret) + '&' +
301              quote(request_token_secret, safe='')).encode('latin1'),
302             raw.encode('latin1'),
303             hashlib.sha1
304         ).digest()
305         h = b64encode(h).rstrip(b'\n')
306         sign = quote(h)
307         query = u"{}&oauth_signature={}".format(base_query, sign)
308         response = self.client.get(u'/api/oauth/access_token/?' + query)
309         access_token_data = parse_qs(response.content.decode('latin1'))
310         access_token = access_token_data['oauth_token'][0]
311
312         self.assertTrue(
313             Token.objects.filter(
314                 key=access_token,
315                 token_type=Token.ACCESS,
316                 user=self.user
317             ).exists())
318
319
320 class AuthorizedTests(ApiTest):
321     fixtures = ['test-books.yaml']
322
323     @classmethod
324     def setUpClass(cls):
325         super(AuthorizedTests, cls).setUpClass()
326         cls.user = User.objects.create(username='test')
327         cls.consumer = Consumer.objects.create(
328             key='client', secret='12345678901234567890123456789012')
329         cls.token = Token.objects.create(
330             key='123456789012345678',
331             secret='12345678901234567890123456789012',
332             user=cls.user,
333             consumer=cls.consumer,
334             token_type=Token.ACCESS,
335             timestamp=time())
336         cls.key = (cls.consumer.secret + '&' + cls.token.secret).encode('latin1')
337
338     @classmethod
339     def tearDownClass(cls):
340         cls.user.delete()
341         cls.consumer.delete()
342         super(AuthorizedTests, cls).tearDownClass()
343
344     def signed(self, url, method='GET', params=None, data=None):
345         auth_params = {
346             "oauth_consumer_key": self.consumer.key,
347             "oauth_nonce": ("%f" % time()).replace('.', ''),
348             "oauth_signature_method": "HMAC-SHA1",
349             "oauth_timestamp": int(time()),
350             "oauth_token": self.token.key,
351             "oauth_version": "1.0",
352         }
353
354         sign_params = {}
355         if params:
356             sign_params.update(params)
357         if data:
358             sign_params.update(data)
359         sign_params.update(auth_params)
360         raw = "&".join([
361             method.upper(),
362             quote('http://testserver' + url, safe=''),
363             quote("&".join(
364                 quote(str(k), safe='') + "=" + quote(str(v), safe='')
365                 for (k, v) in sorted(sign_params.items())))
366         ])
367         auth_params["oauth_signature"] = quote(b64encode(hmac.new(
368             self.key,
369             raw.encode('latin1'),
370             hashlib.sha1
371         ).digest()).rstrip(b'\n'))
372         auth = 'OAuth realm="API", ' + ', '.join(
373             '{}="{}"'.format(k, v) for (k, v) in auth_params.items())
374
375         if params:
376             url = url + '?' + urlencode(params)
377         return getattr(self.client, method.lower())(
378             url,
379             data=urlencode(data) if data else None,
380             content_type='application/x-www-form-urlencoded',
381             HTTP_AUTHORIZATION=auth,
382         )
383
384     def signed_json(self, url, method='GET', params=None, data=None):
385         return json.loads(self.signed(url, method, params, data).content)
386
387     def test_books(self):
388         self.assertEqual(
389             [b['liked'] for b in self.signed_json('/api/books/')],
390             [False, False, False]
391         )
392         data = self.signed_json('/api/books/child/')
393         self.assertFalse(data['parent']['liked'])
394         self.assertFalse(data['children'][0]['liked'])
395
396         self.assertEqual(
397             self.signed_json('/api/like/parent/'),
398             {"likes": False}
399         )
400         self.signed('/api/like/parent/', 'POST')
401         self.assertEqual(
402             self.signed_json('/api/like/parent/'),
403             {"likes": True}
404         )
405         # There are several endpoints where 'liked' appears.
406         self.assertTrue(self.signed_json('/api/parent_books/')[0]['liked'])
407         self.assertTrue(self.signed_json(
408             '/api/filter-books/', params={"search": "parent"})[0]['liked'])
409
410         self.assertTrue(self.signed_json(
411             '/api/books/child/')['parent']['liked'])
412         # Liked books go on shelf.
413         self.assertEqual(
414             [x['slug'] for x in self.signed_json('/api/shelf/likes/')],
415             ['parent'])
416
417         self.signed('/api/like/parent/', 'POST', {"action": "unlike"})
418         self.assertEqual(
419             self.signed_json('/api/like/parent/'),
420             {"likes": False}
421         )
422         self.assertFalse(self.signed_json('/api/parent_books/')[0]['liked'])
423
424     def test_reading(self):
425         self.assertEqual(
426             self.signed_json('/api/reading/parent/'),
427             {"state": "not_started"}
428         )
429         self.signed('/api/reading/parent/reading/', 'post')
430         self.assertEqual(
431             self.signed_json('/api/reading/parent/'),
432             {"state": "reading"}
433         )
434         self.assertEqual(
435             [x['slug'] for x in self.signed_json('/api/shelf/reading/')],
436             ['parent'])
437
438     def test_subscription(self):
439         self.assert_slugs('/api/preview/', ['grandchild'])
440         self.assertEqual(
441             self.signed_json('/api/username/'),
442             {"username": "test", "premium": False})
443         self.assertEqual(
444             self.signed('/api/epub/grandchild/').status_code,
445             403)
446
447         with patch('club.models.Membership.is_active_for', return_value=True):
448             self.assertEqual(
449                 self.signed_json('/api/username/'),
450                 {"username": "test", "premium": True})
451             with patch('django.core.files.storage.Storage.open',
452                        return_value=BytesIO(b"<epub>")):
453                 self.assertEqual(
454                     self.signed('/api/epub/grandchild/').content,
455                     b"<epub>")
456
457     def test_publish(self):
458         response = self.signed('/api/books/',
459                                method='POST',
460                                data={"data": json.dumps({})})
461         self.assertEqual(response.status_code, 403)
462
463         response = self.signed('/api/pictures/',
464                                method='POST',
465                                data={"data": json.dumps({})})
466         self.assertEqual(response.status_code, 403)
467
468         self.user.is_superuser = True
469         self.user.save()
470
471         with patch('catalogue.models.Book.from_xml_file') as mock:
472             response = self.signed('/api/books/',
473                                    method='POST',
474                                    data={"data": json.dumps({
475                                        "book_xml": "<utwor/>"
476                                    })})
477             self.assertTrue(mock.called)
478         self.assertEqual(response.status_code, 201)
479
480         with patch('picture.models.Picture.from_xml_file') as mock:
481             response = self.signed('/api/pictures/',
482                                    method='POST',
483                                    data={"data": json.dumps({
484                                        "picture_xml": "<utwor/>",
485                                        "picture_image_data": "Kg==",
486                                    })})
487             self.assertTrue(mock.called)
488         self.assertEqual(response.status_code, 201)
489
490         self.user.is_superuser = False
491         self.user.save()