mygpo/api/tests.py

   1 import copy
   2 from datetime import datetime, timedelta
   3 import json
   4 import unittest
   5 import os
   6 import unittest.mock
   7 from urllib.parse import urlencode
   8
   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
  14
  15 from openapi_spec_validator import validate_spec_url
  16 from jsonschema import ValidationError
  17
  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
  25
  26
  27 class AdvancedAPITests(unittest.TestCase):
  28     def setUp(self):
  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)
  34         self.user.save()
  35         self.user.is_active = True
  36         self.client = Client()
  37
  38         self.extra = {
  39             "HTTP_AUTHORIZATION": create_auth_string(self.username, self.password)
  40         }
  41
  42         self.action_data = [
  43             {
  44                 "podcast": "http://example.com/feed.rss",
  45                 "episode": "http://example.com/files/s01e20.mp3",
  46                 "device": "gpodder_abcdef123",
  47                 "action": "download",
  48                 "timestamp": "2009-12-12T09:00:00",
  49             },
  50             {
  51                 "podcast": "http://example.org/podcast.php",
  52                 "episode": "http://ftp.example.org/foo.ogg",
  53                 "action": "play",
  54                 "started": 15,
  55                 "position": 120,
  56                 "total": 500,
  57             },
  58         ]
  59
  60     def tearDown(self):
  61         self.user.delete()
  62
  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)
  66
  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))
  73
  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"
  78
  79         response = self._upload_episode_actions(self.user, action_data, self.extra)
  80
  81         self.assertEqual(response.status_code, 400, response.content)
  82
  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
  87         )
  88
  89     def compare_action_list(self, as1, as2):
  90         for a1 in as1:
  91             found = False
  92             for a2 in as2:
  93                 if self.compare_actions(a1, a2):
  94                     found = True
  95
  96             if not found:
  97                 raise ValueError("%s not found in %s" % (a1, as2))
  98                 return False
  99
 100         return True
 101
 102     def compare_actions(self, a1, a2):
 103         for key, val in a1.items():
 104             if a2.get(key, None) != val:
 105                 return False
 106         return True
 107
 108
 109 class SubscriptionAPITests(unittest.TestCase):
 110     """Tests the Subscription API"""
 111
 112     def setUp(self):
 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)
 119         self.user.save()
 120         self.user.is_active = True
 121         self.client = Client()
 122
 123         self.extra = {
 124             "HTTP_AUTHORIZATION": create_auth_string(self.username, self.password)
 125         }
 126
 127         self.action_data = {"add": ["http://example.com/podcast.rss"]}
 128
 129         self.url = reverse(
 130             "subscriptions-api",
 131             kwargs={
 132                 "version": "2",
 133                 "username": self.user.username,
 134                 "device_uid": self.device_uid,
 135             },
 136         )
 137
 138     def tearDown(self):
 139         self.user.delete()
 140
 141     def test_set_get_subscriptions(self):
 142         """Tests that an upload subscription is returned back correctly"""
 143
 144         # upload a subscription
 145         response = self.client.post(
 146             self.url,
 147             json.dumps(self.action_data),
 148             content_type="application/json",
 149             **self.extra,
 150         )
 151         self.assertEqual(response.status_code, 200, response.content)
 152
 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", []))
 159
 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)
 164
 165
 166 class DirectoryTest(TestCase):
 167     """Test Directory API"""
 168
 169     def setUp(self):
 170         self.podcast = Podcast.objects.get_or_create_for_url(
 171             "http://example.com/directory-podcast.xml", defaults={"title": "My Podcast"}
 172         ).object
 173         self.episode = Episode.objects.get_or_create_for_url(
 174             self.podcast,
 175             "http://example.com/directory-podcast/1.mp3",
 176             defaults={"title": "My Episode"},
 177         ).object
 178         self.client = Client()
 179
 180     def test_episode_info(self):
 181         """Test that the expected number of queries is executed"""
 182         url = (
 183             reverse("api-episode-info")
 184             + "?"
 185             + urlencode((("podcast", self.podcast.url), ("url", self.episode.url)))
 186         )
 187
 188         resp = self.client.get(url)
 189
 190         self.assertEqual(resp.status_code, 200)
 191
 192
 193 class EpisodeActionTests(TestCase):
 194     def setUp(self):
 195         self.podcast = Podcast.objects.get_or_create_for_url(
 196             "http://example.com/directory-podcast.xml", defaults={"title": "My Podcast"}
 197         ).object
 198         self.episode = Episode.objects.get_or_create_for_url(
 199             self.podcast,
 200             "http://example.com/directory-podcast/1.mp3",
 201             defaults={"title": "My Episode"},
 202         ).object
 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)
 208         self.user.save()
 209         self.user.is_active = True
 210         self.client = Client()
 211         self.extra = {
 212             "HTTP_AUTHORIZATION": create_auth_string(self.username, self.password)
 213         }
 214
 215     def tearDown(self):
 216         self.episode.delete()
 217         self.podcast.delete()
 218         self.user.delete()
 219
 220     @override_settings(MAX_EPISODE_ACTIONS=10)
 221     def test_limit_actions(self):
 222         """Test that max MAX_EPISODE_ACTIONS episodes are returned"""
 223
 224         timestamps = []
 225         t = datetime.utcnow()
 226         for n in range(15):
 227             timestamp = t - timedelta(seconds=n)
 228             EpisodeHistoryEntry.objects.create(
 229                 timestamp=timestamp,
 230                 episode=self.episode,
 231                 user=self.user,
 232                 action=EpisodeHistoryEntry.DOWNLOAD,
 233             )
 234             timestamps.append(timestamp)
 235
 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"]
 241
 242         # 10 actions should be returned
 243         self.assertEqual(len(actions), 10)
 244
 245         timestamps = sorted(timestamps)
 246
 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"])
 250
 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"])
 254
 255     def test_no_actions(self):
 256         """Test when there are no actions to return"""
 257
 258         t1 = get_timestamp(datetime.utcnow())
 259
 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"]
 265
 266         # 10 actions should be returned
 267         self.assertEqual(len(actions), 0)
 268
 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)
 275
 276
 277 class SimpleAPITests(unittest.TestCase):
 278     def setUp(self):
 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)
 285         self.user.save()
 286         self.user.is_active = True
 287         self.client = Client()
 288         self.extra = {
 289             "HTTP_AUTHORIZATION": create_auth_string(self.username, self.password)
 290         }
 291         self.formats = ["txt", "json", "jsonp", "opml"]
 292         self.subscriptions_urls = dict(
 293             (fmt, self.get_subscriptions_url(fmt)) for fmt in self.formats
 294         )
 295         self.blank_values = {
 296             "txt": b"\n",
 297             "json": b"[]",
 298             "opml": Exporter("Subscriptions").generate([]),
 299         }
 300         self.all_subscriptions_url = reverse(
 301             "api-all-subscriptions",
 302             kwargs={"format": "txt", "username": self.user.username},
 303         )
 304         self.toplist_urls = dict(
 305             (fmt, self.get_toplist_url(fmt)) for fmt in self.formats
 306         )
 307         self.search_urls = dict((fmt, self.get_search_url(fmt)) for fmt in self.formats)
 308
 309     def tearDown(self):
 310         self.user.delete()
 311
 312     def get_toplist_url(self, fmt):
 313         return reverse("api-simple-toplist-50", kwargs={"format": fmt})
 314
 315     def get_subscriptions_url(self, fmt):
 316         return reverse(
 317             "api-simple-subscriptions",
 318             kwargs={
 319                 "format": fmt,
 320                 "username": self.user.username,
 321                 "device_uid": self.device_uid,
 322             },
 323         )
 324
 325     def get_search_url(self, fmt):
 326         return reverse("api-simple-search", kwargs={"format": fmt})
 327
 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)
 332
 333     def test_get_subscriptions_empty(self):
 334         testers = {
 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, []),
 339         }
 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)
 345
 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)
 350
 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"}
 355         ).object
 356         with unittest.mock.patch(
 357             "mygpo.users.models.Client.get_subscribed_podcasts"
 358         ) as mock_get:
 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)
 365
 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"}
 370         ).object
 371         payloads = {
 372             "txt": sample_url,
 373             "json": json.dumps([sample_url]),
 374             #'opml': Exporter('Subscriptions').generate([sample_url]),
 375             "opml": Exporter("Subscriptions").generate([podcast]),
 376         }
 377         payloads = dict(
 378             (fmt, format_podcast_list([podcast], fmt, "test title").content)
 379             for fmt in self.formats
 380         )
 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)
 386
 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)
 392
 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
 396         )
 397         self.assertEqual(response.status_code, 400, response.content)
 398
 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
 402         )
 403         self.assertEqual(response.status_code, 400, response.content)
 404
 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)
 408
 409     def test_get_toplist_invalid_scale(self):
 410         response = self.client.get(
 411             self.toplist_urls["opml"], data={"scale_logo": 0}, **self.extra
 412         )
 413         self.assertEqual(response.status_code, 400, response.content)
 414
 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
 418         )
 419         self.assertEqual(response.status_code, 400, response.content)
 420
 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)
 424
 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"
 429
 430         self._test_response_for_data(
 431             self.search_urls["json"], data, expected_status, expected_content
 432         )
 433
 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"
 438
 439         self._test_response_for_data(
 440             self.search_urls["opml"], data, expected_status, expected_content
 441         )
 442
 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}"
 447
 448         self._test_response_for_data(
 449             self.search_urls["opml"], data, expected_status, expected_content
 450         )
 451
 452     def test_search_valid_query_status(self):
 453         data = {"scale_logo": 1, "q": "foo"}
 454         expected_status = 200
 455
 456         response = self.client.get(self.search_urls["json"], data)
 457         self.assertEqual(response.status_code, expected_status)
 458
 459
 460 class OpenAPIDefinitionValidityTest(TestCase):
 461     "Test the validity of the OpenAPI definition file"
 462
 463     def test_api_definition_validity(self):
 464         validate_spec_url("file://" + os.path.abspath("./mygpo/api/openapi.yaml"))