mygpo/users/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 import unittest
  19 import doctest
  20 from collections import Counter
  21
  22 from django.test import TestCase
  23 from django.test.utils import override_settings
  24
  25 import mygpo.utils
  26 from mygpo.core.models import Podcast
  27 from mygpo.maintenance.merge import PodcastMerger
  28 from mygpo.api.backend import get_device
  29 from mygpo.users.models import User, Device
  30 from mygpo.db.couchdb.podcast import podcast_for_url
  31 from mygpo.db.couchdb.podcast_state import subscribed_podcast_ids_by_user_id
  32
  33
  34 class DeviceSyncTests(unittest.TestCase):
  35
  36     def setUp(self):
  37         self.user = User(username='test')
  38         self.user.email = 'test@invalid.com'
  39         self.user.set_password('secret!')
  40         self.user.save()
  41
  42
  43     def test_group(self):
  44         dev1 = Device(uid='d1')
  45         self.user.devices.append(dev1)
  46
  47         dev2 = Device(uid='d2')
  48         self.user.devices.append(dev2)
  49
  50         group = self.user.get_grouped_devices().next()
  51         self.assertEquals(group.is_synced, False)
  52         self.assertIn(dev1, group.devices)
  53         self.assertIn(dev2, group.devices)
  54
  55
  56         dev3 = Device(uid='d3')
  57         self.user.devices.append(dev3)
  58
  59         self.user.sync_devices(dev1, dev3)
  60
  61         groups = self.user.get_grouped_devices()
  62         g1 = groups.next()
  63
  64         self.assertEquals(g1.is_synced, True)
  65         self.assertIn(dev1, g1.devices)
  66         self.assertIn(dev3, g1.devices)
  67
  68         g2 = groups.next()
  69         self.assertEquals(g2.is_synced, False)
  70         self.assertIn(dev2, g2.devices)
  71
  72
  73         targets = self.user.get_sync_targets(dev1)
  74         target = targets.next()
  75         self.assertEquals(target, dev2)
  76
  77
  78 @override_settings(CACHE={})
  79 class UnsubscribeMergeTests(TestCase):
  80     """ Test if merged podcasts can be properly unsubscribed
  81
  82     TODO: this test fails intermittently """
  83
  84     P2_URL = 'http://test.org/podcast/'
  85
  86     def setUp(self):
  87         self.podcast1 = Podcast(urls=['http://example.com/feed.rss'])
  88         self.podcast2 = Podcast(urls=[self.P2_URL])
  89         self.podcast1.save()
  90         self.podcast2.save()
  91
  92         self.user = User(username='test-merge')
  93         self.user.email = 'test@example.com'
  94         self.user.set_password('secret!')
  95         self.user.save()
  96
  97         self.device = get_device(self.user, 'dev', '')
  98
  99     def test_merge_podcasts(self):
 100         self.podcast2.subscribe(self.user, self.device)
 101
 102         # merge podcast2 into podcast1
 103         pm = PodcastMerger([self.podcast1, self.podcast2], Counter(), [])
 104         pm.merge()
 105
 106         # seems that setting delayed_commit = false in the CouchDB config, as
 107         # well as a delay here fix the intermittent failures.
 108         # TODO: further investiation needed
 109         import time
 110         time.sleep(2)
 111
 112         # get podcast for URL of podcast2 and unsubscribe from it
 113         p = podcast_for_url(self.P2_URL)
 114         p.unsubscribe(self.user, self.device)
 115
 116         subscriptions = subscribed_podcast_ids_by_user_id(self.user._id)
 117         self.assertEqual(0, len(subscriptions))
 118
 119     def tearDown(self):
 120         self.podcast1.delete()
 121         self.user.delete()
 122
 123
 124 def suite():
 125     suite = unittest.TestSuite()
 126     suite.addTest(unittest.TestLoader().loadTestsFromTestCase(DeviceSyncTests))
 127     suite.addTest(unittest.TestLoader().loadTestsFromTestCase(UnsubscribeMergeTests))
 128     return suite