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.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]["actor"]
 462
 463     def test_cant_post_to_someone_elses_feed(self, test_app):
 464         """ Test that can't post to someone elses feed """
 465         response, data = self._upload_image(test_app, GOOD_JPG)
 466         self.active_user = self.other_user
 467
 468         with self.mock_oauth():
 469             with pytest.raises(AppError) as excinfo:
 470                 self._post_image_to_feed(test_app, data)
 471
 472             assert "403 FORBIDDEN" in excinfo.value.args[0]
 473
 474     def test_object_endpoint_requestable(self, test_app):
 475         """ Test that object endpoint can be requested """
 476         response, data = self._upload_image(test_app, GOOD_JPG)
 477         response, data = self._post_image_to_feed(test_app, data)
 478         object_id = data["object"]["id"]
 479
 480         with self.mock_oauth():
 481             response = test_app.get(data["object"]["links"]["self"]["href"])
 482             data = json.loads(response.body.decode())
 483
 484             assert response.status_code == 200
 485
 486             assert object_id == data["id"]
 487             assert "url" in data
 488             assert "links" in data
 489             assert data["objectType"] == "image"
 490
 491     def test_delete_media_by_activity(self, test_app):
 492         """ Test that an image can be deleted by a delete activity to feed """
 493         response, data = self._upload_image(test_app, GOOD_JPG)
 494         response, data = self._post_image_to_feed(test_app, data)
 495         object_id = data["object"]["id"]
 496
 497         activity = {
 498             "verb": "delete",
 499             "object": {
 500                 "id": object_id,
 501                 "objectType": "image",
 502             }
 503         }
 504
 505         response = self._activity_to_feed(test_app, activity)[1]
 506
 507         # Check the media is no longer in the database
 508         media = MediaEntry.query.filter_by(public_id=object_id).first()
 509
 510         assert media is None
 511
 512         # Check we've been given the full delete activity back
 513         assert "id" in response
 514         assert response["verb"] == "delete"
 515         assert "object" in response
 516         assert response["object"]["id"] == object_id
 517         assert response["object"]["objectType"] == "image"
 518
 519     def test_delete_comment_by_activity(self, test_app):
 520         """ Test that a comment is deleted by a delete activity to feed """
 521         # First upload an image to comment against
 522         response, data = self._upload_image(test_app, GOOD_JPG)
 523         response, data = self._post_image_to_feed(test_app, data)
 524
 525         # Post a comment to delete
 526         activity = {
 527             "verb": "post",
 528             "object": {
 529                 "objectType": "comment",
 530                 "content": "This is a comment.",
 531                 "inReplyTo": data["object"],
 532             }
 533         }
 534
 535         comment = self._activity_to_feed(test_app, activity)[1]
 536
 537         # Now delete the image
 538         activity = {
 539             "verb": "delete",
 540             "object": {
 541                 "id": comment["object"]["id"],
 542                 "objectType": "comment",
 543             }
 544         }
 545
 546         delete = self._activity_to_feed(test_app, activity)[1]
 547
 548         # Verify the comment no longer exists
 549         assert TextComment.query.filter_by(public_id=comment["object"]["id"]).first() is None
 550         comment_id = comment["object"]["id"]
 551
 552         # Check we've got a delete activity back
 553         assert "id" in delete
 554         assert delete["verb"] == "delete"
 555         assert "object" in delete
 556         assert delete["object"]["id"] == comment["object"]["id"]
 557         assert delete["object"]["objectType"] == "comment"
 558
 559     def test_edit_comment(self, test_app):
 560         """ Test that someone can update their own comment """
 561         # First upload an image to comment against
 562         response, data = self._upload_image(test_app, GOOD_JPG)
 563         response, data = self._post_image_to_feed(test_app, data)
 564
 565         # Post a comment to edit
 566         activity = {
 567             "verb": "post",
 568             "object": {
 569                 "objectType": "comment",
 570                 "content": "This is a comment",
 571                 "inReplyTo": data["object"],
 572             }
 573         }
 574
 575         comment = self._activity_to_feed(test_app, activity)[1]
 576
 577         # Now create an update activity to change the content
 578         activity = {
 579             "verb": "update",
 580             "object": {
 581                 "id": comment["object"]["id"],
 582                 "content": "This is my fancy new content string!",
 583                 "objectType": "comment",
 584             },
 585         }
 586
 587         comment = self._activity_to_feed(test_app, activity)[1]
 588
 589         # Verify the comment reflects the changes
 590         model = TextComment.query.filter_by(public_id=comment["object"]["id"]).first()
 591
 592         assert model.content == activity["object"]["content"]