mygpo/maintenance/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 from datetime import datetime
  20 import unittest
  21 from collections import Counter
  22
  23 from django.test import TestCase, TransactionTestCase
  24 from django.test.utils import override_settings
  25 from django.contrib.auth import get_user_model
  26
  27 from mygpo.podcasts.models import Podcast, Episode
  28 from mygpo.users.models import EpisodeAction
  29 from mygpo.maintenance.merge import PodcastMerger
  30 from mygpo.utils import get_timestamp
  31 from mygpo.db.couchdb.episode_state import episode_state_for_user_episode, \
  32     add_episode_actions
  33
  34
  35 def u():
  36     return uuid.uuid1()
  37
  38
  39 class SimpleMergeTests(TestCase):
  40
  41     def setUp(self):
  42         self.podcast1 = Podcast.objects.get_or_create_for_url(
  43             'http://example.com/simple-merge-test-feed.rss',
  44             defaults={'title': 'Podcast 1'},
  45         )
  46         self.podcast2 = Podcast.objects.get_or_create_for_url(
  47             'http://simple-merge-test.org/podcast/',
  48             defaults={'title': 'Podcast 2'},
  49         )
  50
  51         self.episode1 = Episode.objects.get_or_create_for_url(
  52             self.podcast1, 'http://example.com/simple-merge-test-episode1.mp3',
  53             defaults={
  54                 'title': 'Episode 1 A',
  55             })
  56         self.episode2 = Episode.objects.get_or_create_for_url(
  57             self.podcast2, 'http://example.com/simple-merge-test-episode1.mp3',
  58             defaults={
  59                 'title': 'Episode 1 B',
  60             })
  61
  62     def test_merge_podcasts(self):
  63         merge(self.podcast1, self.podcast2)
  64
  65
  66 @override_settings(CACHE={})
  67 class MergeTests(TransactionTestCase):
  68     """ Tests merging of two podcasts, their episodes and states """
  69
  70     def setUp(self):
  71         self.podcast1 = Podcast.objects.get_or_create_for_url(
  72             'http://example.com/merge-test-feed.rss',
  73             defaults={'title': 'Podcast 1'},
  74         )
  75         self.podcast2 = Podcast.objects.get_or_create_for_url(
  76             'http://merge-test.org/podcast/',
  77             defaults={'title': 'Podcast 2'},
  78         )
  79
  80         self.episode1 = Episode.objects.get_or_create_for_url(
  81             self.podcast1, 'http://example.com/merge-test-episode1.mp3',
  82             defaults={
  83                 'title': 'Episode 1 A',
  84             })
  85         self.episode2 = Episode.objects.get_or_create_for_url(
  86             self.podcast2, 'http://example.com/merge-test-episode1.mp3',
  87             defaults={
  88                 'title': 'Episode 1 B',
  89             })
  90
  91         User = get_user_model()
  92         self.user = User(username='test-merge')
  93         self.user.email = 'test-merge-tests@example.com'
  94         self.user.set_password('secret!')
  95         self.user.save()
  96
  97     def test_merge_podcasts(self):
  98
  99         # Create additional data that will be merged
 100         state1 = episode_state_for_user_episode(self.user, self.episode1)
 101         state2 = episode_state_for_user_episode(self.user, self.episode2)
 102
 103         action1 = EpisodeAction(action='play',
 104                                 timestamp=datetime.utcnow(),
 105                                 upload_timestamp=get_timestamp(
 106                                     datetime.utcnow()))
 107         action2 = EpisodeAction(action='download',
 108                                 timestamp=datetime.utcnow(),
 109                                 upload_timestamp=get_timestamp(
 110                                     datetime.utcnow()))
 111
 112         add_episode_actions(state1, [action1])
 113         add_episode_actions(state2, [action2])
 114
 115         # decide which episodes to merge
 116         groups = [(0, [self.episode1, self.episode2])]
 117         counter = Counter()
 118
 119         pm = PodcastMerger([self.podcast1, self.podcast2], counter, groups)
 120         pm.merge()
 121
 122         state1 = episode_state_for_user_episode(self.user, self.episode1)
 123
 124         # both actions must be present in state1
 125         self.assertIn(action1, state1.actions)
 126         self.assertIn(action2, state1.actions)
 127
 128     def tearDown(self):
 129         self.episode1.delete()
 130         self.podcast1.delete()
 131         self.user.delete()
 132
 133
 134 class MergeGroupTests(TransactionTestCase):
 135     """ Tests merging of two podcasts, one of which is part of a group """
 136
 137     def setUp(self):
 138         self.podcast1 = Podcast.objects.get_or_create_for_url(
 139             'http://example.com/group-merge-feed.rss',
 140             defaults={
 141                 'title': 'Podcast 1',
 142             },
 143         )
 144         self.podcast2 = Podcast.objects.get_or_create_for_url(
 145             'http://test.org/group-merge-podcast/',
 146             defaults={
 147                 'title': 'Podcast 2',
 148             },
 149         )
 150         self.podcast3 = Podcast.objects.get_or_create_for_url(
 151             'http://group-test.org/feed/',
 152             defaults={
 153                 'title': 'Podcast 3',
 154             },
 155         )
 156
 157         self.episode1 = Episode.objects.get_or_create_for_url(
 158             self.podcast1, 'http://example.com/group-merge-episode1.mp3',
 159             defaults={
 160                 'title': 'Episode 1 A',
 161             },
 162         )
 163         self.episode2 = Episode.objects.get_or_create_for_url(
 164             self.podcast2, 'http://example.com/group-merge-episode1.mp3',
 165             defaults={
 166                 'title': 'Episode 1 B',
 167             },
 168         )
 169         self.episode3 = Episode.objects.get_or_create_for_url(
 170             self.podcast3, 'http://example.com/group-merge-media.mp3',
 171             defaults={
 172                 'title': 'Episode 2',
 173             },
 174         )
 175
 176         self.podcast2.group_with(self.podcast3, 'My Group', 'Feed1', 'Feed2')
 177
 178         User = get_user_model()
 179         self.user = User(username='test-merge-group')
 180         self.user.email = 'test-merge-group-tests@example.com'
 181         self.user.set_password('secret!')
 182         self.user.save()
 183
 184     def test_merge_podcasts(self):
 185         podcast1 = Podcast.objects.get(pk=self.podcast1.pk)
 186         podcast2 = Podcast.objects.get(pk=self.podcast2.pk)
 187         podcast3 = Podcast.objects.get(pk=self.podcast3.pk)
 188
 189         # assert that the podcasts are actually grouped
 190         self.assertEqual(podcast2.group, podcast3.group)
 191
 192         # Create additional data that will be merged
 193         state1 = episode_state_for_user_episode(self.user, self.episode1)
 194         state2 = episode_state_for_user_episode(self.user, self.episode2)
 195
 196         action1 = EpisodeAction(action='play',
 197                                 timestamp=datetime.utcnow(),
 198                                 upload_timestamp=get_timestamp(
 199                                     datetime.utcnow()))
 200         action2 = EpisodeAction(action='download',
 201                                 timestamp=datetime.utcnow(),
 202                                 upload_timestamp=get_timestamp(
 203                                     datetime.utcnow()))
 204
 205         add_episode_actions(state1, [action1])
 206         add_episode_actions(state2, [action2])
 207
 208         # decide which episodes to merge
 209         groups = [(0, [self.episode1, self.episode2])]
 210         counter = Counter()
 211
 212         episode2_id = self.episode2.id
 213
 214         pm = PodcastMerger([podcast2, podcast1], counter, groups)
 215         pm.merge()
 216
 217         state1 = episode_state_for_user_episode(self.user, self.episode1)
 218
 219         self.assertIn(action1, state1.actions)
 220         self.assertIn(action2, state1.actions)
 221
 222         episode1 = Episode.objects.get(pk=self.episode1.pk)
 223
 224         # episode2 has been merged into episode1, so it must contain its
 225         # merged _id
 226         self.assertEqual([x.uuid for x in episode1.merged_uuids.all()],
 227                          [episode2_id])
 228
 229     def tearDown(self):
 230         self.episode1.delete()
 231         self.podcast2.delete()
 232         self.user.delete()
 233
 234
 235 def suite():
 236     suite = unittest.TestSuite()
 237     suite.addTest(unittest.TestLoader().loadTestsFromTestCase(MergeTests))
 238     suite.addTest(unittest.TestLoader().loadTestsFromTestCase(MergeGroupTests))
 239     return suite