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