#
# 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 logging
-import base64
-
-from pkg_resources import resource_filename
-
+try:
+ import mock
+except ImportError:
+ import unittest.mock as mock
import pytest
+from webtest import AppError
+
+from .resources import GOOD_JPG
from mediagoblin import mg_globals
-from mediagoblin.tools import template, pluginapi
+from mediagoblin.db.models import User, MediaEntry, TextComment
from mediagoblin.tests.tools import fixture_add_user
+from mediagoblin.moderation.tools import take_away_privileges
-_log = logging.getLogger(__name__)
+class TestAPI(object):
+ """ Test mediagoblin's pump.io complient APIs """
-def resource(filename):
- '''
- Borrowed from the submission tests
- '''
- return resource_filename('mediagoblin.tests', 'test_submission/' + filename)
+ @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']
+ )
+ self.active_user = self.user
-GOOD_JPG = resource('good.jpg')
-GOOD_PNG = resource('good.png')
-EVIL_FILE = resource('evil')
-EVIL_JPG = resource('evil.jpg')
-EVIL_PNG = resource('evil.png')
-BIG_BLUE = resource('bigblue.png')
+ 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 self.mock_oauth():
+ response = test_app.post(
+ "/api/user/{0}/feed".format(self.active_user.username),
+ json.dumps(activity),
+ headers=headers
+ )
-class TestAPI(object):
- def setup(self):
- self.db = mg_globals.database
+ return response, json.loads(response.body.decode())
+
+ def _upload_image(self, test_app, image, custom_filename=None):
+ """ Uploads and image to MediaGoblin via pump.io API """
+ data = open(image, "rb").read()
+ headers = {
+ "Content-Type": "image/jpeg",
+ "Content-Length": str(len(data))
+ }
+
+ if custom_filename is not None:
+ headers["X-File-Name"] = custom_filename
+
+ with self.mock_oauth():
+ response = test_app.post(
+ "/api/user/{0}/uploads".format(self.active_user.username),
+ data,
+ headers=headers
+ )
+ image = json.loads(response.body.decode())
+
+ 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.active_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 mock_oauth(self):
+ """ Returns a mock.patch for the oauth_required decorator """
+ return mock.patch(
+ target="mediagoblin.decorators.oauth_required",
+ new_callable=self.mocked_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, data = self._post_image_to_feed(test_app, image)
+ assert response.status_code == 200
+ assert data["object"]["fullImage"]["url"].endswith("unknown.jpe")
+ assert data["object"]["image"]["url"].endswith("unknown.thumbnail.jpe")
+
+ def test_can_post_image_custom_filename(self, test_app):
+ """ Tests an image can be posted to the API with custom filename """
+ # First request we need to do is to upload the image
+ response, image = self._upload_image(test_app, GOOD_JPG,
+ custom_filename="hello.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, data = self._post_image_to_feed(test_app, image)
+ assert response.status_code == 200
+ assert data["object"]["fullImage"]["url"].endswith("hello.jpg")
+ assert data["object"]["image"]["url"].endswith("hello.thumbnail.jpg")
+
+ def test_can_post_image_tags(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)
+ assert response.status_code == 200
+
+ image["tags"] = ["hello", "world"]
+
+ # Check that we got the response we're expecting
+ response, data = self._post_image_to_feed(test_app, image)
+ assert response.status_code == 200
+ assert data["object"]["tags"] == ["hello", "world"]
+
+ 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 self.mock_oauth():
+ # 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.args[0]
+
+ 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 self.mock_oauth():
+ 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.args[0]
+
+ def test_only_able_to_update_own_image(self, test_app):
+ """ Test uploader is the only person who can update an image """
+ response, data = self._upload_image(test_app, GOOD_JPG)
+ response, data = self._post_image_to_feed(test_app, data)
+
+ activity = {
+ "verb": "update",
+ "object": data["object"],
+ }
+
+ headers = {
+ "Content-Type": "application/json",
+ }
+
+ # Lets change the image uploader to be self.other_user, this is easier
+ # than uploading the image as someone else as the way
+ # self.mocked_oauth_required and self._upload_image.
+ media = MediaEntry.query \
+ .filter_by(public_id=data["object"]["id"]) \
+ .first()
+ media.actor = self.other_user.id
+ media.save()
+
+ # Now lets try and edit the image as self.user, this should produce a
+ # 403 error.
+ with self.mock_oauth():
+ with pytest.raises(AppError) as excinfo:
+ test_app.post(
+ "/api/user/{0}/feed".format(self.user.username),
+ json.dumps(activity),
+ headers=headers
+ )
+
+ assert "403 FORBIDDEN" in excinfo.value.args[0]
+
+ 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 self.mock_oauth():
+ 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.decode())["object"]
+
+ # Check everything has been set on the media correctly
+ media = MediaEntry.query.filter_by(public_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.public_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 self.mock_oauth():
+ 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.args[0]
+
+ 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 self.mock_oauth():
+ request = test_app.get(object_uri)
+
+ image = json.loads(request.body.decode())
+ entry = MediaEntry.query.filter_by(public_id=image["id"]).first()
+
+ assert entry is not None
+
+ assert request.status_code == 200
+
+ assert "image" in image
+ assert "fullImage" in image
+ assert "pump_io" in image
+ assert "links" in image
+ assert "tags" 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(public_id=data["object"]["id"]) \
+ .first()
+ comment = media.get_comments()[0].comment()
+
+ # Tests that it matches in the database
+ assert comment.actor == self.user.id
+ assert comment.content == content
+
+ # Test that the response is what we should be given
+ 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 self.mock_oauth():
+ 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.args[0]
+
+ def test_unable_to_update_someone_elses_comment(self, test_app):
+ """ Test that you're able to update someoen elses comment. """
+ # 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",
+ }
+
+ # Post the comment.
+ response, comment_data = self._activity_to_feed(test_app, activity)
+
+ # change who uploaded the comment as it's easier than changing
+ comment = TextComment.query \
+ .filter_by(public_id=comment_data["object"]["id"]) \
+ .first()
+ comment.actor = self.other_user.id
+ comment.save()
+
+ # Update the comment as someone else.
+ comment_data["object"]["content"] = "Yep"
+ activity = {
+ "verb": "update",
+ "object": comment_data["object"]
+ }
+
+ with self.mock_oauth():
+ with pytest.raises(AppError) as excinfo:
+ test_app.post(
+ "/api/user/{0}/feed".format(self.user.username),
+ json.dumps(activity),
+ headers=headers
+ )
+
+ assert "403 FORBIDDEN" in excinfo.value.args[0]
+
+ def test_profile(self, test_app):
+ """ Tests profile endpoint """
+ uri = "/api/user/{0}/profile".format(self.user.username)
+ with self.mock_oauth():
+ response = test_app.get(uri)
+ profile = json.loads(response.body.decode())
+
+ assert response.status_code == 200
+
+ assert profile["preferredUsername"] == self.user.username
+ assert profile["objectType"] == "person"
+
+ assert "links" in profile
+
+ def test_user(self, test_app):
+ """ Test the user endpoint """
+ uri = "/api/user/{0}/".format(self.user.username)
+ with self.mock_oauth():
+ response = test_app.get(uri)
+ user = json.loads(response.body.decode())
+
+ assert response.status_code == 200
+
+ assert user["nickname"] == self.user.username
+ assert user["updated"] == self.user.created.isoformat()
+ assert user["published"] == self.user.created.isoformat()
+
+ # Test profile exists but self.test_profile will test the value
+ assert "profile" in response
+
+ def test_whoami_without_login(self, test_app):
+ """ Test that whoami endpoint returns error when not logged in """
+ with pytest.raises(AppError) as excinfo:
+ test_app.get("/api/whoami")
+
+ assert "401 UNAUTHORIZED" in excinfo.value.args[0]
+
+ def test_read_feed(self, test_app):
+ """ Test able to read objects from the feed """
+ response, image_data = self._upload_image(test_app, GOOD_JPG)
+ response, data = self._post_image_to_feed(test_app, image_data)
+
+ uri = "/api/user/{0}/feed".format(self.active_user.username)
+ with self.mock_oauth():
+ response = test_app.get(uri)
+ feed = json.loads(response.body.decode())
+
+ assert response.status_code == 200
+
+ # Check it has the attributes it should
+ assert "displayName" in feed
+ assert "objectTypes" in feed
+ assert "url" in feed
+ assert "links" in feed
+ assert "author" in feed
+ assert "items" in feed
+
+ # Check that image i uploaded is there
+ assert feed["items"][0]["verb"] == "post"
+ assert feed["items"][0]["id"] == data["id"]
+ assert feed["items"][0]["object"]["objectType"] == "image"
+ assert feed["items"][0]["object"]["id"] == data["object"]["id"]
+
+ default_limit = 20
+ items_count = default_limit * 2
+ for i in range(items_count):
+ response, image_data = self._upload_image(test_app, GOOD_JPG)
+ self._post_image_to_feed(test_app, image_data)
+ items_count += 1 # because there already is one
+
+ #
+ # default returns default_limit items
+ #
+ with self.mock_oauth():
+ response = test_app.get(uri)
+ feed = json.loads(response.body.decode())
+ assert len(feed["items"]) == default_limit
+
+ #
+ # silentely ignore count and offset that that are
+ # not a number
+ #
+ with self.mock_oauth():
+ response = test_app.get(uri + "?count=BAD&offset=WORSE")
+ feed = json.loads(response.body.decode())
+ assert len(feed["items"]) == default_limit
+
+ #
+ # if offset is less than default_limit items
+ # from the end of the feed, return less than
+ # default_limit
+ #
+ with self.mock_oauth():
+ near_the_end = items_count - default_limit / 2
+ response = test_app.get(uri + "?offset=%d" % near_the_end)
+ feed = json.loads(response.body.decode())
+ assert len(feed["items"]) < default_limit
+
+ #
+ # count=5 returns 5 items
+ #
+ with self.mock_oauth():
+ response = test_app.get(uri + "?count=5")
+ feed = json.loads(response.body.decode())
+ assert len(feed["items"]) == 5
+
+ def test_read_another_feed(self, test_app):
+ """ Test able to read objects from someone else's feed """
+ response, data = self._upload_image(test_app, GOOD_JPG)
+ response, data = self._post_image_to_feed(test_app, data)
+
+ # Change the active user to someone else.
+ self.active_user = self.other_user
+
+ # Fetch the feed
+ url = "/api/user/{0}/feed".format(self.user.username)
+ with self.mock_oauth():
+ response = test_app.get(url)
+ feed = json.loads(response.body.decode())
+
+ assert response.status_code == 200
+
+ # Check it has the attributes it ought to.
+ assert "displayName" in feed
+ assert "objectTypes" in feed
+ assert "url" in feed
+ assert "links" in feed
+ assert "author" in feed
+ assert "items" in feed
+
+ # Assert the uploaded image is there
+ assert feed["items"][0]["verb"] == "post"
+ assert feed["items"][0]["id"] == data["id"]
+ assert feed["items"][0]["object"]["objectType"] == "image"
+ assert feed["items"][0]["object"]["id"] == data["object"]["id"]
+
+ def test_cant_post_to_someone_elses_feed(self, test_app):
+ """ Test that can't post to someone elses feed """
+ response, data = self._upload_image(test_app, GOOD_JPG)
+ self.active_user = self.other_user
+
+ with self.mock_oauth():
+ with pytest.raises(AppError) as excinfo:
+ self._post_image_to_feed(test_app, data)
+
+ assert "403 FORBIDDEN" in excinfo.value.args[0]
+
+ def test_object_endpoint_requestable(self, test_app):
+ """ Test that object endpoint can be requested """
+ response, data = self._upload_image(test_app, GOOD_JPG)
+ response, data = self._post_image_to_feed(test_app, data)
+ object_id = data["object"]["id"]
+
+ with self.mock_oauth():
+ response = test_app.get(data["object"]["links"]["self"]["href"])
+ data = json.loads(response.body.decode())
+
+ assert response.status_code == 200
+
+ assert object_id == data["id"]
+ assert "url" in data
+ assert "links" in data
+ assert data["objectType"] == "image"
+
+ def test_delete_media_by_activity(self, test_app):
+ """ Test that an image can be deleted by a delete activity to feed """
+ response, data = self._upload_image(test_app, GOOD_JPG)
+ response, data = self._post_image_to_feed(test_app, data)
+ object_id = data["object"]["id"]
+
+ activity = {
+ "verb": "delete",
+ "object": {
+ "id": object_id,
+ "objectType": "image",
+ }
+ }
+
+ response = self._activity_to_feed(test_app, activity)[1]
+
+ # Check the media is no longer in the database
+ media = MediaEntry.query.filter_by(public_id=object_id).first()
- self.user_password = u'4cc355_70k3N'
- self.user = fixture_add_user(u'joapi', self.user_password)
+ assert media is None
- def login(self, test_app):
- test_app.post(
- '/auth/login/', {
- 'username': self.user.username,
- 'password': self.user_password})
+ # Check we've been given the full delete activity back
+ assert "id" in response
+ assert response["verb"] == "delete"
+ assert "object" in response
+ assert response["object"]["id"] == object_id
+ assert response["object"]["objectType"] == "image"
- def get_context(self, template_name):
- return template.TEMPLATE_TEST_CONTEXT[template_name]
+ def test_delete_comment_by_activity(self, test_app):
+ """ Test that a comment is deleted by a delete activity to feed """
+ # First upload an image to comment against
+ response, data = self._upload_image(test_app, GOOD_JPG)
+ response, data = self._post_image_to_feed(test_app, data)
- def http_auth_headers(self):
- return {'Authorization': 'Basic {0}'.format(
- base64.b64encode(':'.join([
- self.user.username,
- self.user_password])))}
+ # Post a comment to delete
+ activity = {
+ "verb": "post",
+ "object": {
+ "objectType": "comment",
+ "content": "This is a comment.",
+ "inReplyTo": data["object"],
+ }
+ }
- def do_post(self, data, test_app, **kwargs):
- url = kwargs.pop('url', '/api/submit')
- do_follow = kwargs.pop('do_follow', False)
+ comment = self._activity_to_feed(test_app, activity)[1]
- if not 'headers' in kwargs.keys():
- kwargs['headers'] = self.http_auth_headers()
+ # Now delete the image
+ activity = {
+ "verb": "delete",
+ "object": {
+ "id": comment["object"]["id"],
+ "objectType": "comment",
+ }
+ }
- response = test_app.post(url, data, **kwargs)
+ delete = self._activity_to_feed(test_app, activity)[1]
- if do_follow:
- response.follow()
+ # Verify the comment no longer exists
+ assert TextComment.query \
+ .filter_by(public_id=comment["object"]["id"]) \
+ .first() is None
- return response
+ assert "id" in comment["object"]
- def upload_data(self, filename):
- return {'upload_files': [('file', filename)]}
+ # Check we've got a delete activity back
+ assert "id" in delete
+ assert delete["verb"] == "delete"
+ assert "object" in delete
+ assert delete["object"]["id"] == comment["object"]["id"]
+ assert delete["object"]["objectType"] == "comment"
- def test_1_test_test_view(self, test_app):
- self.login(test_app)
+ def test_edit_comment(self, test_app):
+ """ Test that someone can update their own comment """
+ # First upload an image to comment against
+ response, data = self._upload_image(test_app, GOOD_JPG)
+ response, data = self._post_image_to_feed(test_app, data)
- response = test_app.get(
- '/api/test',
- headers=self.http_auth_headers())
+ # Post a comment to edit
+ activity = {
+ "verb": "post",
+ "object": {
+ "objectType": "comment",
+ "content": "This is a comment",
+ "inReplyTo": data["object"],
+ }
+ }
- assert response.body == \
- '{"username": "joapi", "email": "joapi@example.com"}'
+ comment = self._activity_to_feed(test_app, activity)[1]
- def test_2_test_submission(self, test_app):
- self.login(test_app)
+ # Now create an update activity to change the content
+ activity = {
+ "verb": "update",
+ "object": {
+ "id": comment["object"]["id"],
+ "content": "This is my fancy new content string!",
+ "objectType": "comment",
+ },
+ }
- response = self.do_post(
- {'title': 'Great JPG!'},
- test_app,
- **self.upload_data(GOOD_JPG))
+ comment = self._activity_to_feed(test_app, activity)[1]
- assert response.status_int == 200
+ # Verify the comment reflects the changes
+ model = TextComment.query \
+ .filter_by(public_id=comment["object"]["id"]) \
+ .first()
- assert self.db.MediaEntry.query.filter_by(title=u'Great JPG!').first()
+ assert model.content == activity["object"]["content"]
projects / mediagoblin.git / blobdiff