mediagoblin/tests/test_api.py

   1 # GNU MediaGoblin -- federated, autonomous media hosting
   2 # Copyright (C) 2011, 2012 MediaGoblin contributors.  See AUTHORS.
   3 #
   4 # This program is free software: you can redistribute it and/or modify
   5 # it under the terms of the GNU Affero General Public License as published by
   6 # the Free Software Foundation, either version 3 of the License, or
   7 # (at your option) any later version.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU Affero General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU Affero General Public License
  15 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16 import json
  17
  18 try:
  19     import mock
  20 except ImportError:
  21     import unittest.mock as mock
  22 import pytest
  23
  24 from webtest import AppError
  25
  26 from .resources import GOOD_JPG
  27 from mediagoblin import mg_globals
  28 from mediagoblin.db.models import User, MediaEntry, TextComment
  29 from mediagoblin.tests.tools import fixture_add_user
  30 from mediagoblin.moderation.tools import take_away_privileges
  31
  32
  33 class TestAPI(object):
  34     """ Test mediagoblin's pump.io complient APIs """
  35
  36     @pytest.fixture(autouse=True)
  37     def setup(self, test_app):
  38         self.test_app = test_app
  39         self.db = mg_globals.database
  40
  41         self.user = fixture_add_user(privileges=[u'active', u'uploader',
  42                                                  u'commenter'])
  43         self.other_user = fixture_add_user(
  44             username="otheruser",
  45             privileges=[u'active', u'uploader', u'commenter']
  46         )
  47         self.active_user = self.user
  48
  49     def _activity_to_feed(self, test_app, activity, headers=None):
  50         """ Posts an activity to the user's feed """
  51         if headers:
  52             headers.setdefault("Content-Type", "application/json")
  53         else:
  54             headers = {"Content-Type": "application/json"}
  55
  56         with self.mock_oauth():
  57             response = test_app.post(
  58                 "/api/user/{0}/feed".format(self.active_user.username),
  59                 json.dumps(activity),
  60                 headers=headers
  61             )
  62
  63         return response, json.loads(response.body.decode())
  64
  65     def _upload_image(self, test_app, image, custom_filename=None):
  66         """ Uploads and image to MediaGoblin via pump.io API """
  67         data = open(image, "rb").read()
  68         headers = {
  69             "Content-Type": "image/jpeg",
  70             "Content-Length": str(len(data))
  71         }
  72
  73         if custom_filename is not None:
  74             headers["X-File-Name"] = custom_filename
  75
  76         with self.mock_oauth():
  77             response = test_app.post(
  78                 "/api/user/{0}/uploads".format(self.active_user.username),
  79                 data,
  80                 headers=headers
  81             )
  82             image = json.loads(response.body.decode())
  83
  84         return response, image
  85
  86     def _post_image_to_feed(self, test_app, image):
  87         """ Posts an already uploaded image to feed """
  88         activity = {
  89             "verb": "post",
  90             "object": image,
  91         }
  92
  93         return self._activity_to_feed(test_app, activity)
  94
  95     def mocked_oauth_required(self, *args, **kwargs):
  96         """ Mocks mediagoblin.decorator.oauth_required to always validate """
  97
  98         def fake_controller(controller, request, *args, **kwargs):
  99             request.user = User.query.filter_by(id=self.active_user.id).first()
 100             return controller(request, *args, **kwargs)
 101
 102         def oauth_required(c):
 103             return lambda *args, **kwargs: fake_controller(c, *args, **kwargs)
 104
 105         return oauth_required
 106
 107     def mock_oauth(self):
 108         """ Returns a mock.patch for the oauth_required decorator """
 109         return mock.patch(
 110             target="mediagoblin.decorators.oauth_required",
 111             new_callable=self.mocked_oauth_required
 112         )
 113
 114     def test_can_post_image(self, test_app):
 115         """ Tests that an image can be posted to the API """
 116         # First request we need to do is to upload the image
 117         response, image = self._upload_image(test_app, GOOD_JPG)
 118
 119         # I should have got certain things back
 120         assert response.status_code == 200
 121
 122         assert "id" in image
 123         assert "fullImage" in image
 124         assert "url" in image["fullImage"]
 125         assert "url" in image
 126         assert "author" in image
 127         assert "published" in image
 128         assert "updated" in image
 129         assert image["objectType"] == "image"
 130
 131         # Check that we got the response we're expecting
 132         response, data = self._post_image_to_feed(test_app, image)
 133         assert response.status_code == 200
 134         assert data["object"]["fullImage"]["url"].endswith("unknown.jpe")
 135         assert data["object"]["image"]["url"].endswith("unknown.thumbnail.jpe")
 136
 137     def test_can_post_image_custom_filename(self, test_app):
 138         """ Tests an image can be posted to the API with custom filename """
 139         # First request we need to do is to upload the image
 140         response, image = self._upload_image(test_app, GOOD_JPG,
 141                                              custom_filename="hello.jpg")
 142
 143         # I should have got certain things back
 144         assert response.status_code == 200
 145
 146         assert "id" in image
 147         assert "fullImage" in image
 148         assert "url" in image["fullImage"]
 149         assert "url" in image
 150         assert "author" in image
 151         assert "published" in image
 152         assert "updated" in image
 153         assert image["objectType"] == "image"
 154
 155         # Check that we got the response we're expecting
 156         response, data = self._post_image_to_feed(test_app, image)
 157         assert response.status_code == 200
 158         assert data["object"]["fullImage"]["url"].endswith("hello.jpg")
 159         assert data["object"]["image"]["url"].endswith("hello.thumbnail.jpg")
 160
 161     def test_can_post_image_tags(self, test_app):
 162         """ Tests that an image can be posted to the API """
 163         # First request we need to do is to upload the image
 164         response, image = self._upload_image(test_app, GOOD_JPG)
 165         assert response.status_code == 200
 166
 167         image["tags"] = ["hello", "world"]
 168
 169         # Check that we got the response we're expecting
 170         response, data = self._post_image_to_feed(test_app, image)
 171         assert response.status_code == 200
 172         assert data["object"]["tags"] == ["hello", "world"]
 173
 174     def test_unable_to_upload_as_someone_else(self, test_app):
 175         """ Test that can't upload as someoen else """
 176         data = open(GOOD_JPG, "rb").read()
 177         headers = {
 178             "Content-Type": "image/jpeg",
 179             "Content-Length": str(len(data))
 180         }
 181
 182         with self.mock_oauth():
 183             # Will be self.user trying to upload as self.other_user
 184             with pytest.raises(AppError) as excinfo:
 185                 test_app.post(
 186                     "/api/user/{0}/uploads".format(self.other_user.username),
 187                     data,
 188                     headers=headers
 189                 )
 190
 191             assert "403 FORBIDDEN" in excinfo.value.args[0]
 192
 193     def test_unable_to_post_feed_as_someone_else(self, test_app):
 194         """ Tests that can't post an image to someone else's feed """
 195         response, data = self._upload_image(test_app, GOOD_JPG)
 196
 197         activity = {
 198             "verb": "post",
 199             "object": data
 200         }
 201
 202         headers = {
 203             "Content-Type": "application/json",
 204         }
 205
 206         with self.mock_oauth():
 207             with pytest.raises(AppError) as excinfo:
 208                 test_app.post(
 209                     "/api/user/{0}/feed".format(self.other_user.username),
 210                     json.dumps(activity),
 211                     headers=headers
 212                 )
 213
 214             assert "403 FORBIDDEN" in excinfo.value.args[0]
 215
 216     def test_only_able_to_update_own_image(self, test_app):
 217         """ Test uploader is the only person who can update an image """
 218         response, data = self._upload_image(test_app, GOOD_JPG)
 219         response, data = self._post_image_to_feed(test_app, data)
 220
 221         activity = {
 222             "verb": "update",
 223             "object": data["object"],
 224         }
 225
 226         headers = {
 227             "Content-Type": "application/json",
 228         }
 229
 230         # Lets change the image uploader to be self.other_user, this is easier
 231         # than uploading the image as someone else as the way
 232         # self.mocked_oauth_required and self._upload_image.
 233         media = MediaEntry.query \
 234             .filter_by(public_id=data["object"]["id"]) \
 235             .first()
 236         media.actor = self.other_user.id
 237         media.save()
 238
 239         # Now lets try and edit the image as self.user, this should produce a
 240         # 403 error.
 241         with self.mock_oauth():
 242             with pytest.raises(AppError) as excinfo:
 243                 test_app.post(
 244                     "/api/user/{0}/feed".format(self.user.username),
 245                     json.dumps(activity),
 246                     headers=headers
 247                 )
 248
 249             assert "403 FORBIDDEN" in excinfo.value.args[0]
 250
 251     def test_upload_image_with_filename(self, test_app):
 252         """ Tests that you can upload an image with filename and description """
 253         response, data = self._upload_image(test_app, GOOD_JPG)
 254         response, data = self._post_image_to_feed(test_app, data)
 255
 256         image = data["object"]
 257
 258         # Now we need to add a title and description
 259         title = "My image ^_^"
 260         description = "This is my super awesome image :D"
 261         license = "CC-BY-SA"
 262
 263         image["displayName"] = title
 264         image["content"] = description
 265         image["license"] = license
 266
 267         activity = {"verb": "update", "object": image}
 268
 269         with self.mock_oauth():
 270             response = test_app.post(
 271                 "/api/user/{0}/feed".format(self.user.username),
 272                 json.dumps(activity),
 273                 headers={"Content-Type": "application/json"}
 274             )
 275
 276         image = json.loads(response.body.decode())["object"]
 277
 278         # Check everything has been set on the media correctly
 279         media = MediaEntry.query.filter_by(public_id=image["id"]).first()
 280         assert media.title == title
 281         assert media.description == description
 282         assert media.license == license
 283
 284         # Check we're being given back everything we should on an update
 285         assert image["id"] == media.public_id
 286         assert image["displayName"] == title
 287         assert image["content"] == description
 288         assert image["license"] == license
 289
 290     def test_only_uploaders_post_image(self, test_app):
 291         """ Test that only uploaders can upload images """
 292         # Remove uploader permissions from user
 293         take_away_privileges(self.user.username, u"uploader")
 294
 295         # Now try and upload a image
 296         data = open(GOOD_JPG, "rb").read()
 297         headers = {
 298             "Content-Type": "image/jpeg",
 299             "Content-Length": str(len(data)),
 300         }
 301
 302         with self.mock_oauth():
 303             with pytest.raises(AppError) as excinfo:
 304                 test_app.post(
 305                     "/api/user/{0}/uploads".format(self.user.username),
 306                     data,
 307                     headers=headers
 308                 )
 309
 310             # Assert that we've got a 403
 311             assert "403 FORBIDDEN" in excinfo.value.args[0]
 312
 313     def test_object_endpoint(self, test_app):
 314         """ Tests that object can be looked up at endpoint """
 315         # Post an image
 316         response, data = self._upload_image(test_app, GOOD_JPG)
 317         response, data = self._post_image_to_feed(test_app, data)
 318
 319         # Now lookup image to check that endpoint works.
 320         image = data["object"]
 321
 322         assert "links" in image
 323         assert "self" in image["links"]
 324
 325         # Get URI and strip testing host off
 326         object_uri = image["links"]["self"]["href"]
 327         object_uri = object_uri.replace("http://localhost:80", "")
 328
 329         with self.mock_oauth():
 330             request = test_app.get(object_uri)
 331
 332         image = json.loads(request.body.decode())
 333         entry = MediaEntry.query.filter_by(public_id=image["id"]).first()
 334
 335         assert entry is not None
 336
 337         assert request.status_code == 200
 338
 339         assert "image" in image
 340         assert "fullImage" in image
 341         assert "pump_io" in image
 342         assert "links" in image
 343         assert "tags" in image
 344
 345     def test_post_comment(self, test_app):
 346         """ Tests that I can post an comment media """
 347         # Upload some media to comment on
 348         response, data = self._upload_image(test_app, GOOD_JPG)
 349         response, data = self._post_image_to_feed(test_app, data)
 350
 351         content = "Hai this is a comment on this lovely picture ^_^"
 352
 353         activity = {
 354             "verb": "post",
 355             "object": {
 356                 "objectType": "comment",
 357                 "content": content,
 358                 "inReplyTo": data["object"],
 359             }
 360         }
 361
 362         response, comment_data = self._activity_to_feed(test_app, activity)
 363         assert response.status_code == 200
 364
 365         # Find the objects in the database
 366         media = MediaEntry.query \
 367             .filter_by(public_id=data["object"]["id"]) \
 368             .first()
 369         comment = media.get_comments()[0].comment()
 370
 371         # Tests that it matches in the database
 372         assert comment.actor == self.user.id
 373         assert comment.content == content
 374
 375         # Test that the response is what we should be given
 376         assert comment.content == comment_data["object"]["content"]
 377
 378     def test_unable_to_post_comment_as_someone_else(self, test_app):
 379         """ Tests that you're unable to post a comment as someone else. """
 380         # Upload some media to comment on
 381         response, data = self._upload_image(test_app, GOOD_JPG)
 382         response, data = self._post_image_to_feed(test_app, data)
 383
 384         activity = {
 385             "verb": "post",
 386             "object": {
 387                 "objectType": "comment",
 388                 "content": "comment commenty comment ^_^",
 389                 "inReplyTo": data["object"],
 390             }
 391         }
 392
 393         headers = {
 394             "Content-Type": "application/json",
 395         }
 396
 397         with self.mock_oauth():
 398             with pytest.raises(AppError) as excinfo:
 399                 test_app.post(
 400                     "/api/user/{0}/feed".format(self.other_user.username),
 401                     json.dumps(activity),
 402                     headers=headers
 403                 )
 404
 405             assert "403 FORBIDDEN" in excinfo.value.args[0]
 406
 407     def test_unable_to_update_someone_elses_comment(self, test_app):
 408         """ Test that you're able to update someoen elses comment. """
 409         # Upload some media to comment on
 410         response, data = self._upload_image(test_app, GOOD_JPG)
 411         response, data = self._post_image_to_feed(test_app, data)
 412
 413         activity = {
 414             "verb": "post",
 415             "object": {
 416                 "objectType": "comment",
 417                 "content": "comment commenty comment ^_^",
 418                 "inReplyTo": data["object"],
 419             }
 420         }
 421
 422         headers = {
 423             "Content-Type": "application/json",
 424         }
 425
 426         # Post the comment.
 427         response, comment_data = self._activity_to_feed(test_app, activity)
 428
 429         # change who uploaded the comment as it's easier than changing
 430         comment = TextComment.query \
 431             .filter_by(public_id=comment_data["object"]["id"]) \
 432             .first()
 433         comment.actor = self.other_user.id
 434         comment.save()
 435
 436         # Update the comment as someone else.
 437         comment_data["object"]["content"] = "Yep"
 438         activity = {
 439             "verb": "update",
 440             "object": comment_data["object"]
 441         }
 442
 443         with self.mock_oauth():
 444             with pytest.raises(AppError) as excinfo:
 445                 test_app.post(
 446                     "/api/user/{0}/feed".format(self.user.username),
 447                     json.dumps(activity),
 448                     headers=headers
 449                 )
 450
 451             assert "403 FORBIDDEN" in excinfo.value.args[0]
 452
 453     def test_profile(self, test_app):
 454         """ Tests profile endpoint """
 455         uri = "/api/user/{0}/profile".format(self.user.username)
 456         with self.mock_oauth():
 457             response = test_app.get(uri)
 458             profile = json.loads(response.body.decode())
 459
 460             assert response.status_code == 200
 461
 462             assert profile["preferredUsername"] == self.user.username
 463             assert profile["objectType"] == "person"
 464
 465             assert "links" in profile
 466
 467     def test_user(self, test_app):
 468         """ Test the user endpoint """
 469         uri = "/api/user/{0}/".format(self.user.username)
 470         with self.mock_oauth():
 471             response = test_app.get(uri)
 472             user = json.loads(response.body.decode())
 473
 474             assert response.status_code == 200
 475
 476             assert user["nickname"] == self.user.username
 477             assert user["updated"] == self.user.created.isoformat()
 478             assert user["published"] == self.user.created.isoformat()
 479
 480             # Test profile exists but self.test_profile will test the value
 481             assert "profile" in response
 482
 483     def test_whoami_without_login(self, test_app):
 484         """ Test that whoami endpoint returns error when not logged in """
 485         with pytest.raises(AppError) as excinfo:
 486             test_app.get("/api/whoami")
 487
 488         assert "401 UNAUTHORIZED" in excinfo.value.args[0]
 489
 490     def test_read_feed(self, test_app):
 491         """ Test able to read objects from the feed """
 492         response, image_data = self._upload_image(test_app, GOOD_JPG)
 493         response, data = self._post_image_to_feed(test_app, image_data)
 494
 495         uri = "/api/user/{0}/feed".format(self.active_user.username)
 496         with self.mock_oauth():
 497             response = test_app.get(uri)
 498             feed = json.loads(response.body.decode())
 499
 500             assert response.status_code == 200
 501
 502             # Check it has the attributes it should
 503             assert "displayName" in feed
 504             assert "objectTypes" in feed
 505             assert "url" in feed
 506             assert "links" in feed
 507             assert "author" in feed
 508             assert "items" in feed
 509
 510             # Check that image i uploaded is there
 511             assert feed["items"][0]["verb"] == "post"
 512             assert feed["items"][0]["id"] == data["id"]
 513             assert feed["items"][0]["object"]["objectType"] == "image"
 514             assert feed["items"][0]["object"]["id"] == data["object"]["id"]
 515
 516         default_limit = 20
 517         items_count = default_limit * 2
 518         for i in range(items_count):
 519             response, image_data = self._upload_image(test_app, GOOD_JPG)
 520             self._post_image_to_feed(test_app, image_data)
 521         items_count += 1  # because there already is one
 522
 523         #
 524         # default returns default_limit items
 525         #
 526         with self.mock_oauth():
 527             response = test_app.get(uri)
 528             feed = json.loads(response.body.decode())
 529             assert len(feed["items"]) == default_limit
 530
 531         #
 532         # silentely ignore count and offset that that are
 533         # not a number
 534         #
 535         with self.mock_oauth():
 536             response = test_app.get(uri + "?count=BAD&offset=WORSE")
 537             feed = json.loads(response.body.decode())
 538             assert len(feed["items"]) == default_limit
 539
 540         #
 541         # if offset is less than default_limit items
 542         # from the end of the feed, return less than
 543         # default_limit
 544         #
 545         with self.mock_oauth():
 546             near_the_end = items_count - default_limit / 2
 547             response = test_app.get(uri + "?offset=%d" % near_the_end)
 548             feed = json.loads(response.body.decode())
 549             assert len(feed["items"]) < default_limit
 550
 551         #
 552         # count=5 returns 5 items
 553         #
 554         with self.mock_oauth():
 555             response = test_app.get(uri + "?count=5")
 556             feed = json.loads(response.body.decode())
 557             assert len(feed["items"]) == 5
 558
 559     def test_read_another_feed(self, test_app):
 560         """ Test able to read objects from someone else's feed """
 561         response, data = self._upload_image(test_app, GOOD_JPG)
 562         response, data = self._post_image_to_feed(test_app, data)
 563
 564         # Change the active user to someone else.
 565         self.active_user = self.other_user
 566
 567         # Fetch the feed
 568         url = "/api/user/{0}/feed".format(self.user.username)
 569         with self.mock_oauth():
 570             response = test_app.get(url)
 571             feed = json.loads(response.body.decode())
 572
 573             assert response.status_code == 200
 574
 575             # Check it has the attributes it ought to.
 576             assert "displayName" in feed
 577             assert "objectTypes" in feed
 578             assert "url" in feed
 579             assert "links" in feed
 580             assert "author" in feed
 581             assert "items" in feed
 582
 583             # Assert the uploaded image is there
 584             assert feed["items"][0]["verb"] == "post"
 585             assert feed["items"][0]["id"] == data["id"]
 586             assert feed["items"][0]["object"]["objectType"] == "image"
 587             assert feed["items"][0]["object"]["id"] == data["object"]["id"]
 588
 589     def test_cant_post_to_someone_elses_feed(self, test_app):
 590         """ Test that can't post to someone elses feed """
 591         response, data = self._upload_image(test_app, GOOD_JPG)
 592         self.active_user = self.other_user
 593
 594         with self.mock_oauth():
 595             with pytest.raises(AppError) as excinfo:
 596                 self._post_image_to_feed(test_app, data)
 597
 598             assert "403 FORBIDDEN" in excinfo.value.args[0]
 599
 600     def test_object_endpoint_requestable(self, test_app):
 601         """ Test that object endpoint can be requested """
 602         response, data = self._upload_image(test_app, GOOD_JPG)
 603         response, data = self._post_image_to_feed(test_app, data)
 604         object_id = data["object"]["id"]
 605
 606         with self.mock_oauth():
 607             response = test_app.get(data["object"]["links"]["self"]["href"])
 608             data = json.loads(response.body.decode())
 609
 610             assert response.status_code == 200
 611
 612             assert object_id == data["id"]
 613             assert "url" in data
 614             assert "links" in data
 615             assert data["objectType"] == "image"
 616
 617     def test_delete_media_by_activity(self, test_app):
 618         """ Test that an image can be deleted by a delete activity to feed """
 619         response, data = self._upload_image(test_app, GOOD_JPG)
 620         response, data = self._post_image_to_feed(test_app, data)
 621         object_id = data["object"]["id"]
 622
 623         activity = {
 624             "verb": "delete",
 625             "object": {
 626                 "id": object_id,
 627                 "objectType": "image",
 628             }
 629         }
 630
 631         response = self._activity_to_feed(test_app, activity)[1]
 632
 633         # Check the media is no longer in the database
 634         media = MediaEntry.query.filter_by(public_id=object_id).first()
 635
 636         assert media is None
 637
 638         # Check we've been given the full delete activity back
 639         assert "id" in response
 640         assert response["verb"] == "delete"
 641         assert "object" in response
 642         assert response["object"]["id"] == object_id
 643         assert response["object"]["objectType"] == "image"
 644
 645     def test_delete_comment_by_activity(self, test_app):
 646         """ Test that a comment is deleted by a delete activity to feed """
 647         # First upload an image to comment against
 648         response, data = self._upload_image(test_app, GOOD_JPG)
 649         response, data = self._post_image_to_feed(test_app, data)
 650
 651         # Post a comment to delete
 652         activity = {
 653             "verb": "post",
 654             "object": {
 655                 "objectType": "comment",
 656                 "content": "This is a comment.",
 657                 "inReplyTo": data["object"],
 658             }
 659         }
 660
 661         comment = self._activity_to_feed(test_app, activity)[1]
 662
 663         # Now delete the image
 664         activity = {
 665             "verb": "delete",
 666             "object": {
 667                 "id": comment["object"]["id"],
 668                 "objectType": "comment",
 669             }
 670         }
 671
 672         delete = self._activity_to_feed(test_app, activity)[1]
 673
 674         # Verify the comment no longer exists
 675         assert TextComment.query \
 676             .filter_by(public_id=comment["object"]["id"]) \
 677             .first() is None
 678
 679         assert "id" in comment["object"]
 680
 681         # Check we've got a delete activity back
 682         assert "id" in delete
 683         assert delete["verb"] == "delete"
 684         assert "object" in delete
 685         assert delete["object"]["id"] == comment["object"]["id"]
 686         assert delete["object"]["objectType"] == "comment"
 687
 688     def test_edit_comment(self, test_app):
 689         """ Test that someone can update their own comment """
 690         # First upload an image to comment against
 691         response, data = self._upload_image(test_app, GOOD_JPG)
 692         response, data = self._post_image_to_feed(test_app, data)
 693
 694         # Post a comment to edit
 695         activity = {
 696             "verb": "post",
 697             "object": {
 698                 "objectType": "comment",
 699                 "content": "This is a comment",
 700                 "inReplyTo": data["object"],
 701             }
 702         }
 703
 704         comment = self._activity_to_feed(test_app, activity)[1]
 705
 706         # Now create an update activity to change the content
 707         activity = {
 708             "verb": "update",
 709             "object": {
 710                 "id": comment["object"]["id"],
 711                 "content": "This is my fancy new content string!",
 712                 "objectType": "comment",
 713             },
 714         }
 715
 716         comment = self._activity_to_feed(test_app, activity)[1]
 717
 718         # Verify the comment reflects the changes
 719         model = TextComment.query \
 720             .filter_by(public_id=comment["object"]["id"]) \
 721             .first()
 722
 723         assert model.content == activity["object"]["content"]