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 from django.contrib.auth import get_user_model
  29
  30 from mygpo.podcasts.models import Podcast, Episode
  31 from mygpo.api.advanced import episodes
  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         User = get_user_model()
  40         self.password = 'asdf'
  41         self.username = 'adv-api-user'
  42         self.user = User(username=self.username, email='user@example.com')
  43         self.user.set_password(self.password)
  44         self.user.save()
  45         self.user.is_active = True
  46         self.client = Client()
  47
  48         self.extra = {
  49             'HTTP_AUTHORIZATION': create_auth_string(self.username,
  50                                                      self.password)
  51         }
  52
  53         self.action_data = [
  54             {
  55                 "podcast": "http://example.com/feed.rss",
  56                 "episode": "http://example.com/files/s01e20.mp3",
  57                 "device": "gpodder_abcdef123",
  58                 "action": "download",
  59                 "timestamp": "2009-12-12T09:00:00"
  60             },
  61             {
  62                 "podcast": "http://example.org/podcast.php",
  63                 "episode": "http://ftp.example.org/foo.ogg",
  64                 "action": "play",
  65                 "started": 15,
  66                 "position": 120,
  67                 "total":  500
  68             }
  69         ]
  70
  71     def tearDown(self):
  72         self.user.delete()
  73
  74     def test_episode_actions(self):
  75         url = reverse(episodes, kwargs={
  76             'version': '2',
  77             'username': self.user.username,
  78         })
  79
  80         # upload actions
  81         response = self.client.post(url, json.dumps(self.action_data),
  82                                     content_type="application/json",
  83                                     **self.extra)
  84         self.assertEqual(response.status_code, 200, response.content)
  85
  86         response = self.client.get(url, {'since': '0'}, **self.extra)
  87         self.assertEqual(response.status_code, 200, response.content)
  88         response_obj = json.loads(response.content)
  89         actions = response_obj['actions']
  90         self.assertTrue(self.compare_action_list(self.action_data, actions))
  91
  92     def compare_action_list(self, as1, as2):
  93         for a1 in as1:
  94             found = False
  95             for a2 in as2:
  96                 if self.compare_actions(a1, a2):
  97                     found = True
  98
  99             if not found:
 100                 raise ValueError('%s not found in %s' % (a1, as2))
 101                 return False
 102
 103         return True
 104
 105     def compare_actions(self, a1, a2):
 106         for key, val in a1.items():
 107             if a2.get(key, None) != val:
 108                 return False
 109         return True
 110
 111
 112 class SubscriptionAPITests(unittest.TestCase):
 113     """ Tests the Subscription API """
 114
 115     def setUp(self):
 116         User = get_user_model()
 117         self.password = 'asdf'
 118         self.username = 'subscription-api-user'
 119         self.device_uid = 'test-device'
 120         self.user = User(username=self.username, email='user@example.com')
 121         self.user.set_password(self.password)
 122         self.user.save()
 123         self.user.is_active = True
 124         self.client = Client()
 125
 126         self.extra = {
 127             'HTTP_AUTHORIZATION': create_auth_string(self.username,
 128                                                      self.password)
 129         }
 130
 131         self.action_data = {
 132             'add': ['http://example.com/podcast.rss'],
 133         }
 134
 135         self.url = reverse('subscriptions-api', kwargs={
 136             'version': '2',
 137             'username': self.user.username,
 138             'device_uid': self.device_uid,
 139         })
 140
 141     def tearDown(self):
 142         self.user.delete()
 143
 144     def test_set_get_subscriptions(self):
 145         """ Tests that an upload subscription is returned back correctly """
 146
 147         # upload a subscription
 148         response = self.client.post(self.url, json.dumps(self.action_data),
 149                                     content_type="application/json",
 150                                     **self.extra)
 151         self.assertEqual(response.status_code, 200, response.content)
 152
 153         # verify that the subscription is returned correctly
 154         response = self.client.get(self.url, {'since': '0'}, **self.extra)
 155         self.assertEqual(response.status_code, 200, response.content)
 156         response_obj = json.loads(response.content)
 157         self.assertEqual(self.action_data['add'], response_obj['add'])
 158         self.assertEqual([], response_obj.get('remove', []))
 159
 160     def test_unauth_request(self):
 161         """ Tests that an unauthenticated request gives a 401 response """
 162         response = self.client.get(self.url, {'since': '0'})
 163         self.assertEqual(response.status_code, 401, response.content)
 164
 165
 166 class DirectoryTest(TestCase):
 167     """ Test Directory API """
 168
 169     def setUp(self):
 170         self.podcast = Podcast.objects.get_or_create_for_url(
 171             'http://example.com/directory-podcast.xml',
 172             defaults = {
 173                 'title': 'My Podcast',
 174             },
 175         )
 176         self.episode = Episode.objects.get_or_create_for_url(
 177             self.podcast,
 178             'http://example.com/directory-podcast/1.mp3',
 179             defaults = {
 180                 'title': 'My Episode',
 181             },
 182         )
 183
 184     def test_episode_info(self):
 185         """ Test that the expected number of queries is executed """
 186         url = reverse('api-episode-info') + '?' + urlencode(
 187             (('podcast', self.podcast.url), ('url', self.episode.url)))
 188
 189         with self.assertNumQueries(4):
 190             resp = anon_request(url)
 191
 192         self.assertEqual(resp.status_code, 200)
 193
 194
 195 def load_tests(loader, tests, ignore):
 196     tl = unittest.TestLoader()
 197     tests.addTest(tl.loadTestsFromTestCase(AdvancedAPITests))
 198     tests.addTest(tl.loadTestsFromTestCase(SubscriptionAPITests))
 199     tests.addTest(tl.loadTestsFromTestCase(DirectoryTest))
 200     return tests