mygpo/api/tests.py

   1 import json
   2 import uuid
   3 import copy
   4 import unittest
   5 from datetime import datetime, timedelta
   6 from urllib.parse import urlencode
   7
   8 from django.test.client import Client
   9 from django.test import TestCase
  10 from django.urls import reverse
  11 from django.contrib.auth import get_user_model
  12 from django.test.utils import override_settings
  13
  14 from mygpo.podcasts.models import Podcast, Episode
  15 from mygpo.api.advanced import episodes
  16 from mygpo.history.models import EpisodeHistoryEntry
  17 from mygpo.test import create_auth_string, anon_request
  18 from mygpo.utils import get_timestamp
  19
  20
  21 class AdvancedAPITests(unittest.TestCase):
  22
  23     def setUp(self):
  24         User = get_user_model()
  25         self.password = 'asdf'
  26         self.username = 'adv-api-user'
  27         self.user = User(username=self.username, email='user@example.com')
  28         self.user.set_password(self.password)
  29         self.user.save()
  30         self.user.is_active = True
  31         self.client = Client()
  32
  33         self.extra = {
  34             'HTTP_AUTHORIZATION': create_auth_string(self.username,
  35                                                      self.password)
  36         }
  37
  38         self.action_data = [
  39             {
  40                 "podcast": "http://example.com/feed.rss",
  41                 "episode": "http://example.com/files/s01e20.mp3",
  42                 "device": "gpodder_abcdef123",
  43                 "action": "download",
  44                 "timestamp": "2009-12-12T09:00:00"
  45             },
  46             {
  47                 "podcast": "http://example.org/podcast.php",
  48                 "episode": "http://ftp.example.org/foo.ogg",
  49                 "action": "play",
  50                 "started": 15,
  51                 "position": 120,
  52                 "total":  500
  53             }
  54         ]
  55
  56     def tearDown(self):
  57         self.user.delete()
  58
  59     def test_episode_actions(self):
  60         response = self._upload_episode_actions(self.user, self.action_data,
  61                                                 self.extra)
  62         self.assertEqual(response.status_code, 200, response.content)
  63
  64         url = reverse(episodes, kwargs={
  65             'version': '2',
  66             'username': self.user.username,
  67         })
  68         response = self.client.get(url, {'since': '0'}, **self.extra)
  69         self.assertEqual(response.status_code, 200, response.content)
  70         response_obj = json.loads(response.content.decode('utf-8'))
  71         actions = response_obj['actions']
  72         self.assertTrue(self.compare_action_list(self.action_data, actions))
  73
  74     def test_invalid_client_id(self):
  75         """ Invalid Client ID should return 400 """
  76         action_data = copy.deepcopy(self.action_data)
  77         action_data[0]['device'] = "gpodder@abcdef123"
  78
  79         response = self._upload_episode_actions(self.user, action_data,
  80                                                 self.extra)
  81
  82         self.assertEqual(response.status_code, 400, response.content)
  83
  84     def _upload_episode_actions(self, user, action_data, extra):
  85         url = reverse(episodes, kwargs={
  86             'version': '2',
  87             'username': self.user.username,
  88         })
  89         return self.client.post(url, json.dumps(action_data),
  90                                 content_type="application/json",
  91                                 **extra)
  92
  93     def compare_action_list(self, as1, as2):
  94         for a1 in as1:
  95             found = False
  96             for a2 in as2:
  97                 if self.compare_actions(a1, a2):
  98                     found = True
  99
 100             if not found:
 101                 raise ValueError('%s not found in %s' % (a1, as2))
 102                 return False
 103
 104         return True
 105
 106     def compare_actions(self, a1, a2):
 107         for key, val in a1.items():
 108             if a2.get(key, None) != val:
 109                 return False
 110         return True
 111
 112
 113 class SubscriptionAPITests(unittest.TestCase):
 114     """ Tests the Subscription API """
 115
 116     def setUp(self):
 117         User = get_user_model()
 118         self.password = 'asdf'
 119         self.username = 'subscription-api-user'
 120         self.device_uid = 'test-device'
 121         self.user = User(username=self.username, email='user@example.com')
 122         self.user.set_password(self.password)
 123         self.user.save()
 124         self.user.is_active = True
 125         self.client = Client()
 126
 127         self.extra = {
 128             'HTTP_AUTHORIZATION': create_auth_string(self.username,
 129                                                      self.password)
 130         }
 131
 132         self.action_data = {
 133             'add': ['http://example.com/podcast.rss'],
 134         }
 135
 136         self.url = reverse('subscriptions-api', kwargs={
 137             'version': '2',
 138             'username': self.user.username,
 139             'device_uid': self.device_uid,
 140         })
 141
 142     def tearDown(self):
 143         self.user.delete()
 144
 145     def test_set_get_subscriptions(self):
 146         """ Tests that an upload subscription is returned back correctly """
 147
 148         # upload a subscription
 149         response = self.client.post(self.url, json.dumps(self.action_data),
 150                                     content_type="application/json",
 151                                     **self.extra)
 152         self.assertEqual(response.status_code, 200, response.content)
 153
 154         # verify that the subscription is returned correctly
 155         response = self.client.get(self.url, {'since': '0'}, **self.extra)
 156         self.assertEqual(response.status_code, 200, response.content)
 157         response_obj = json.loads(response.content.decode('utf-8'))
 158         self.assertEqual(self.action_data['add'], response_obj['add'])
 159         self.assertEqual([], response_obj.get('remove', []))
 160
 161     def test_unauth_request(self):
 162         """ Tests that an unauthenticated request gives a 401 response """
 163         response = self.client.get(self.url, {'since': '0'})
 164         self.assertEqual(response.status_code, 401, response.content)
 165
 166
 167 class DirectoryTest(TestCase):
 168     """ Test Directory API """
 169
 170     def setUp(self):
 171         self.podcast = Podcast.objects.get_or_create_for_url(
 172             'http://example.com/directory-podcast.xml',
 173             defaults = {
 174                 'title': 'My Podcast',
 175             },
 176         ).object
 177         self.episode = Episode.objects.get_or_create_for_url(
 178             self.podcast,
 179             'http://example.com/directory-podcast/1.mp3',
 180             defaults = {
 181                 'title': 'My Episode',
 182             },
 183         ).object
 184         self.client = Client()
 185
 186     def test_episode_info(self):
 187         """ Test that the expected number of queries is executed """
 188         url = reverse('api-episode-info') + '?' + urlencode(
 189             (('podcast', self.podcast.url), ('url', self.episode.url)))
 190
 191         resp = self.client.get(url)
 192
 193         self.assertEqual(resp.status_code, 200)
 194
 195
 196 class EpisodeActionTests(TestCase):
 197
 198     def setUp(self):
 199         self.podcast = Podcast.objects.get_or_create_for_url(
 200             'http://example.com/directory-podcast.xml',
 201             defaults = {
 202                 'title': 'My Podcast',
 203             },
 204         ).object
 205         self.episode = Episode.objects.get_or_create_for_url(
 206             self.podcast,
 207             'http://example.com/directory-podcast/1.mp3',
 208             defaults = {
 209                 'title': 'My Episode',
 210             },
 211         ).object
 212         User = get_user_model()
 213         self.password = 'asdf'
 214         self.username = 'adv-api-user'
 215         self.user = User(username=self.username, email='user@example.com')
 216         self.user.set_password(self.password)
 217         self.user.save()
 218         self.user.is_active = True
 219         self.client = Client()
 220         self.extra = {
 221             'HTTP_AUTHORIZATION': create_auth_string(self.username,
 222                                                      self.password)
 223         }
 224
 225     def tearDown(self):
 226         self.episode.delete()
 227         self.podcast.delete()
 228         self.user.delete()
 229
 230     @override_settings(MAX_EPISODE_ACTIONS=10)
 231     def test_limit_actions(self):
 232         """ Test that max MAX_EPISODE_ACTIONS episodes are returned """
 233
 234         timestamps = []
 235         t = datetime.utcnow()
 236         for n in range(15):
 237             timestamp = t - timedelta(seconds=n)
 238             EpisodeHistoryEntry.objects.create(
 239                 timestamp = timestamp,
 240                 episode = self.episode,
 241                 user = self.user,
 242                 action = EpisodeHistoryEntry.DOWNLOAD,
 243             )
 244             timestamps.append(timestamp)
 245
 246         url = reverse(episodes, kwargs={
 247             'version': '2',
 248             'username': self.user.username,
 249         })
 250         response = self.client.get(url, {'since': '0'}, **self.extra)
 251         self.assertEqual(response.status_code, 200, response.content)
 252         response_obj = json.loads(response.content.decode('utf-8'))
 253         actions = response_obj['actions']
 254
 255         # 10 actions should be returned
 256         self.assertEqual(len(actions), 10)
 257
 258         timestamps = sorted(timestamps)
 259
 260         # the first 10 actions, according to their timestamp should be returned
 261         for action, timestamp in zip(actions, timestamps):
 262             self.assertEqual(timestamp.isoformat(), action['timestamp'])
 263
 264         # the `timestamp` field in the response should be the timestamp of the
 265         # last returned action
 266         self.assertEqual(
 267             get_timestamp(timestamps[9]),
 268             response_obj['timestamp']
 269         )
 270
 271
 272     def test_no_actions(self):
 273         """ Test when there are no actions to return """
 274
 275         t1 = get_timestamp(datetime.utcnow())
 276
 277         url = reverse(episodes, kwargs={
 278             'version': '2',
 279             'username': self.user.username,
 280         })
 281         response = self.client.get(url, {'since': '0'}, **self.extra)
 282         self.assertEqual(response.status_code, 200, response.content)
 283         response_obj = json.loads(response.content.decode('utf-8'))
 284         actions = response_obj['actions']
 285
 286         # 10 actions should be returned
 287         self.assertEqual(len(actions), 0)
 288
 289         returned = response_obj['timestamp']
 290         t2 = get_timestamp(datetime.utcnow())
 291         # the `timestamp` field in the response should be the timestamp of the
 292         # last returned action
 293         self.assertGreaterEqual(returned, t1)
 294         self.assertGreaterEqual(t2, returned)