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 django.test import TestCase
  19 from django.test.client import Client
  20 from django.contrib.auth.models import User
  21 from mygpo.api.models import Device, Podcast, Subscription, SubscriptionAction, UserProfile, EpisodeAction
  22 from django.http import HttpRequest
  23 from mygpo.api.simple import subscriptions
  24 from mygpo.api.advanced import devices
  25 import time
  26
  27 try:
  28     #try to import the JSON module (if we are on Python 2.6)
  29     import json
  30 except ImportError:
  31     # No JSON module available - fallback to simplejson (Python < 2.6)
  32     print "No JSON module available - fallback to simplejson (Python < 2.6)"
  33     import simplejson as json
  34
  35
  36 class SyncTest(TestCase):
  37     # FIXME: Broken testcase - please fix!
  38     def tXest_sync_actions(self):
  39         # this test does not yet complete when run as a unit test
  40         # because django does not set up the views
  41         u  = User.objects.create(username='u')
  42         d1 = Device.objects.create(name='d1', type='other', user=u)
  43         d2 = Device.objects.create(name='d2', type='other', user=u)
  44         p1 = Podcast.objects.create(title='p1', url='http://p1.com/')
  45         p2 = Podcast.objects.create(title='p2', url='http://p2.com/')
  46         p3 = Podcast.objects.create(title='p3', url='http://p3.com/')
  47         p4 = Podcast.objects.create(title='p4', url='http://p4.com/')
  48
  49         s1 = SubscriptionAction.objects.create(device=d1, podcast=p1, action='1')
  50         time.sleep(1)
  51         s2 = SubscriptionAction.objects.create(device=d2, podcast=p2, action='1')
  52         time.sleep(1)
  53         u2 = SubscriptionAction.objects.create(device=d2, podcast=p2, action='-1')
  54         time.sleep(1)
  55         s3 = SubscriptionAction.objects.create(device=d1, podcast=p3, action='1')
  56         time.sleep(1)
  57         s3_= SubscriptionAction.objects.create(device=d2, podcast=p3, action='1')
  58         time.sleep(1)
  59         s4 = SubscriptionAction.objects.create(device=d2, podcast=p4, action='1')
  60         time.sleep(1)
  61         u3 = SubscriptionAction.objects.create(device=d2, podcast=p3, action='-1')
  62
  63         #d1: p1, p3
  64         #d2: p2, -p2, p3, p4, -p3
  65
  66         d1.sync_with(d2)
  67
  68         #d1: -p3, +p4
  69         #d2: +p1
  70
  71         sa1 = d1.get_sync_actions()
  72         sa2 = d2.get_sync_actions()
  73
  74         self.assertEqual( len(sa1), 2)
  75         self.assertEqual( len(sa2), 1)
  76
  77         self.assertEqual( sa1[p4].device, d2)
  78         self.assertEqual( sa1[p4].podcast, p4)
  79         self.assertEqual( sa1[p4].action, 1)
  80
  81         self.assertEqual( sa1[p3].device, d2)
  82         self.assertEqual( sa1[p3].podcast, p3)
  83         self.assertEqual( sa1[p3].action, -1)
  84
  85         self.assertEqual( sa2[p1].device, d1)
  86         self.assertEqual( sa2[p1].podcast, p1)
  87         self.assertEqual( sa2[p1].action, 1)
  88
  89 class AdvancedAPITest(TestCase):
  90     def setUp(self):
  91         self.user = User.objects.create(username='adv', email='adv@mygpo')
  92         self.user.set_password('adv1')
  93         self.user.save()
  94         self.device1 = Device.objects.create(user=self.user, name='d1', uid='uid1', type='desktop')
  95         self.device2 = Device.objects.create(user=self.user, name='d2', uid='uid2', type='mobile')
  96         self.url1 = 'http://example.com/feed.rss'
  97         self.url2 = 'http://example.org/podcast.php'
  98         self.url3 = 'http://example.net/foo.xml'
  99
 100         self.client = Client()
 101         l = self.client.login(username='adv', password='adv1')
 102         self.assertEqual(l, True)
 103
 104     def test_device_list(self):
 105         response = self.client.get('/api/1/devices/%s.json' % self.user.username)
 106         json_list = response.content
 107         list = json.loads(json_list)
 108
 109         self.assertEqual(response.status_code, 200)
 110         self.assertEqual(len(list), 2)
 111         self.assertEqual(list[0], {u'id': u'uid1', u'caption': u'd1', u'type': u'desktop', u'subscriptions': 0})
 112         self.assertEqual(list[1], {u'id': u'uid2', u'caption': u'd2', u'type': u'mobile',  u'subscriptions': 0})
 113
 114     def test_rename_device(self):
 115         req = {'caption': 'd2!', 'type': 'server'}
 116         reqs = json.dumps(req)
 117
 118         self.client.post('/api/1/devices/%s/%s.json' % (self.user.username, self.device2.uid), data={'data': reqs})
 119
 120         response = self.client.get('/api/1/devices/%s.json' % self.user.username)
 121         json_list = response.content
 122         list = json.loads(json_list)
 123         self.assertEqual(response.status_code, 200)
 124         self.assertEqual(len(list), 2)
 125         self.assertEqual(list[0], {u'id': u'uid1', u'caption': u'd1',  u'type': u'desktop', u'subscriptions': 0})
 126         self.assertEqual(list[1], {u'id': u'uid2', u'caption': u'd2!', u'type': u'server',  u'subscriptions': 0})
 127
 128
 129     def test_add_remove_subscriptions(self):
 130         req = {"add": [self.url1, self.url2, self.url3]}
 131         reqs = json.dumps(req)
 132
 133         #adding 3 subscriptions
 134         response = self.client.post('/api/1/subscriptions/%s/%s.json' % (self.user.username, self.device1.uid), data={'data': reqs})
 135         self.assertEqual(response.status_code, 200)
 136
 137         resp = json.loads(response.content)
 138         add_timestamp = resp['timestamp']
 139
 140         #qerying subscription changes
 141         response = self.client.get('/api/1/subscriptions/%s/%s.json' % (self.user.username, self.device1.uid), {'since': add_timestamp-1})
 142         changes = json.loads(response.content)
 143
 144         self.assertEqual(changes['add'], [self.url1, self.url2, self.url3])
 145         self.assertEqual(len(changes['remove']), 0)
 146
 147         #removing the 1st and 3rd subscription
 148         req = {"add": [], 'remove': [self.url1, self.url3]}
 149         reqs = json.dumps(req)
 150         time.sleep(1)
 151         response = self.client.post('/api/1/subscriptions/%s/%s.json' % (self.user.username, self.device1.uid), data={'data': reqs})
 152         self.assertEqual(response.status_code, 200)
 153
 154         resp = json.loads(response.content)
 155         timestamp = resp['timestamp']
 156
 157         #changes since beginning, should return 1 addition, 2 removals
 158         response = self.client.get('/api/1/subscriptions/%s/%s.json' % (self.user.username, self.device1.uid), {'since': add_timestamp-1})
 159         changes = json.loads(response.content)
 160         self.assertEqual(changes['add'], [self.url2])
 161         self.assertEqual(changes['remove'], [self.url1, self.url3])
 162
 163
 164     def test_episode_update(self):
 165         req =  [{"podcast": self.url1,
 166                  "episode": "http://example.com/files/s01e20.mp3",
 167                  "device": self.device1.uid,
 168                  "action": "download",
 169                  "timestamp": "2009-12-12T09:00:00"},
 170                 {"podcast": self.url2,
 171                  "episode": "http://ftp.example.org/foo.ogg",
 172                  "action": "play",
 173                  "position": "01:00:00"}]
 174         reqs = json.dumps(req)
 175         response = self.client.post('/api/1/episodes/%s.json' % self.user.username, data={'data': reqs})
 176         self.assertEqual(response.status_code, 200)
 177
 178         resp = json.loads(response.content)
 179         timestamp = resp['timestamp']
 180
 181         response = self.client.get('/api/1/episodes/%s.json' % self.user.username)
 182         changes = json.loads(response.content)
 183         self.assertEqual(len(changes['actions']), 2)
 184         self.assertTrue(self.hash_subset({u'action': u'download', u'podcast': self.url1, u'episode': u'http://example.com/files/s01e20.mp3'}, changes['actions'][0]))
 185         self.assertTrue(self.hash_subset({u'action': u'play', u'podcast': self.url2, u'episode': u'http://ftp.example.org/foo.ogg'}, changes['actions'][1]))
 186
 187 #        #currently fails, seems to be a bug in the test client
 188 #        response = self.client.get('/api/1/episodes/%s.json' % self.user.username, {'podcast': self.url2})
 189 #        self.assertEqual(response.status_code, 200)
 190 #        changes = json.loads(response.content)
 191 #        self.assertEqual(len(changes['actions']), 1)
 192 #        self.assertTrue(self.hash_subset({u'action': u'play', u'podcast': self.url2, u'episode': u'http://ftp.example.org/foo.ogg'}, changes['actions'][0]))
 193
 194
 195     def hash_subset(self, hash1, hash2):
 196         for x in hash1.keys():
 197             if not x in hash2:
 198                 return False
 199             if hash2[x] != hash1[x]:
 200                 return False
 201         return True
 202
 203
 204 class SimpleAPITest(TestCase):
 205
 206     def setUp(self):
 207         un = 'u'
 208         pw = 'u'
 209         self.user = User.objects.create(username=un, email='u@mygpo')
 210         self.user.set_password(pw)
 211         self.user.save()
 212         UserProfile.objects.create(user=self.user)
 213
 214         r = HttpRequest()
 215         r.method = 'PUT'
 216         r.user = self.user
 217
 218         self.req = r
 219         self.p1 = 'http://www.podcast1.com'
 220         self.p2 = 'http://www.podcast2.com'
 221         self.p3 = 'http://www.podcast3.com'
 222
 223         self.client = Client()
 224         login = self.client.login(username=un, password=pw)
 225         self.assertEqual(login, True)
 226
 227     def test_put_get_txt(self):
 228         d1 = '1'
 229         f1 = 'txt'
 230
 231         data_txt_1 = '%s\n%s\n' % (self.p1, self.p2)
 232         data_txt_2 = '%s\n%s\n' % (self.p2, self.p3)
 233
 234         #1. put 2 new podcasts
 235         self.req.raw_post_data = data_txt_1
 236         put = subscriptions(request=self.req, username=self.user.username, device_uid=d1, format=f1)
 237         # Successful requests should return the empty string + status code 200
 238         self.assertEqual(put.status_code, 200)
 239         self.assertEqual(put.content, '')
 240
 241         #device 1 txt
 242         device = Device.objects.get(uid=d1, user=self.user)
 243
 244         #get subscriptions
 245         s = [p.podcast for p in device.get_subscriptions()]
 246         urls = [p.url for p in s]
 247         self.assertEqual(len(urls), 2)
 248         self.assertEqual(urls[0], self.p1)
 249         self.assertEqual(urls[1], self.p2)
 250
 251         #2. put 1 new podcast and delete 1 old
 252         time.sleep(1)
 253         self.req.raw_post_data = data_txt_2
 254         subscriptions(request=self.req, username=self.user.username, device_uid=d1, format=f1)
 255
 256         #get subscriptions
 257         s = [p.podcast for p in device.get_subscriptions()]
 258         urls = [p.url for p in s]
 259         self.assertEqual( len(s), 2)
 260         self.assertEqual(urls[0], self.p2)
 261         self.assertEqual(urls[1], self.p3)
 262
 263     def test_put_get_json(self):
 264         d2 = '2'
 265         f2 = 'json'
 266
 267         data_json_1 = '[\n"%s",\n"%s"\n]' % (self.p1, self.p2)
 268         data_json_2 = '[\n"%s",\n"%s"\n]' % (self.p2, self.p3)
 269
 270         #1. put 2 new podcasts
 271         self.req.raw_post_data = data_json_1
 272         subscriptions(request=self.req, username=self.user.username, device_uid=d2, format=f2)
 273
 274         #device 2 json
 275         device = Device.objects.get(uid=d2, user=self.user)
 276
 277         #get subscriptions
 278         s = [p.podcast for p in device.get_subscriptions()]
 279         urls = [p.url for p in s]
 280         self.assertEqual( len(s), 2)
 281         self.assertEqual(urls[0], self.p1)
 282         self.assertEqual(urls[1], self.p2)
 283
 284         #2. put 1 new podcast and delete 1 old
 285         time.sleep(1)
 286         self.req.raw_post_data = data_json_2
 287         subscriptions(request=self.req, username=self.user.username, device_uid=d2, format=f2)
 288
 289         #get subscriptions
 290         s = [p.podcast for p in device.get_subscriptions()]
 291         urls = [p.url for p in s]
 292         self.assertEqual( len(s), 2)
 293         self.assertEqual(urls[0], self.p2)
 294         self.assertEqual(urls[1], self.p3)
 295
 296     def test_put_get_opml(self):
 297         d3 = '3'
 298         f3 = 'opml'
 299
 300         data_opml_1 = '<?xml version="1.0" encoding="UTF-8"?>\n<opml version="2.0">\n<head>\n<title>subscription list</title>\n\
 301                        <dateCreated>Tue, 01 Dec 2009 10:06:18 +0000</dateCreated>\n</head>\n<body>\n\
 302                        <outline text="p1" title="p1" type="rss" xmlUrl="%s"/>\n\
 303                        <outline text="p2" title="p2" type="rss" xmlUrl="%s"/>\n\
 304                        </body>\n</opml>\n' % (self.p1, self.p2)
 305         data_opml_2 = '<?xml version="1.0" encoding="UTF-8"?><opml version="2.0"><head><title>subscription list</title>\
 306                        <dateCreated>Tue, 01 Dec 2009 10:06:18 +0000</dateCreated></head><body>\
 307                        <outline text="p2" title="p2" type="rss" xmlUrl="%s"/>\
 308                        <outline text="p3" title="p3" type="rss" xmlUrl="%s"/>\
 309                        </body></opml>' % (self.p2, self.p3)
 310
 311         #1. put 2 new podcasts
 312         self.req.raw_post_data = data_opml_1
 313         subscriptions(request=self.req, username=self.user.username, device_uid=d3, format=f3)
 314
 315         #device 3 opml
 316         device = Device.objects.get(uid=d3, user=self.user)
 317
 318         #get subscriptions
 319         s = [p.podcast for p in device.get_subscriptions()]
 320         urls = [p.url for p in s]
 321         self.assertEqual( len(s), 2)
 322         self.assertEqual(urls[0], self.p1)
 323         self.assertEqual(urls[1], self.p2)
 324
 325         #2. put 1 new podcast and delete 1 old
 326         time.sleep(1)
 327         self.req.raw_post_data = data_opml_2
 328         subscriptions(request=self.req, username=self.user.username, device_uid=d3, format=f3)
 329
 330         #get subscriptions
 331         s = [p.podcast for p in device.get_subscriptions()]
 332         urls = [p.url for p in s]
 333         self.assertEqual( len(s), 2)
 334         self.assertEqual(urls[0], self.p2)
 335         self.assertEqual(urls[1], self.p3)
 336
 337
 338     def test_put_json_client(self):
 339         device = 'ipod'
 340         urls = [self.p1, self.p2, self.p3]
 341         reqs = json.dumps(urls)
 342
 343         #put json - add 3 podcasts
 344         response = self.client.put('/subscriptions/%s/%s.json' % (self.user.username, device), data={'raw_post_data':reqs})
 345         self.assertEqual(response.status_code, 200)
 346         self.assertEqual(response.content, '')
 347
 348         #get json - get 3 podcasts
 349         response = self.client.get('/subscriptions/%s/%s.json' % (self.user.username, device), {})
 350         get = json.loads(response.content)
 351
 352         self.assertEqual(get, [self.p1, self.p2, self.p3])
 353
 354     def test_put_txt_get_json_client(self):
 355         device = 'mobdev'
 356         reqs = '%s\n%s\n%s\n' % (self.p1, self.p2, self.p3)
 357
 358         #put txt - add 3 podcasts
 359         response = self.client.put('/subscriptions/%s/%s.txt' % (self.user.username, device), data={'raw_post_data':reqs})
 360         self.assertEqual(response.status_code, 200)
 361         self.assertEqual(response.content, '')
 362
 363         #qerying subscription changes
 364         response = self.client.get('/subscriptions/%s/%s.json' % (self.user.username, device), {})
 365         get = json.loads(response.content)
 366
 367         self.assertEqual(get, [self.p1, self.p2, self.p3])
 368
 369     def test_put_opml_get_json_client(self):
 370         device = 'pc'
 371         reqs = '<?xml version="1.0" encoding="UTF-8"?>\n<opml version="2.0">\n<head>\n<title>subscription list</title>\n\
 372                        <dateCreated>Tue, 01 Dec 2009 10:06:18 +0000</dateCreated>\n</head>\n<body>\n\
 373                        <outline text="p1" title="p1" type="rss" xmlUrl="%s"/>\n\
 374                        <outline text="p2" title="p2" type="rss" xmlUrl="%s"/>\n\
 375                        <outline text="p3" title="p3" type="rss" xmlUrl="%s"/>\n\
 376                        </body>\n</opml>\n' % (self.p1, self.p2, self.p3)
 377
 378         #put json - add 3 podcasts
 379         response = self.client.put('/subscriptions/%s/%s.opml' % (self.user.username, device), data={'raw_post_data':reqs})
 380         self.assertEqual(response.status_code, 200)
 381         self.assertEqual(response.content, '')
 382
 383         #qerying subscription changes
 384         response = self.client.get('/subscriptions/%s/%s.json' % (self.user.username, device), {})
 385         get = json.loads(response.content)
 386
 387         self.assertEqual(get, [self.p1, self.p2, self.p3])
 388
 389     def test_put_invalid_format(self):
 390         device = 'ipod'
 391         urls = [self.p1, self.p2, self.p3]
 392         reqs = json.dumps(urls)
 393
 394         #put json - add 3 podcasts
 395         response = self.client.put('/subscriptions/%s/%s.iso' % (self.user.username, device), data={'raw_post_data':reqs})
 396         self.assertEqual(response.status_code, 400)
 397         self.assertEqual(response.content, 'Invalid format')
 398
 399
 400     def test_put_valid_json_get_invalid_format(self):
 401         device = 'ipod'
 402         urls = [self.p1, self.p2, self.p3]
 403         reqs = json.dumps(urls)
 404
 405         #put json - add 3 podcasts
 406         response = self.client.put('/subscriptions/%s/%s.json' % (self.user.username, device), data={'raw_post_data':reqs})
 407         self.assertEqual(response.status_code, 200)
 408         self.assertEqual(response.content, '')
 409
 410         #qerying subscription changes
 411         response = self.client.get('/subscriptions/%s/%s.iso' % (self.user.username, device), {})
 412         self.assertEqual(response.status_code, 400)
 413         self.assertEqual(response.content, 'Invalid format')
 414
 415     def test_get_invalid_device(self):
 416         device = 'invalid'
 417
 418         #qerying subscription changes
 419         response = self.client.get('/subscriptions/%s/%s.txt' % (self.user.username, device), {})
 420         self.assertEqual(response.status_code, 404)
 421         self.assertEqual(response.content, 'Invalid device ID')
 422
 423     def test_get_invalid_user(self):
 424         username = 'invalid'
 425         device = 'invalid'
 426
 427         #qerying subscription changes
 428         response = self.client.get('/subscriptions/%s/%s.txt' % (username, device), {})
 429         self.assertEqual(response.status_code, 401)
 430         self.assertEqual(response.content, 'Invalid user')
 431