mygpo/api/tests.py

   1 #
   2 # This file is part of my.gpodder.org.
   3 #
   4 # my.gpodder.org is free software: you can redistribute it and/or modify it
   5 # under the terms of the GNU Affero General Public License as published by
   6 # the Free Software Foundation, either version 3 of the License, or (at your
   7 # option) any later version.
   8 #
   9 # my.gpodder.org is distributed in the hope that it will be useful, but
  10 # WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
  11 # or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public
  12 # License for more details.
  13 #
  14 # You should have received a copy of the GNU Affero General Public License
  15 # along with my.gpodder.org. If not, see <http://www.gnu.org/licenses/>.
  16 #
  17
  18 from __future__ import unicode_literals
  19
  20 import unittest
  21 import doctest
  22 from urllib import urlencode
  23 from copy import deepcopy
  24
  25 from django.test.client import Client
  26 from django.test import TestCase
  27 from django.core.urlresolvers import reverse
  28
  29 from mygpo.podcasts.models import Podcast, Episode
  30 from mygpo.api.advanced import episodes
  31 from mygpo.users.models import User
  32 from mygpo.test import create_auth_string, anon_request
  33 from mygpo.core.json import json
  34
  35
  36 class AdvancedAPITests(unittest.TestCase):
  37
  38     def setUp(self):
  39         self.password = 'asdf'
  40         self.username = 'adv-api-user'
  41         self.user = User(username=self.username, email='user@example.com')
  42         self.user.set_password(self.password)
  43         self.user.save()
  44         self.user.is_active = True
  45         self.client = Client()
  46
  47         self.extra = {
  48             'HTTP_AUTHORIZATION': create_auth_string(self.username,
  49                                                      self.password)
  50         }
  51
  52         self.action_data = [
  53             {
  54                 "podcast": "http://example.com/feed.rss",
  55                 "episode": "http://example.com/files/s01e20.mp3",
  56                 "device": "gpodder_abcdef123",
  57                 "action": "download",
  58                 "timestamp": "2009-12-12T09:00:00"
  59             },
  60             {
  61                 "podcast": "http://example.org/podcast.php",
  62                 "episode": "http://ftp.example.org/foo.ogg",
  63                 "action": "play",
  64                 "started": 15,
  65                 "position": 120,
  66                 "total":  500
  67             }
  68         ]
  69
  70     def tearDown(self):
  71         self.user.delete()
  72
  73     def test_episode_actions(self):
  74         url = reverse(episodes, kwargs={
  75             'version': '2',
  76             'username': self.user.username,
  77         })
  78
  79         # upload actions
  80         response = self.client.post(url, json.dumps(self.action_data),
  81                                     content_type="application/json",
  82                                     **self.extra)
  83         self.assertEqual(response.status_code, 200, response.content)
  84
  85         response = self.client.get(url, {'since': '0'}, **self.extra)
  86         self.assertEqual(response.status_code, 200, response.content)
  87         response_obj = json.loads(response.content)
  88         actions = response_obj['actions']
  89         self.assertTrue(self.compare_action_list(self.action_data, actions))
  90
  91     def compare_action_list(self, as1, as2):
  92         for a1 in as1:
  93             found = False
  94             for a2 in as2:
  95                 if self.compare_actions(a1, a2):
  96                     found = True
  97
  98             if not found:
  99                 raise ValueError('%s not found in %s' % (a1, as2))
 100                 return False
 101
 102         return True
 103
 104     def compare_actions(self, a1, a2):
 105         for key, val in a1.items():
 106             if a2.get(key, None) != val:
 107                 return False
 108         return True
 109
 110
 111 class SubscriptionAPITests(unittest.TestCase):
 112     """ Tests the Subscription API """
 113
 114     def setUp(self):
 115         self.password = 'asdf'
 116         self.username = 'subscription-api-user'
 117         self.device_uid = 'test-device'
 118         self.user = User(username=self.username, email='user@example.com')
 119         self.user.set_password(self.password)
 120         self.user.save()
 121         self.user.is_active = True
 122         self.client = Client()
 123
 124         self.extra = {
 125             'HTTP_AUTHORIZATION': create_auth_string(self.username,
 126                                                      self.password)
 127         }
 128
 129         self.action_data = {
 130             'add': ['http://example.com/podcast.rss'],
 131         }
 132
 133         self.url = reverse('subscriptions-api', kwargs={
 134             'version': '2',
 135             'username': self.user.username,
 136             'device_uid': self.device_uid,
 137         })
 138
 139     def tearDown(self):
 140         self.user.delete()
 141
 142     def test_set_get_subscriptions(self):
 143         """ Tests that an upload subscription is returned back correctly """
 144
 145         # upload a subscription
 146         response = self.client.post(self.url, json.dumps(self.action_data),
 147                                     content_type="application/json",
 148                                     **self.extra)
 149         self.assertEqual(response.status_code, 200, response.content)
 150
 151         # verify that the subscription is returned correctly
 152         response = self.client.get(self.url, {'since': '0'}, **self.extra)
 153         self.assertEqual(response.status_code, 200, response.content)
 154         response_obj = json.loads(response.content)
 155         self.assertEqual(self.action_data['add'], response_obj['add'])
 156         self.assertEqual([], response_obj.get('remove', []))
 157
 158     def test_unauth_request(self):
 159         """ Tests that an unauthenticated request gives a 401 response """
 160         response = self.client.get(self.url, {'since': '0'})
 161         self.assertEqual(response.status_code, 401, response.content)
 162
 163
 164 class DirectoryTest(TestCase):
 165     """ Test Directory API """
 166
 167     def setUp(self):
 168         self.podcast = Podcast.objects.get_or_create_for_url(
 169             'http://example.com/directory-podcast.xml',
 170             defaults = {
 171                 'title': 'My Podcast',
 172             },
 173         )
 174         self.episode = Episode.objects.get_or_create_for_url(
 175             self.podcast,
 176             'http://example.com/directory-podcast/1.mp3',
 177             defaults = {
 178                 'title': 'My Episode',
 179             },
 180         )
 181
 182     def test_episode_info(self):
 183         """ Test that the expected number of queries is executed """
 184         url = reverse('api-episode-info') + '?' + urlencode(
 185             (('podcast', self.podcast.url), ('url', self.episode.url)))
 186
 187         with self.assertNumQueries(4):
 188             resp = anon_request(url)
 189
 190         self.assertEqual(resp.status_code, 200)
 191
 192
 193 def suite():
 194     tl = unittest.TestLoader()
 195     suite = unittest.TestSuite()
 196     suite.addTest(tl.loadTestsFromTestCase(AdvancedAPITests))
 197     suite.addTest(tl.loadTestsFromTestCase(SubscriptionAPITests))
 198     suite.addTest(tl.loadTestsFromTestCase(DirectoryTest))
 199     return suite