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 from datetime import datetime
  19 import unittest
  20
  21 from django.test import TestCase
  22
  23 from mygpo.core.models import Podcast, Episode
  24 from mygpo.users.models import EpisodeAction, User
  25 from mygpo.maintenance.merge import PodcastMerger
  26 from mygpo.counter import Counter
  27 from mygpo.db.couchdb.episode_state import episode_state_for_user_episode
  28
  29
  30 class MergeTests(TestCase):
  31
  32     def setUp(self):
  33         self.podcast1 = Podcast(urls=['http://example.com/feed.rss'])
  34         self.podcast2 = Podcast(urls=['http://test.org/podcast/'])
  35         self.podcast1.save()
  36         self.podcast2.save()
  37
  38         self.episode1 = Episode(podcast=self.podcast1.get_id(),
  39                 urls = ['http://example.com/episode1.mp3'])
  40         self.episode2 = Episode(podcast=self.podcast2.get_id(),
  41                 urls = ['http://example.com/episode1.mp3'])
  42
  43         self.episode1.save()
  44         self.episode2.save()
  45
  46         self.user = User(username='test')
  47         self.user.email = 'test@example.com'
  48         self.user.set_password('secret!')
  49         self.user.save()
  50
  51
  52     def test_merge_podcasts(self):
  53
  54         state1 = episode_state_for_user_episode(self.user, self.episode1)
  55         state2 = episode_state_for_user_episode(self.user, self.episode2)
  56
  57         action1 = EpisodeAction(action='play', timestamp=datetime.utcnow())
  58         action2 = EpisodeAction(action='download', timestamp=datetime.utcnow())
  59
  60         state1.add_actions([action1])
  61         state2.add_actions([action2])
  62
  63         state1.save()
  64         state2.save()
  65
  66         # decide which episodes to merge
  67         groups = [(0, [self.episode1, self.episode2])]
  68         counter = Counter()
  69
  70         pm = PodcastMerger([self.podcast1, self.podcast2], counter, groups)
  71         pm.merge()
  72
  73         state1 = episode_state_for_user_episode(self.user, self.episode1)
  74
  75         self.assertIn(action1, state1.actions)
  76         self.assertIn(action2, state1.actions)
  77
  78
  79
  80     def tearDown(self):
  81         self.podcast1.delete()
  82         self.episode1.delete()
  83
  84         #self.podcast2.delete()
  85         #self.episode2.delete()
  86
  87         self.user.delete()
  88
  89
  90 def suite():
  91     suite = unittest.TestSuite()
  92     suite.addTest(unittest.TestLoader().loadTestsFromTestCase(MergeTests))
  93     return suite