mygpo/maintenance/tests.py

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