2 from datetime
import datetime
, timedelta
7 from urllib
.parse
import urlencode
9 from django
.test
.client
import Client
10 from django
.test
import TestCase
11 from django
.urls
import reverse
12 from django
.contrib
.auth
import get_user_model
13 from django
.test
.utils
import override_settings
15 from openapi_spec_validator
import validate_spec_url
16 from jsonschema
import ValidationError
18 from mygpo
.podcasts
.models
import Podcast
, Episode
19 from mygpo
.api
.advanced
import episodes
20 from mygpo
.api
.opml
import Exporter
, Importer
21 from mygpo
.api
.simple
import format_podcast_list
22 from mygpo
.history
.models
import EpisodeHistoryEntry
23 from mygpo
.test
import create_auth_string
24 from mygpo
.utils
import get_timestamp
27 class AdvancedAPITests(unittest
.TestCase
):
29 User
= get_user_model()
30 self
.password
= "asdf"
31 self
.username
= "adv-api-user"
32 self
.user
= User(username
=self
.username
, email
="user@example.com")
33 self
.user
.set_password(self
.password
)
35 self
.user
.is_active
= True
36 self
.client
= Client()
39 "HTTP_AUTHORIZATION": create_auth_string(self
.username
, self
.password
)
44 "podcast": "http://example.com/feed.rss",
45 "episode": "http://example.com/files/s01e20.mp3",
46 "device": "gpodder_abcdef123",
48 "timestamp": "2009-12-12T09:00:00",
51 "podcast": "http://example.org/podcast.php",
52 "episode": "http://ftp.example.org/foo.ogg",
63 def test_episode_actions(self
):
64 response
= self
._upload
_episode
_actions
(self
.user
, self
.action_data
, self
.extra
)
65 self
.assertEqual(response
.status_code
, 200, response
.content
)
67 url
= reverse(episodes
, kwargs
={"version": "2", "username": self
.user
.username
})
68 response
= self
.client
.get(url
, {"since": "0"}, **self
.extra
)
69 self
.assertEqual(response
.status_code
, 200, response
.content
)
70 response_obj
= json
.loads(response
.content
.decode("utf-8"))
71 actions
= response_obj
["actions"]
72 self
.assertTrue(self
.compare_action_list(self
.action_data
, actions
))
74 def test_invalid_client_id(self
):
75 """Invalid Client ID should return 400"""
76 action_data
= copy
.deepcopy(self
.action_data
)
77 action_data
[0]["device"] = "gpodder@abcdef123"
79 response
= self
._upload
_episode
_actions
(self
.user
, action_data
, self
.extra
)
81 self
.assertEqual(response
.status_code
, 400, response
.content
)
83 def _upload_episode_actions(self
, user
, action_data
, extra
):
84 url
= reverse(episodes
, kwargs
={"version": "2", "username": self
.user
.username
})
85 return self
.client
.post(
86 url
, json
.dumps(action_data
), content_type
="application/json", **extra
89 def compare_action_list(self
, as1
, as2
):
93 if self
.compare_actions(a1
, a2
):
97 raise ValueError("%s not found in %s" % (a1
, as2
))
102 def compare_actions(self
, a1
, a2
):
103 for key
, val
in a1
.items():
104 if a2
.get(key
, None) != val
:
109 class SubscriptionAPITests(unittest
.TestCase
):
110 """Tests the Subscription API"""
113 User
= get_user_model()
114 self
.password
= "asdf"
115 self
.username
= "subscription-api-user"
116 self
.device_uid
= "test-device"
117 self
.user
= User(username
=self
.username
, email
="user@example.com")
118 self
.user
.set_password(self
.password
)
120 self
.user
.is_active
= True
121 self
.client
= Client()
124 "HTTP_AUTHORIZATION": create_auth_string(self
.username
, self
.password
)
127 self
.action_data
= {"add": ["http://example.com/podcast.rss"]}
133 "username": self
.user
.username
,
134 "device_uid": self
.device_uid
,
141 def test_set_get_subscriptions(self
):
142 """Tests that an upload subscription is returned back correctly"""
144 # upload a subscription
145 response
= self
.client
.post(
147 json
.dumps(self
.action_data
),
148 content_type
="application/json",
151 self
.assertEqual(response
.status_code
, 200, response
.content
)
153 # verify that the subscription is returned correctly
154 response
= self
.client
.get(self
.url
, {"since": "0"}, **self
.extra
)
155 self
.assertEqual(response
.status_code
, 200, response
.content
)
156 response_obj
= json
.loads(response
.content
.decode("utf-8"))
157 self
.assertEqual(self
.action_data
["add"], response_obj
["add"])
158 self
.assertEqual([], response_obj
.get("remove", []))
160 def test_unauth_request(self
):
161 """Tests that an unauthenticated request gives a 401 response"""
162 response
= self
.client
.get(self
.url
, {"since": "0"})
163 self
.assertEqual(response
.status_code
, 401, response
.content
)
166 class DirectoryTest(TestCase
):
167 """Test Directory API"""
170 self
.podcast
= Podcast
.objects
.get_or_create_for_url(
171 "http://example.com/directory-podcast.xml", defaults
={"title": "My Podcast"}
173 self
.episode
= Episode
.objects
.get_or_create_for_url(
175 "http://example.com/directory-podcast/1.mp3",
176 defaults
={"title": "My Episode"},
178 self
.client
= Client()
180 def test_episode_info(self
):
181 """Test that the expected number of queries is executed"""
183 reverse("api-episode-info")
185 + urlencode((("podcast", self
.podcast
.url
), ("url", self
.episode
.url
)))
188 resp
= self
.client
.get(url
)
190 self
.assertEqual(resp
.status_code
, 200)
193 class EpisodeActionTests(TestCase
):
195 self
.podcast
= Podcast
.objects
.get_or_create_for_url(
196 "http://example.com/directory-podcast.xml", defaults
={"title": "My Podcast"}
198 self
.episode
= Episode
.objects
.get_or_create_for_url(
200 "http://example.com/directory-podcast/1.mp3",
201 defaults
={"title": "My Episode"},
203 User
= get_user_model()
204 self
.password
= "asdf"
205 self
.username
= "adv-api-user"
206 self
.user
= User(username
=self
.username
, email
="user@example.com")
207 self
.user
.set_password(self
.password
)
209 self
.user
.is_active
= True
210 self
.client
= Client()
212 "HTTP_AUTHORIZATION": create_auth_string(self
.username
, self
.password
)
216 self
.episode
.delete()
217 self
.podcast
.delete()
220 @override_settings(MAX_EPISODE_ACTIONS
=10)
221 def test_limit_actions(self
):
222 """Test that max MAX_EPISODE_ACTIONS episodes are returned"""
225 t
= datetime
.utcnow()
227 timestamp
= t
- timedelta(seconds
=n
)
228 EpisodeHistoryEntry
.objects
.create(
230 episode
=self
.episode
,
232 action
=EpisodeHistoryEntry
.DOWNLOAD
,
234 timestamps
.append(timestamp
)
236 url
= reverse(episodes
, kwargs
={"version": "2", "username": self
.user
.username
})
237 response
= self
.client
.get(url
, {"since": "0"}, **self
.extra
)
238 self
.assertEqual(response
.status_code
, 200, response
.content
)
239 response_obj
= json
.loads(response
.content
.decode("utf-8"))
240 actions
= response_obj
["actions"]
242 # 10 actions should be returned
243 self
.assertEqual(len(actions
), 10)
245 timestamps
= sorted(timestamps
)
247 # the first 10 actions, according to their timestamp should be returned
248 for action
, timestamp
in zip(actions
, timestamps
):
249 self
.assertEqual(timestamp
.isoformat(), action
["timestamp"])
251 # the `timestamp` field in the response should be the timestamp of the
252 # last returned action
253 self
.assertEqual(get_timestamp(timestamps
[9]), response_obj
["timestamp"])
255 def test_no_actions(self
):
256 """Test when there are no actions to return"""
258 t1
= get_timestamp(datetime
.utcnow())
260 url
= reverse(episodes
, kwargs
={"version": "2", "username": self
.user
.username
})
261 response
= self
.client
.get(url
, {"since": "0"}, **self
.extra
)
262 self
.assertEqual(response
.status_code
, 200, response
.content
)
263 response_obj
= json
.loads(response
.content
.decode("utf-8"))
264 actions
= response_obj
["actions"]
266 # 10 actions should be returned
267 self
.assertEqual(len(actions
), 0)
269 returned
= response_obj
["timestamp"]
270 t2
= get_timestamp(datetime
.utcnow())
271 # the `timestamp` field in the response should be the timestamp of the
272 # last returned action
273 self
.assertGreaterEqual(returned
, t1
)
274 self
.assertGreaterEqual(t2
, returned
)
277 class SimpleAPITests(unittest
.TestCase
):
279 User
= get_user_model()
280 self
.password
= "asdf"
281 self
.username
= "subscription-api-user"
282 self
.device_uid
= "test-device"
283 self
.user
= User(username
=self
.username
, email
="user@example.com")
284 self
.user
.set_password(self
.password
)
286 self
.user
.is_active
= True
287 self
.client
= Client()
289 "HTTP_AUTHORIZATION": create_auth_string(self
.username
, self
.password
)
291 self
.formats
= ["txt", "json", "jsonp", "opml"]
292 self
.subscriptions_urls
= dict(
293 (fmt
, self
.get_subscriptions_url(fmt
)) for fmt
in self
.formats
295 self
.blank_values
= {
298 "opml": Exporter("Subscriptions").generate([]),
300 self
.all_subscriptions_url
= reverse(
301 "api-all-subscriptions",
302 kwargs
={"format": "txt", "username": self
.user
.username
},
304 self
.toplist_urls
= dict(
305 (fmt
, self
.get_toplist_url(fmt
)) for fmt
in self
.formats
307 self
.search_urls
= dict((fmt
, self
.get_search_url(fmt
)) for fmt
in self
.formats
)
312 def get_toplist_url(self
, fmt
):
313 return reverse("api-simple-toplist-50", kwargs
={"format": fmt
})
315 def get_subscriptions_url(self
, fmt
):
317 "api-simple-subscriptions",
320 "username": self
.user
.username
,
321 "device_uid": self
.device_uid
,
325 def get_search_url(self
, fmt
):
326 return reverse("api-simple-search", kwargs
={"format": fmt
})
328 def _test_response_for_data(self
, url
, data
, status_code
, content
):
329 response
= self
.client
.get(url
, data
)
330 self
.assertEqual(response
.status_code
, status_code
)
331 self
.assertEqual(response
.content
, content
)
333 def test_get_subscriptions_empty(self
):
335 "txt": lambda c
: self
.assertEqual(c
, b
""),
336 "json": lambda c
: self
.assertEqual(c
, b
"[]"),
337 "jsonp": lambda c
: self
.assertEqual(c
, b
"test([])"),
338 "opml": lambda c
: self
.assertListEqual(Importer(c
).items
, []),
340 for fmt
in self
.formats
:
341 url
= self
.subscriptions_urls
[fmt
]
342 response
= self
.client
.get(url
, data
={"jsonp": "test"}, **self
.extra
)
343 self
.assertEqual(response
.status_code
, 200, response
.content
)
344 testers
[fmt
](response
.content
)
346 def test_get_subscriptions_invalid_jsonp(self
):
347 url
= self
.subscriptions_urls
["jsonp"]
348 response
= self
.client
.get(url
, data
={"jsonp": "!"}, **self
.extra
)
349 self
.assertEqual(response
.status_code
, 400, response
.content
)
351 def test_get_subscriptions_with_content(self
):
352 sample_url
= "http://example.com/directory-podcast.xml"
353 podcast
= Podcast
.objects
.get_or_create_for_url(
354 sample_url
, defaults
={"title": "My Podcast"}
356 with unittest
.mock
.patch(
357 "mygpo.users.models.Client.get_subscribed_podcasts"
359 mock_get
.return_value
= [podcast
]
360 response
= self
.client
.get(self
.subscriptions_urls
["txt"], **self
.extra
)
361 self
.assertEqual(response
.status_code
, 200, response
.content
)
362 retrieved_urls
= response
.content
.split(b
"\n")[:-1]
363 expected_urls
= [sample_url
.encode()]
364 self
.assertEqual(retrieved_urls
, expected_urls
)
366 def test_post_subscription_valid(self
):
367 sample_url
= "http://example.com/directory-podcast.xml"
368 podcast
= Podcast
.objects
.get_or_create_for_url(
369 sample_url
, defaults
={"title": "My Podcast"}
373 "json": json
.dumps([sample_url
]),
374 #'opml': Exporter('Subscriptions').generate([sample_url]),
375 "opml": Exporter("Subscriptions").generate([podcast
]),
378 (fmt
, format_podcast_list([podcast
], fmt
, "test title").content
)
379 for fmt
in self
.formats
381 for fmt
in self
.formats
:
382 url
= self
.subscriptions_urls
[fmt
]
383 payload
= payloads
[fmt
]
384 response
= self
.client
.generic("POST", url
, payload
, **self
.extra
)
385 self
.assertEqual(response
.status_code
, 200, response
.content
)
387 def test_post_subscription_invalid(self
):
388 url
= self
.subscriptions_urls
["json"]
389 payload
= "invalid_json"
390 response
= self
.client
.generic("POST", url
, payload
, **self
.extra
)
391 self
.assertEqual(response
.status_code
, 400, response
.content
)
393 def test_get_all_subscriptions_invalid_scale(self
):
394 response
= self
.client
.get(
395 self
.all_subscriptions_url
, data
={"scale_logo": 0}, **self
.extra
397 self
.assertEqual(response
.status_code
, 400, response
.content
)
399 def test_get_all_subscriptions_non_numeric_scale(self
):
400 response
= self
.client
.get(
401 self
.all_subscriptions_url
, data
={"scale_logo": "a"}, **self
.extra
403 self
.assertEqual(response
.status_code
, 400, response
.content
)
405 def test_get_all_subscriptions_valid_empty(self
):
406 response
= self
.client
.get(self
.all_subscriptions_url
, **self
.extra
)
407 self
.assertEqual(response
.status_code
, 200, response
.content
)
409 def test_get_toplist_invalid_scale(self
):
410 response
= self
.client
.get(
411 self
.toplist_urls
["opml"], data
={"scale_logo": 0}, **self
.extra
413 self
.assertEqual(response
.status_code
, 400, response
.content
)
415 def test_get_toplist_non_numeric_scale(self
):
416 response
= self
.client
.get(
417 self
.toplist_urls
["txt"], data
={"scale_logo": "a"}, **self
.extra
419 self
.assertEqual(response
.status_code
, 400, response
.content
)
421 def test_get_toplist_valid_empty(self
):
422 response
= self
.client
.get(self
.toplist_urls
["json"], **self
.extra
)
423 self
.assertEqual(response
.status_code
, 200, response
.content
)
425 def test_search_non_numeric_scale_logo(self
):
426 data
= {"scale_logo": "a"}
427 expected_status
= 400
428 expected_content
= b
"scale_logo has to be a numeric value"
430 self
._test
_response
_for
_data
(
431 self
.search_urls
["json"], data
, expected_status
, expected_content
434 def test_search_scale_out_of_range(self
):
435 data
= {"scale_logo": 3000}
436 expected_status
= 400
437 expected_content
= b
"scale_logo has to be a number from 1 to 256"
439 self
._test
_response
_for
_data
(
440 self
.search_urls
["opml"], data
, expected_status
, expected_content
443 def test_search_no_query(self
):
444 data
= {"scale_logo": 1}
445 expected_status
= 400
446 expected_content
= b
"/search.opml|txt|json?q={query}"
448 self
._test
_response
_for
_data
(
449 self
.search_urls
["opml"], data
, expected_status
, expected_content
452 def test_search_valid_query_status(self
):
453 data
= {"scale_logo": 1, "q": "foo"}
454 expected_status
= 200
456 response
= self
.client
.get(self
.search_urls
["json"], data
)
457 self
.assertEqual(response
.status_code
, expected_status
)
460 class OpenAPIDefinitionValidityTest(TestCase
):
461 "Test the validity of the OpenAPI definition file"
463 def test_api_definition_validity(self
):
464 validate_spec_url("file://" + os
.path
.abspath("./mygpo/api/openapi.yaml"))