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.db.couchdb.podcast_state import (subscribed_podcast_ids_by_user_id,
  33     subscribe, unsubscribe, )
  34
  35
  36 class DeviceSyncTests(unittest.TestCase):
  37
  38     def setUp(self):
  39         self.user = UserProxy(username='test')
  40         self.user.email = 'test@invalid.com'
  41         self.user.set_password('secret!')
  42         self.user.save()
  43
  44
  45     def test_group(self):
  46         dev1 = Client.objects.create(id=uuid.uuid1(), user=self.user, uid='d1')
  47         dev2 = Client.objects.create(id=uuid.uuid1(), user=self.user, uid='d2')
  48
  49         group = self.user.get_grouped_devices().next()
  50         self.assertEquals(group.is_synced, False)
  51         self.assertIn(dev1, group.devices)
  52         self.assertIn(dev2, group.devices)
  53
  54
  55         dev3 = Client.objects.create(id=uuid.uuid1(), user=self.user, uid='d3')
  56
  57         dev1.sync_with(dev3)
  58
  59         groups = self.user.get_grouped_devices()
  60
  61         g2 = groups.next()
  62         self.assertEquals(g2.is_synced, False)
  63         self.assertIn(dev2, g2.devices)
  64
  65         g1 = groups.next()
  66         self.assertEquals(g1.is_synced, True)
  67         self.assertIn(dev1, g1.devices)
  68         self.assertIn(dev3, g1.devices)
  69
  70         targets = dev1.get_sync_targets()
  71         target = targets.next()
  72         self.assertEquals(target, dev2)
  73
  74     def tearDown(self):
  75         Client.objects.filter(user=self.user).delete()
  76         self.user.delete()
  77
  78
  79 @override_settings(CACHE={})
  80 class UnsubscribeMergeTests(TestCase):
  81     """ Test if merged podcasts can be properly unsubscribed
  82
  83     TODO: this test fails intermittently """
  84
  85     P2_URL = 'http://test.org/podcast/'
  86
  87     def setUp(self):
  88         self.podcast1 = Podcast.objects.get_or_create_for_url('http://example.com/feed.rss')
  89         self.podcast2 = Podcast.objects.get_or_create_for_url(self.P2_URL)
  90
  91         User = get_user_model()
  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         subscribe(self.podcast2, 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.objects.get(urls__url=self.P2_URL)
 114         unsubscribe(p, self.user, self.device)
 115
 116         subscriptions = subscribed_podcast_ids_by_user_id(self.user.profile.uuid.hex)
 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