mygpo/users/tests.py

   1 import uuid
   2 import unittest
   3 from collections import Counter
   4
   5 from django.urls import reverse
   6 from django.test.client import Client as TClient
   7 from django.test import TestCase
   8 from django.test.utils import override_settings
   9 from django.contrib.auth import get_user_model
  10
  11 from mygpo.test import create_auth_string, create_user
  12 from mygpo.podcasts.models import Podcast
  13 from mygpo.maintenance.merge import PodcastMerger
  14 from mygpo.api.backend import get_device
  15 from mygpo.users.models import Client, SyncGroup, UserProxy
  16 from mygpo.subscriptions.tasks import subscribe, unsubscribe
  17
  18
  19 class DeviceSyncTests(unittest.TestCase):
  20     def setUp(self):
  21         self.user = UserProxy(username="test")
  22         self.user.email = "test@invalid.com"
  23         self.user.set_password("secret!")
  24         self.user.save()
  25
  26     def test_group(self):
  27         dev1 = Client.objects.create(id=uuid.uuid1(), user=self.user, uid="d1")
  28         dev2 = Client.objects.create(id=uuid.uuid1(), user=self.user, uid="d2")
  29
  30         group = next(self.user.get_grouped_devices())
  31         self.assertEqual(group.is_synced, False)
  32         self.assertIn(dev1, group.devices)
  33         self.assertIn(dev2, group.devices)
  34
  35         dev3 = Client.objects.create(id=uuid.uuid1(), user=self.user, uid="d3")
  36
  37         dev1.sync_with(dev3)
  38
  39         groups = self.user.get_grouped_devices()
  40
  41         g2 = next(groups)
  42         self.assertEqual(g2.is_synced, False)
  43         self.assertIn(dev2, g2.devices)
  44
  45         g1 = next(groups)
  46         self.assertEqual(g1.is_synced, True)
  47         self.assertIn(dev1, g1.devices)
  48         self.assertIn(dev3, g1.devices)
  49
  50         targets = dev1.get_sync_targets()
  51         target = next(targets)
  52         self.assertEqual(target, dev2)
  53
  54     def tearDown(self):
  55         Client.objects.filter(user=self.user).delete()
  56         self.user.delete()
  57
  58
  59 @override_settings(CACHE={})
  60 class UnsubscribeMergeTests(TestCase):
  61     """Test if merged podcasts can be properly unsubscribed"""
  62
  63     P2_URL = "http://test.org/podcast/"
  64
  65     def setUp(self):
  66         self.podcast1 = Podcast.objects.get_or_create_for_url(
  67             "http://example.com/feed.rss"
  68         ).object
  69
  70         self.podcast2 = Podcast.objects.get_or_create_for_url(self.P2_URL).object
  71
  72         User = get_user_model()
  73         self.user = User(username="test-merge")
  74         self.user.email = "test@example.com"
  75         self.user.set_password("secret!")
  76         self.user.save()
  77
  78         self.device = get_device(self.user, "dev", "")
  79
  80     def test_merge_podcasts(self):
  81         subscribe(self.podcast2.pk, self.user.pk, self.device.uid)
  82
  83         # merge podcast2 into podcast1
  84         pm = PodcastMerger([self.podcast1, self.podcast2], Counter(), [])
  85         pm.merge()
  86
  87         # get podcast for URL of podcast2 and unsubscribe from it
  88         p = Podcast.objects.get(urls__url=self.P2_URL)
  89         unsubscribe(p.pk, self.user.pk, self.device.uid)
  90
  91         subscriptions = Podcast.objects.filter(subscription__user=self.user)
  92         self.assertEqual(0, len(subscriptions))
  93
  94     def tearDown(self):
  95         self.podcast1.delete()
  96         self.user.delete()
  97
  98
  99 class AuthTests(TestCase):
 100     def setUp(self):
 101         self.user, pwd = create_user()
 102         self.client = TClient()
 103         wrong_pwd = pwd + "1234"
 104         self.extra = {
 105             "HTTP_AUTHORIZATION": create_auth_string(self.user.username, wrong_pwd)
 106         }
 107
 108     def test_queries_failed_auth(self):
 109         """Verifies the number of queries that are executed on failed auth"""
 110         url = reverse("api-all-subscriptions", args=(self.user.username, "opml"))
 111         with self.assertNumQueries(1):
 112             resp = self.client.get(url, **self.extra)
 113         self.assertEqual(resp.status_code, 401, resp.content)