mygpo/administration/tests.py

   1 """
   2 This file demonstrates writing tests using the unittest module. These will pass
   3 when you run "manage.py test".
   4
   5 Replace this with more appropriate tests for your application.
   6 """
   7
   8 import uuid
   9 import time
  10 from datetime import datetime
  11 from collections import Counter
  12
  13 from django.test import TestCase
  14 from django.contrib.auth import get_user_model
  15
  16 from mygpo.podcasts.models import Podcast, Episode
  17 from mygpo.users.models import Client
  18 from mygpo.history.models import EpisodeHistoryEntry
  19 from mygpo.maintenance.merge import PodcastMerger
  20 from mygpo.subscriptions import subscribe, unsubscribe
  21
  22
  23 class SimpleTest(TestCase):
  24
  25     def test_merge(self):
  26
  27         p1 = Podcast.objects.get_or_create_for_url('http://example.com/podcast1.rss')
  28         p2 = Podcast.objects.get_or_create_for_url('http://example.com/podcast2.rss')
  29
  30         e1 = Episode.objects.get_or_create_for_url(p1, 'http://example.com/podcast1/e1.mp3')
  31         e1.title = 'Episode 1'
  32         e1.save()
  33
  34         e2 = Episode.objects.get_or_create_for_url(p2, 'http://example.com/podcast1/e2.mp3')
  35         e2.title = 'Episode 2'
  36         e2.save()
  37
  38         e3 = Episode.objects.get_or_create_for_url(p2, 'http://example.com/podcast2/e2.mp3')
  39         e3.title = 'Episode 3'
  40         e3.save()
  41
  42         e4 = Episode.objects.get_or_create_for_url(p2, 'http://example.com/podcast2/e3.mp3')
  43         e4.title = 'Episode 4'
  44         e4.save()
  45
  46         User = get_user_model()
  47         user = User()
  48         user.username = 'user-test_merge'
  49         user.email = 'user-test_merge@example.com'
  50         user.set_password('secret')
  51         user.save()
  52
  53         device1 = Client.objects.create(user=user, uid='dev1', id=uuid.uuid1())
  54         device2 = Client.objects.create(user=user, uid='dev2', id=uuid.uuid1())
  55
  56         subscribe(p1, user, device1)
  57         time.sleep(1)
  58         unsubscribe(p1, user, device1)
  59         time.sleep(1)
  60         subscribe(p1, user, device1)
  61         subscribe(p2, user, device2)
  62
  63         action1 = EpisodeHistoryEntry.objects.create(
  64             episode = e1,
  65             user = user,
  66             action = EpisodeHistoryEntry.PLAY,
  67             timestamp = datetime.utcnow(),
  68         )
  69
  70         action3 = EpisodeHistoryEntry.objects.create(
  71             episode = e3,
  72             user = user,
  73             action = EpisodeHistoryEntry.PLAY,
  74             timestamp = datetime.utcnow(),
  75         )
  76
  77         # we need that for later
  78         e3_id = e3.pk
  79
  80         actions = Counter()
  81
  82         # decide which episodes to merge
  83         groups = [(0, [e1]), (1, [e2, e3]), (2, [e4])]
  84
  85         # carry out the merge
  86         pm = PodcastMerger([p1, p2], actions, groups)
  87         pm.merge()
  88
  89         e1 = Episode.objects.get(pk=e1.pk)
  90         history1 = EpisodeHistoryEntry.objects.filter(episode=e1, user=user)
  91         self.assertEqual(len(history1), 1)
  92
  93         # check if merged episode's id can still be accessed
  94         e3 = Episode.objects.filter(podcast=p1).get_by_any_id(e3_id)
  95         history3 = EpisodeHistoryEntry.objects.filter(episode=e3, user=user)
  96         self.assertEqual(len(history3), 1)
  97
  98         p1 = Podcast.objects.get(pk=p1.get_id())
  99         subscribed_clients = Client.objects.filter(subscription__podcast=p1)
 100         self.assertEqual(len(subscribed_clients), 2)
 101
 102         episodes = p1.episode_set.all()
 103         self.assertEqual(len(episodes), 3)