src/api/tests/tests.py

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