348 lines
12 KiB
Python
348 lines
12 KiB
Python
# GNU MediaGoblin -- federated, autonomous media hosting
|
|
# Copyright (C) 2011, 2012 MediaGoblin contributors. See AUTHORS.
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
import json
|
|
|
|
import mock
|
|
import pytest
|
|
|
|
from webtest import AppError
|
|
|
|
from .resources import GOOD_JPG
|
|
from mediagoblin import mg_globals
|
|
from mediagoblin.db.models import User, MediaEntry
|
|
from mediagoblin.tests.tools import fixture_add_user
|
|
from mediagoblin.moderation.tools import take_away_privileges
|
|
|
|
class TestAPI(object):
|
|
""" Test mediagoblin's pump.io complient APIs """
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def setup(self, test_app):
|
|
self.test_app = test_app
|
|
self.db = mg_globals.database
|
|
|
|
self.user = fixture_add_user(privileges=[u'active', u'uploader', u'commenter'])
|
|
self.other_user = fixture_add_user(
|
|
username="otheruser",
|
|
privileges=[u'active', u'uploader', u'commenter']
|
|
)
|
|
|
|
def _activity_to_feed(self, test_app, activity, headers=None):
|
|
""" Posts an activity to the user's feed """
|
|
if headers:
|
|
headers.setdefault("Content-Type", "application/json")
|
|
else:
|
|
headers = {"Content-Type": "application/json"}
|
|
|
|
with mock.patch("mediagoblin.decorators.oauth_required",
|
|
new_callable=self.mocked_oauth_required):
|
|
response = test_app.post(
|
|
"/api/user/{0}/feed".format(self.user.username),
|
|
json.dumps(activity),
|
|
headers=headers
|
|
)
|
|
|
|
return response, json.loads(response.body)
|
|
|
|
def _upload_image(self, test_app, image):
|
|
""" Uploads and image to MediaGoblin via pump.io API """
|
|
data = open(image, "rb").read()
|
|
headers = {
|
|
"Content-Type": "image/jpeg",
|
|
"Content-Length": str(len(data))
|
|
}
|
|
|
|
|
|
with mock.patch("mediagoblin.decorators.oauth_required",
|
|
new_callable=self.mocked_oauth_required):
|
|
response = test_app.post(
|
|
"/api/user/{0}/uploads".format(self.user.username),
|
|
data,
|
|
headers=headers
|
|
)
|
|
image = json.loads(response.body)
|
|
|
|
return response, image
|
|
|
|
def _post_image_to_feed(self, test_app, image):
|
|
""" Posts an already uploaded image to feed """
|
|
activity = {
|
|
"verb": "post",
|
|
"object": image,
|
|
}
|
|
|
|
return self._activity_to_feed(test_app, activity)
|
|
|
|
|
|
def mocked_oauth_required(self, *args, **kwargs):
|
|
""" Mocks mediagoblin.decorator.oauth_required to always validate """
|
|
|
|
def fake_controller(controller, request, *args, **kwargs):
|
|
request.user = User.query.filter_by(id=self.user.id).first()
|
|
return controller(request, *args, **kwargs)
|
|
|
|
def oauth_required(c):
|
|
return lambda *args, **kwargs: fake_controller(c, *args, **kwargs)
|
|
|
|
return oauth_required
|
|
|
|
def test_can_post_image(self, test_app):
|
|
""" Tests that an image can be posted to the API """
|
|
# First request we need to do is to upload the image
|
|
response, image = self._upload_image(test_app, GOOD_JPG)
|
|
|
|
# I should have got certain things back
|
|
assert response.status_code == 200
|
|
|
|
assert "id" in image
|
|
assert "fullImage" in image
|
|
assert "url" in image["fullImage"]
|
|
assert "url" in image
|
|
assert "author" in image
|
|
assert "published" in image
|
|
assert "updated" in image
|
|
assert image["objectType"] == "image"
|
|
|
|
# Check that we got the response we're expecting
|
|
response, _ = self._post_image_to_feed(test_app, image)
|
|
assert response.status_code == 200
|
|
|
|
def test_unable_to_upload_as_someone_else(self, test_app):
|
|
""" Test that can't upload as someoen else """
|
|
data = open(GOOD_JPG, "rb").read()
|
|
headers = {
|
|
"Content-Type": "image/jpeg",
|
|
"Content-Length": str(len(data))
|
|
}
|
|
|
|
with mock.patch("mediagoblin.decorators.oauth_required",
|
|
new_callable=self.mocked_oauth_required):
|
|
|
|
# Will be self.user trying to upload as self.other_user
|
|
with pytest.raises(AppError) as excinfo:
|
|
test_app.post(
|
|
"/api/user/{0}/uploads".format(self.other_user.username),
|
|
data,
|
|
headers=headers
|
|
)
|
|
|
|
assert "403 FORBIDDEN" in excinfo.value.message
|
|
|
|
def test_unable_to_post_feed_as_someone_else(self, test_app):
|
|
""" Tests that can't post an image to someone else's feed """
|
|
response, data = self._upload_image(test_app, GOOD_JPG)
|
|
|
|
activity = {
|
|
"verb": "post",
|
|
"object": data
|
|
}
|
|
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
with mock.patch("mediagoblin.decorators.oauth_required",
|
|
new_callable=self.mocked_oauth_required):
|
|
with pytest.raises(AppError) as excinfo:
|
|
test_app.post(
|
|
"/api/user/{0}/feed".format(self.other_user.username),
|
|
json.dumps(activity),
|
|
headers=headers
|
|
)
|
|
|
|
assert "403 FORBIDDEN" in excinfo.value.message
|
|
|
|
def test_upload_image_with_filename(self, test_app):
|
|
""" Tests that you can upload an image with filename and description """
|
|
response, data = self._upload_image(test_app, GOOD_JPG)
|
|
response, data = self._post_image_to_feed(test_app, data)
|
|
|
|
image = data["object"]
|
|
|
|
# Now we need to add a title and description
|
|
title = "My image ^_^"
|
|
description = "This is my super awesome image :D"
|
|
license = "CC-BY-SA"
|
|
|
|
image["displayName"] = title
|
|
image["content"] = description
|
|
image["license"] = license
|
|
|
|
activity = {"verb": "update", "object": image}
|
|
|
|
with mock.patch("mediagoblin.decorators.oauth_required",
|
|
new_callable=self.mocked_oauth_required):
|
|
response = test_app.post(
|
|
"/api/user/{0}/feed".format(self.user.username),
|
|
json.dumps(activity),
|
|
headers={"Content-Type": "application/json"}
|
|
)
|
|
|
|
image = json.loads(response.body)["object"]
|
|
|
|
# Check everything has been set on the media correctly
|
|
media = MediaEntry.query.filter_by(id=image["id"]).first()
|
|
assert media.title == title
|
|
assert media.description == description
|
|
assert media.license == license
|
|
|
|
# Check we're being given back everything we should on an update
|
|
assert image["id"] == media.id
|
|
assert image["displayName"] == title
|
|
assert image["content"] == description
|
|
assert image["license"] == license
|
|
|
|
|
|
def test_only_uploaders_post_image(self, test_app):
|
|
""" Test that only uploaders can upload images """
|
|
# Remove uploader permissions from user
|
|
take_away_privileges(self.user.username, u"uploader")
|
|
|
|
# Now try and upload a image
|
|
data = open(GOOD_JPG, "rb").read()
|
|
headers = {
|
|
"Content-Type": "image/jpeg",
|
|
"Content-Length": str(len(data)),
|
|
}
|
|
|
|
with mock.patch("mediagoblin.decorators.oauth_required",
|
|
new_callable=self.mocked_oauth_required):
|
|
with pytest.raises(AppError) as excinfo:
|
|
test_app.post(
|
|
"/api/user/{0}/uploads".format(self.user.username),
|
|
data,
|
|
headers=headers
|
|
)
|
|
|
|
# Assert that we've got a 403
|
|
assert "403 FORBIDDEN" in excinfo.value.message
|
|
|
|
def test_object_endpoint(self, test_app):
|
|
""" Tests that object can be looked up at endpoint """
|
|
# Post an image
|
|
response, data = self._upload_image(test_app, GOOD_JPG)
|
|
response, data = self._post_image_to_feed(test_app, data)
|
|
|
|
# Now lookup image to check that endpoint works.
|
|
image = data["object"]
|
|
|
|
assert "links" in image
|
|
assert "self" in image["links"]
|
|
|
|
# Get URI and strip testing host off
|
|
object_uri = image["links"]["self"]["href"]
|
|
object_uri = object_uri.replace("http://localhost:80", "")
|
|
|
|
with mock.patch("mediagoblin.decorators.oauth_required",
|
|
new_callable=self.mocked_oauth_required):
|
|
request = test_app.get(object_uri)
|
|
|
|
image = json.loads(request.body)
|
|
entry = MediaEntry.query.filter_by(id=image["id"]).first()
|
|
|
|
assert request.status_code == 200
|
|
assert entry.id == image["id"]
|
|
|
|
assert "image" in image
|
|
assert "fullImage" in image
|
|
assert "pump_io" in image
|
|
assert "links" in image
|
|
|
|
def test_post_comment(self, test_app):
|
|
""" Tests that I can post an comment media """
|
|
# Upload some media to comment on
|
|
response, data = self._upload_image(test_app, GOOD_JPG)
|
|
response, data = self._post_image_to_feed(test_app, data)
|
|
|
|
content = "Hai this is a comment on this lovely picture ^_^"
|
|
|
|
activity = {
|
|
"verb": "post",
|
|
"object": {
|
|
"objectType": "comment",
|
|
"content": content,
|
|
"inReplyTo": data["object"],
|
|
}
|
|
}
|
|
|
|
response, comment_data = self._activity_to_feed(test_app, activity)
|
|
assert response.status_code == 200
|
|
|
|
# Find the objects in the database
|
|
media = MediaEntry.query.filter_by(id=data["object"]["id"]).first()
|
|
comment = media.get_comments()[0]
|
|
|
|
# Tests that it matches in the database
|
|
assert comment.author == self.user.id
|
|
assert comment.content == content
|
|
|
|
# Test that the response is what we should be given
|
|
assert comment.id == comment_data["object"]["id"]
|
|
assert comment.content == comment_data["object"]["content"]
|
|
|
|
def test_unable_to_post_comment_as_someone_else(self, test_app):
|
|
""" Tests that you're unable to post a comment as someone else. """
|
|
# Upload some media to comment on
|
|
response, data = self._upload_image(test_app, GOOD_JPG)
|
|
response, data = self._post_image_to_feed(test_app, data)
|
|
|
|
activity = {
|
|
"verb": "post",
|
|
"object": {
|
|
"objectType": "comment",
|
|
"content": "comment commenty comment ^_^",
|
|
"inReplyTo": data["object"],
|
|
}
|
|
}
|
|
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
with mock.patch("mediagoblin.decorators.oauth_required",
|
|
new_callable=self.mocked_oauth_required):
|
|
with pytest.raises(AppError) as excinfo:
|
|
test_app.post(
|
|
"/api/user/{0}/feed".format(self.other_user.username),
|
|
json.dumps(activity),
|
|
headers=headers
|
|
)
|
|
|
|
assert "403 FORBIDDEN" in excinfo.value.message
|
|
|
|
|
|
def test_profile(self, test_app):
|
|
""" Tests profile endpoint """
|
|
uri = "/api/user/{0}/profile".format(self.user.username)
|
|
with mock.patch("mediagoblin.decorators.oauth_required",
|
|
new_callable=self.mocked_oauth_required):
|
|
response = test_app.get(uri)
|
|
profile = json.loads(response.body)
|
|
|
|
assert response.status_code == 200
|
|
|
|
assert profile["preferredUsername"] == self.user.username
|
|
assert profile["objectType"] == "person"
|
|
|
|
assert "links" in profile
|
|
|
|
def test_whoami_without_login(self, test_app):
|
|
""" Test that whoami endpoint returns error when not logged in """
|
|
with pytest.raises(AppError) as excinfo:
|
|
response = test_app.get("/api/whoami")
|
|
|
|
assert "401 UNAUTHORIZED" in excinfo.value.message
|