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, Activity, MediaEntry, TextComment
  29 from mediagoblin.tools.routing import extract_url_arguments
  30 from mediagoblin.tests.tools import fixture_add_user
  31 from mediagoblin.moderation.tools import take_away_privileges
  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', u'commenter'])
  42         self.other_user = fixture_add_user(
  43             username="otheruser",
  44             privileges=[u'active', u'uploader', u'commenter']
  45         )
  46         self.active_user = self.user
  47
  48     def _activity_to_feed(self, test_app, activity, headers=None):
  49         """ Posts an activity to the user's feed """
  50         if headers:
  51             headers.setdefault("Content-Type", "application/json")
  52         else:
  53             headers = {"Content-Type": "application/json"}
  54
  55         with self.mock_oauth():
  56             response = test_app.post(
  57                 "/api/user/{0}/feed".format(self.active_user.username),
  58                 json.dumps(activity),
  59                 headers=headers
  60             )
  61
  62         return response, json.loads(response.body.decode())
  63
  64     def _upload_image(self, test_app, image):
  65         """ Uploads and image to MediaGoblin via pump.io API """
  66         data = open(image, "rb").read()
  67         headers = {
  68             "Content-Type": "image/jpeg",
  69             "Content-Length": str(len(data))
  70         }
  71
  72
  73         with self.mock_oauth():
  74             response = test_app.post(
  75                 "/api/user/{0}/uploads".format(self.active_user.username),
  76                 data,
  77                 headers=headers
  78             )
  79             image = json.loads(response.body.decode())
  80
  81         return response, image
  82
  83     def _post_image_to_feed(self, test_app, image):
  84         """ Posts an already uploaded image to feed """
  85         activity = {
  86             "verb": "post",
  87             "object": image,
  88         }
  89
  90         return self._activity_to_feed(test_app, activity)
  91
  92     def mocked_oauth_required(self, *args, **kwargs):
  93         """ Mocks mediagoblin.decorator.oauth_required to always validate """
  94
  95         def fake_controller(controller, request, *args, **kwargs):
  96             request.user = User.query.filter_by(id=self.active_user.id).first()
  97             return controller(request, *args, **kwargs)
  98
  99         def oauth_required(c):
 100             return lambda *args, **kwargs: fake_controller(c, *args, **kwargs)
 101
 102         return oauth_required
 103
 104     def mock_oauth(self):
 105         """ Returns a mock.patch for the oauth_required decorator """
 106         return mock.patch(
 107             target="mediagoblin.decorators.oauth_required",
 108             new_callable=self.mocked_oauth_required
 109         )
 110
 111     def test_can_post_image(self, test_app):
 112         """ Tests that an image can be posted to the API """
 113         # First request we need to do is to upload the image
 114         response, image = self._upload_image(test_app, GOOD_JPG)
 115
 116         # I should have got certain things back
 117         assert response.status_code == 200
 118
 119         assert "id" in image
 120         assert "fullImage" in image
 121         assert "url" in image["fullImage"]
 122         assert "url" in image
 123         assert "author" in image
 124         assert "published" in image
 125         assert "updated" in image
 126         assert image["objectType"] == "image"
 127
 128         # Check that we got the response we're expecting
 129         response, _ = self._post_image_to_feed(test_app, image)
 130         assert response.status_code == 200
 131
 132     def test_unable_to_upload_as_someone_else(self, test_app):
 133         """ Test that can't upload as someoen else """
 134         data = open(GOOD_JPG, "rb").read()
 135         headers = {
 136             "Content-Type": "image/jpeg",
 137             "Content-Length": str(len(data))
 138         }
 139
 140         with self.mock_oauth():
 141             # Will be self.user trying to upload as self.other_user
 142             with pytest.raises(AppError) as excinfo:
 143                 test_app.post(
 144                     "/api/user/{0}/uploads".format(self.other_user.username),
 145                     data,
 146                     headers=headers
 147                 )
 148
 149             assert "403 FORBIDDEN" in excinfo.value.args[0]
 150
 151     def test_unable_to_post_feed_as_someone_else(self, test_app):
 152         """ Tests that can't post an image to someone else's feed """
 153         response, data = self._upload_image(test_app, GOOD_JPG)
 154
 155         activity = {
 156             "verb": "post",
 157             "object": data
 158         }
 159
 160         headers = {
 161             "Content-Type": "application/json",
 162         }
 163
 164         with self.mock_oauth():
 165             with pytest.raises(AppError) as excinfo:
 166                 test_app.post(
 167                     "/api/user/{0}/feed".format(self.other_user.username),
 168                     json.dumps(activity),
 169                     headers=headers
 170                 )
 171
 172             assert "403 FORBIDDEN" in excinfo.value.args[0]
 173
 174     def test_only_able_to_update_own_image(self, test_app):
 175         """ Test's that the uploader is the only person who can update an image """
 176         response, data = self._upload_image(test_app, GOOD_JPG)
 177         response, data = self._post_image_to_feed(test_app, data)
 178
 179         activity = {
 180             "verb": "update",
 181             "object": data["object"],
 182         }
 183
 184         headers = {
 185             "Content-Type": "application/json",
 186         }
 187
 188         # Lets change the image uploader to be self.other_user, this is easier
 189         # than uploading the image as someone else as the way self.mocked_oauth_required
 190         # and self._upload_image.
 191         media = MediaEntry.query.filter_by(public_id=data["object"]["id"]).first()
 192         media.actor = self.other_user.id
 193         media.save()
 194
 195         # Now lets try and edit the image as self.user, this should produce a 403 error.
 196         with self.mock_oauth():
 197             with pytest.raises(AppError) as excinfo:
 198                 test_app.post(
 199                     "/api/user/{0}/feed".format(self.user.username),
 200                     json.dumps(activity),
 201                     headers=headers
 202                 )
 203
 204             assert "403 FORBIDDEN" in excinfo.value.args[0]
 205
 206     def test_upload_image_with_filename(self, test_app):
 207         """ Tests that you can upload an image with filename and description """
 208         response, data = self._upload_image(test_app, GOOD_JPG)
 209         response, data = self._post_image_to_feed(test_app, data)
 210
 211         image = data["object"]
 212
 213         # Now we need to add a title and description
 214         title = "My image ^_^"
 215         description = "This is my super awesome image :D"
 216         license = "CC-BY-SA"
 217
 218         image["displayName"] = title
 219         image["content"] = description
 220         image["license"] = license
 221
 222         activity = {"verb": "update", "object": image}
 223
 224         with self.mock_oauth():
 225             response = test_app.post(
 226                 "/api/user/{0}/feed".format(self.user.username),
 227                 json.dumps(activity),
 228                 headers={"Content-Type": "application/json"}
 229             )
 230
 231         image = json.loads(response.body.decode())["object"]
 232
 233         # Check everything has been set on the media correctly
 234         media = MediaEntry.query.filter_by(public_id=image["id"]).first()
 235         assert media.title == title
 236         assert media.description == description
 237         assert media.license == license
 238
 239         # Check we're being given back everything we should on an update
 240         assert image["id"] == media.public_id
 241         assert image["displayName"] == title
 242         assert image["content"] == description
 243         assert image["license"] == license
 244
 245
 246     def test_only_uploaders_post_image(self, test_app):
 247         """ Test that only uploaders can upload images """
 248         # Remove uploader permissions from user
 249         take_away_privileges(self.user.username, u"uploader")
 250
 251         # Now try and upload a image
 252         data = open(GOOD_JPG, "rb").read()
 253         headers = {
 254             "Content-Type": "image/jpeg",
 255             "Content-Length": str(len(data)),
 256         }
 257
 258         with self.mock_oauth():
 259             with pytest.raises(AppError) as excinfo:
 260                 test_app.post(
 261                     "/api/user/{0}/uploads".format(self.user.username),
 262                     data,
 263                     headers=headers
 264                 )
 265
 266             # Assert that we've got a 403
 267             assert "403 FORBIDDEN" in excinfo.value.args[0]
 268
 269     def test_object_endpoint(self, test_app):
 270         """ Tests that object can be looked up at endpoint """
 271         # Post an image
 272         response, data = self._upload_image(test_app, GOOD_JPG)
 273         response, data = self._post_image_to_feed(test_app, data)
 274
 275         # Now lookup image to check that endpoint works.
 276         image = data["object"]
 277
 278         assert "links" in image
 279         assert "self" in image["links"]
 280
 281         # Get URI and strip testing host off
 282         object_uri = image["links"]["self"]["href"]
 283         object_uri = object_uri.replace("http://localhost:80", "")
 284
 285         with self.mock_oauth():
 286             request = test_app.get(object_uri)
 287
 288         image = json.loads(request.body.decode())
 289         entry = MediaEntry.query.filter_by(public_id=image["id"]).first()
 290
 291         assert request.status_code == 200
 292
 293         assert "image" in image
 294         assert "fullImage" in image
 295         assert "pump_io" in image
 296         assert "links" in image
 297
 298     def test_post_comment(self, test_app):
 299         """ Tests that I can post an comment media """
 300         # Upload some media to comment on
 301         response, data = self._upload_image(test_app, GOOD_JPG)
 302         response, data = self._post_image_to_feed(test_app, data)
 303
 304         content = "Hai this is a comment on this lovely picture ^_^"
 305
 306         activity = {
 307             "verb": "post",
 308             "object": {
 309                 "objectType": "comment",
 310                 "content": content,
 311                 "inReplyTo": data["object"],
 312             }
 313         }
 314
 315         response, comment_data = self._activity_to_feed(test_app, activity)
 316         assert response.status_code == 200
 317
 318         # Find the objects in the database
 319         media = MediaEntry.query.filter_by(public_id=data["object"]["id"]).first()
 320         comment = media.get_comments()[0]
 321
 322         # Tests that it matches in the database
 323         assert comment.actor == self.user.id
 324         assert comment.content == content
 325
 326         # Test that the response is what we should be given
 327         assert comment.content == comment_data["object"]["content"]
 328
 329     def test_unable_to_post_comment_as_someone_else(self, test_app):
 330         """ Tests that you're unable to post a comment as someone else. """
 331         # Upload some media to comment on
 332         response, data = self._upload_image(test_app, GOOD_JPG)
 333         response, data = self._post_image_to_feed(test_app, data)
 334
 335         activity = {
 336             "verb": "post",
 337             "object": {
 338                 "objectType": "comment",
 339                 "content": "comment commenty comment ^_^",
 340                 "inReplyTo": data["object"],
 341             }
 342         }
 343
 344         headers = {
 345             "Content-Type": "application/json",
 346         }
 347
 348         with self.mock_oauth():
 349             with pytest.raises(AppError) as excinfo:
 350                 test_app.post(
 351                     "/api/user/{0}/feed".format(self.other_user.username),
 352                     json.dumps(activity),
 353                     headers=headers
 354                 )
 355
 356             assert "403 FORBIDDEN" in excinfo.value.args[0]
 357
 358     def test_unable_to_update_someone_elses_comment(self, test_app):
 359         """ Test that you're able to update someoen elses comment. """
 360         # Upload some media to comment on
 361         response, data = self._upload_image(test_app, GOOD_JPG)
 362         response, data = self._post_image_to_feed(test_app, data)
 363
 364         activity = {
 365             "verb": "post",
 366             "object": {
 367                 "objectType": "comment",
 368                 "content": "comment commenty comment ^_^",
 369                 "inReplyTo": data["object"],
 370             }
 371         }
 372
 373         headers = {
 374             "Content-Type": "application/json",
 375         }
 376
 377         # Post the comment.
 378         response, comment_data = self._activity_to_feed(test_app, activity)
 379
 380         # change who uploaded the comment as it's easier than changing
 381         comment = TextComment.query.filter_by(public_id=comment_data["object"]["id"]).first()
 382         comment.actor = self.other_user.id
 383         comment.save()
 384
 385         # Update the comment as someone else.
 386         comment_data["object"]["content"] = "Yep"
 387         activity = {
 388             "verb": "update",
 389             "object": comment_data["object"]
 390         }
 391
 392         with self.mock_oauth():
 393             with pytest.raises(AppError) as excinfo:
 394                 test_app.post(
 395                     "/api/user/{0}/feed".format(self.user.username),
 396                     json.dumps(activity),
 397                     headers=headers
 398                 )
 399
 400             assert "403 FORBIDDEN" in excinfo.value.args[0]
 401
 402     def test_profile(self, test_app):
 403         """ Tests profile endpoint """
 404         uri = "/api/user/{0}/profile".format(self.user.username)
 405         with self.mock_oauth():
 406             response = test_app.get(uri)
 407             profile = json.loads(response.body.decode())
 408
 409             assert response.status_code == 200
 410
 411             assert profile["preferredUsername"] == self.user.username
 412             assert profile["objectType"] == "person"
 413
 414             assert "links" in profile
 415
 416     def test_user(self, test_app):
 417         """ Test the user endpoint """
 418         uri = "/api/user/{0}/".format(self.user.username)
 419         with self.mock_oauth():
 420             response = test_app.get(uri)
 421             user = json.loads(response.body.decode())
 422
 423             assert response.status_code == 200
 424
 425             assert user["nickname"] == self.user.username
 426             assert user["updated"] == self.user.created.isoformat()
 427             assert user["published"] == self.user.created.isoformat()
 428
 429             # Test profile exists but self.test_profile will test the value
 430             assert "profile" in response
 431
 432     def test_whoami_without_login(self, test_app):
 433         """ Test that whoami endpoint returns error when not logged in """
 434         with pytest.raises(AppError) as excinfo:
 435             response = test_app.get("/api/whoami")
 436
 437         assert "401 UNAUTHORIZED" in excinfo.value.args[0]
 438
 439     def test_read_feed(self, test_app):
 440         """ Test able to read objects from the feed """
 441         response, data = self._upload_image(test_app, GOOD_JPG)
 442         response, data = self._post_image_to_feed(test_app, data)
 443
 444         uri = "/api/user/{0}/feed".format(self.active_user.username)
 445         with self.mock_oauth():
 446             response = test_app.get(uri)
 447             feed = json.loads(response.body.decode())
 448
 449             assert response.status_code == 200
 450
 451             # Check it has the attributes it should
 452             assert "displayName" in feed
 453             assert "objectTypes" in feed
 454             assert "url" in feed
 455             assert "links" in feed
 456             assert "author" in feed
 457             assert "items" in feed
 458
 459             # Check that image i uploaded is there
 460             assert feed["items"][0]["verb"] == "post"
 461             assert feed["items"][0]["id"] == data["id"]
 462             assert feed["items"][0]["object"]["objectType"] == "image"
 463             assert feed["items"][0]["object"]["id"] == data["object"]["id"]
 464
 465
 466     def test_read_another_feed(self, test_app):
 467         """ Test able to read objects from someone else's feed """
 468         response, data = self._upload_image(test_app, GOOD_JPG)
 469         response, data = self._post_image_to_feed(test_app, data)
 470
 471         # Change the active user to someone else.
 472         self.active_user = self.other_user
 473
 474         # Fetch the feed
 475         url = "/api/user/{0}/feed".format(self.user.username)
 476         with self.mock_oauth():
 477             response = test_app.get(url)
 478             feed = json.loads(response.body.decode())
 479
 480             assert response.status_code == 200
 481
 482             # Check it has the attributes it ought to.
 483             assert "displayName" in feed
 484             assert "objectTypes" in feed
 485             assert "url" in feed
 486             assert "links" in feed
 487             assert "author" in feed
 488             assert "items" in feed
 489
 490             # Assert the uploaded image is there
 491             assert feed["items"][0]["verb"] == "post"
 492             assert feed["items"][0]["id"] == data["id"]
 493             assert feed["items"][0]["object"]["objectType"] == "image"
 494             assert feed["items"][0]["object"]["id"] == data["object"]["id"]
 495
 496     def test_cant_post_to_someone_elses_feed(self, test_app):
 497         """ Test that can't post to someone elses feed """
 498         response, data = self._upload_image(test_app, GOOD_JPG)
 499         self.active_user = self.other_user
 500
 501         with self.mock_oauth():
 502             with pytest.raises(AppError) as excinfo:
 503                 self._post_image_to_feed(test_app, data)
 504
 505             assert "403 FORBIDDEN" in excinfo.value.args[0]
 506
 507     def test_object_endpoint_requestable(self, test_app):
 508         """ Test that object endpoint can be requested """
 509         response, data = self._upload_image(test_app, GOOD_JPG)
 510         response, data = self._post_image_to_feed(test_app, data)
 511         object_id = data["object"]["id"]
 512
 513         with self.mock_oauth():
 514             response = test_app.get(data["object"]["links"]["self"]["href"])
 515             data = json.loads(response.body.decode())
 516
 517             assert response.status_code == 200
 518
 519             assert object_id == data["id"]
 520             assert "url" in data
 521             assert "links" in data
 522             assert data["objectType"] == "image"
 523
 524     def test_delete_media_by_activity(self, test_app):
 525         """ Test that an image can be deleted by a delete activity to feed """
 526         response, data = self._upload_image(test_app, GOOD_JPG)
 527         response, data = self._post_image_to_feed(test_app, data)
 528         object_id = data["object"]["id"]
 529
 530         activity = {
 531             "verb": "delete",
 532             "object": {
 533                 "id": object_id,
 534                 "objectType": "image",
 535             }
 536         }
 537
 538         response = self._activity_to_feed(test_app, activity)[1]
 539
 540         # Check the media is no longer in the database
 541         media = MediaEntry.query.filter_by(public_id=object_id).first()
 542
 543         assert media is None
 544
 545         # Check we've been given the full delete activity back
 546         assert "id" in response
 547         assert response["verb"] == "delete"
 548         assert "object" in response
 549         assert response["object"]["id"] == object_id
 550         assert response["object"]["objectType"] == "image"
 551
 552     def test_delete_comment_by_activity(self, test_app):
 553         """ Test that a comment is deleted by a delete activity to feed """
 554         # First upload an image to comment against
 555         response, data = self._upload_image(test_app, GOOD_JPG)
 556         response, data = self._post_image_to_feed(test_app, data)
 557
 558         # Post a comment to delete
 559         activity = {
 560             "verb": "post",
 561             "object": {
 562                 "objectType": "comment",
 563                 "content": "This is a comment.",
 564                 "inReplyTo": data["object"],
 565             }
 566         }
 567
 568         comment = self._activity_to_feed(test_app, activity)[1]
 569
 570         # Now delete the image
 571         activity = {
 572             "verb": "delete",
 573             "object": {
 574                 "id": comment["object"]["id"],
 575                 "objectType": "comment",
 576             }
 577         }
 578
 579         delete = self._activity_to_feed(test_app, activity)[1]
 580
 581         # Verify the comment no longer exists
 582         assert TextComment.query.filter_by(public_id=comment["object"]["id"]).first() is None
 583         comment_id = comment["object"]["id"]
 584
 585         # Check we've got a delete activity back
 586         assert "id" in delete
 587         assert delete["verb"] == "delete"
 588         assert "object" in delete
 589         assert delete["object"]["id"] == comment["object"]["id"]
 590         assert delete["object"]["objectType"] == "comment"
 591
 592     def test_edit_comment(self, test_app):
 593         """ Test that someone can update their own comment """
 594         # First upload an image to comment against
 595         response, data = self._upload_image(test_app, GOOD_JPG)
 596         response, data = self._post_image_to_feed(test_app, data)
 597
 598         # Post a comment to edit
 599         activity = {
 600             "verb": "post",
 601             "object": {
 602                 "objectType": "comment",
 603                 "content": "This is a comment",
 604                 "inReplyTo": data["object"],
 605             }
 606         }
 607
 608         comment = self._activity_to_feed(test_app, activity)[1]
 609
 610         # Now create an update activity to change the content
 611         activity = {
 612             "verb": "update",
 613             "object": {
 614                 "id": comment["object"]["id"],
 615                 "content": "This is my fancy new content string!",
 616                 "objectType": "comment",
 617             },
 618         }
 619
 620         comment = self._activity_to_feed(test_app, activity)[1]
 621
 622         # Verify the comment reflects the changes
 623         model = TextComment.query.filter_by(public_id=comment["object"]["id"]).first()
 624
 625         assert model.content == activity["object"]["content"]