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