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