3 from collections
import Counter
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
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
19 class DeviceSyncTests(unittest
.TestCase
):
21 self
.user
= UserProxy(username
="test")
22 self
.user
.email
= "test@invalid.com"
23 self
.user
.set_password("secret!")
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")
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
)
35 dev3
= Client
.objects
.create(id=uuid
.uuid1(), user
=self
.user
, uid
="d3")
39 groups
= self
.user
.get_grouped_devices()
42 self
.assertEqual(g2
.is_synced
, False)
43 self
.assertIn(dev2
, g2
.devices
)
46 self
.assertEqual(g1
.is_synced
, True)
47 self
.assertIn(dev1
, g1
.devices
)
48 self
.assertIn(dev3
, g1
.devices
)
50 targets
= dev1
.get_sync_targets()
51 target
= next(targets
)
52 self
.assertEqual(target
, dev2
)
55 Client
.objects
.filter(user
=self
.user
).delete()
59 @override_settings(CACHE
={})
60 class UnsubscribeMergeTests(TestCase
):
61 """Test if merged podcasts can be properly unsubscribed"""
63 P2_URL
= "http://test.org/podcast/"
66 self
.podcast1
= Podcast
.objects
.get_or_create_for_url(
67 "http://example.com/feed.rss"
70 self
.podcast2
= Podcast
.objects
.get_or_create_for_url(self
.P2_URL
).object
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!")
78 self
.device
= get_device(self
.user
, "dev", "")
80 def test_merge_podcasts(self
):
81 subscribe(self
.podcast2
.pk
, self
.user
.pk
, self
.device
.uid
)
83 # merge podcast2 into podcast1
84 pm
= PodcastMerger([self
.podcast1
, self
.podcast2
], Counter(), [])
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
)
91 subscriptions
= Podcast
.objects
.filter(subscription__user
=self
.user
)
92 self
.assertEqual(0, len(subscriptions
))
95 self
.podcast1
.delete()
99 class AuthTests(TestCase
):
101 self
.user
, pwd
= create_user()
102 self
.client
= TClient()
103 wrong_pwd
= pwd
+ "1234"
105 "HTTP_AUTHORIZATION": create_auth_string(self
.user
.username
, wrong_pwd
)
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
)