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