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         unsubscribe(p1, user, device1)
  58         subscribe(p1, user, device1)
  59         subscribe(p2, user, device2)
  60
  61         action1 = EpisodeHistoryEntry.create_entry(user, e1,
  62                                                    EpisodeHistoryEntry.PLAY)
  63         action3 = EpisodeHistoryEntry.create_entry(user, e3,
  64                                                    EpisodeHistoryEntry.PLAY)
  65
  66         # we need that for later
  67         e3_id = e3.pk
  68
  69         actions = Counter()
  70
  71         # decide which episodes to merge
  72         groups = [(0, [e1]), (1, [e2, e3]), (2, [e4])]
  73
  74         # carry out the merge
  75         pm = PodcastMerger([p1, p2], actions, groups)
  76         pm.merge()
  77
  78         e1 = Episode.objects.get(pk=e1.pk)
  79         history1 = EpisodeHistoryEntry.objects.filter(episode=e1, user=user)
  80         self.assertEqual(len(history1), 1)
  81
  82         # check if merged episode's id can still be accessed
  83         e3 = Episode.objects.filter(podcast=p1).get_by_any_id(e3_id)
  84         history3 = EpisodeHistoryEntry.objects.filter(episode=e3, user=user)
  85         self.assertEqual(len(history3), 1)
  86
  87         p1 = Podcast.objects.get(pk=p1.get_id())
  88         subscribed_clients = Client.objects.filter(subscription__podcast=p1)
  89         self.assertEqual(len(subscribed_clients), 2)
  90
  91         episodes = p1.episode_set.all()
  92         self.assertEqual(len(episodes), 3)