heckyel/mediagoblin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Tidy up federation code and add tests to cover more of the APIs

Browse Source
This commit is contained in:
Jessica Tallon 2014-08-05 22:04:50 +01:00 committed by Christopher Allan Webber
parent 32ff6f4dc0
commit 9246a6ba89
11 changed files with 335 additions and 231 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
docs/source/siteadmin/commandline-upload.rst
View File

@ -15,7 +15,13 @@
Command-line uploading
======================
Want to submit media via the command line? It's fairly easy to do::
If you're a site administrator and have access to the server then you
can use the 'addmedia' task. If you're just a user and want to upload
media by the command line you can. This can be done with the pump.io
API. There is `p <https://github.com/xray7224/p/>`_, which will allow you
to easily upload media from the command line, follow p's docs to do that.
To use the addmedia command::
./bin/gmg addmedia username your_media.jpg

13
mediagoblin/db/models.py
View File

@ -439,18 +439,11 @@ class MediaEntry(Base, MediaEntryMixin):
def serialize(self, request, show_comments=True):
""" Unserialize MediaEntry to object """
author = self.get_uploader
url = request.urlgen(
"mediagoblin.user_pages.media_home",
user=author.username,
media=self.slug,
qualified=True
)
context = {
"id": self.id,
"author": author.serialize(request),
"objectType": self.objectType,
"url": url,
"url": self.url_for_self(request.urlgen),
"image": {
"url": request.host_url + self.thumb_url[1:],
},
@ -683,13 +676,13 @@ class MediaComment(Base, MediaCommentMixin):
# Validate inReplyTo has ID
if "id" not in data["inReplyTo"]:
return False
# Validate that the ID is correct
try:
media_id = int(data["inReplyTo"]["id"])
except ValueError:
return False
media = MediaEntry.query.filter_by(id=media_id).first()
if media is None:
return False

2
mediagoblin/federation/decorators.py
View File

@ -36,7 +36,6 @@ def user_has_privilege(privilege_name):
@wraps(controller)
@require_active_login
def wrapper(request, *args, **kwargs):
user_id = request.user.id
if not request.user.has_privilege(privilege_name):
error = "User '{0}' needs '{1}' privilege".format(
request.user.username,
@ -48,4 +47,3 @@ def user_has_privilege(privilege_name):
return wrapper
return user_has_privilege_decorator

12
mediagoblin/federation/routing.py
View File

@ -20,39 +20,39 @@ from mediagoblin.tools.routing import add_route
add_route(
"mediagoblin.federation.user",
"/api/user/<string:username>/",
"mediagoblin.federation.views:user"
"mediagoblin.federation.views:user_endpoint"
)
add_route(
"mediagoblin.federation.user.profile",
"/api/user/<string:username>/profile",
"mediagoblin.federation.views:profile"
"mediagoblin.federation.views:profile_endpoint"
)
# Inbox and Outbox (feed)
add_route(
"mediagoblin.federation.feed",
"/api/user/<string:username>/feed",
"mediagoblin.federation.views:feed"
"mediagoblin.federation.views:feed_endpoint"
)
add_route(
"mediagoblin.federation.user.uploads",
"/api/user/<string:username>/uploads",
"mediagoblin.federation.views:uploads"
"mediagoblin.federation.views:uploads_endpoint"
)
add_route(
"mediagoblin.federation.inbox",
"/api/user/<string:username>/inbox",
"mediagoblin.federation.views:feed"
"mediagoblin.federation.views:feed_endpoint"
)
# object endpoints
add_route(
"mediagoblin.federation.object",
"/api/<string:objectType>/<string:id>",
"mediagoblin.federation.views:object"
"mediagoblin.federation.views:object_endpoint"
)
add_route(
"mediagoblin.federation.object.comments",

389
mediagoblin/federation/views.py
View File

@ -31,47 +31,70 @@ from mediagoblin.submit.lib import new_upload_entry, api_upload_request, \
# MediaTypes
from mediagoblin.media_types.image import MEDIA_TYPE as IMAGE_MEDIA_TYPE
@oauth_required
def profile(request, raw=False):
""" This is /api/user/<username>/profile - This will give profile info """
user = request.matchdict["username"]
requested_user = User.query.filter_by(username=user).first()
# Getters
def get_profile(request):
"""
Gets the user's profile for the endpoint requested.
For example an endpoint which is /api/{username}/feed
as /api/cwebber/feed would get cwebber's profile. This
will return a tuple (username, user_profile). If no user
can be found then this function returns a (None, None).
"""
username = request.matchdict["username"]
user = User.query.filter_by(username=username).first()
if user is None:
return None, None
return user, user.serialize(request)
# Endpoints
@oauth_required
def profile_endpoint(request):
""" This is /api/user/<username>/profile - This will give profile info """
user, user_profile = get_profile(request)
if user is None:
username = request.matchdict["username"]
return json_error(
"No such 'user' with id '{0}'".format(user),
"No such 'user' with username '{0}'".format(username),
status=404
)
if raw:
return (requested_user.username, requested_user.serialize(request))
# user profiles are public so return information
return json_response(requested_user.serialize(request))
return json_response(user_profile)
@oauth_required
def user(request):
def user_endpoint(request):
""" This is /api/user/<username> - This will get the user """
user, user_profile = profile(request, raw=True)
data = {
user, user_profile = get_profile(request)
if user is None:
username = request.matchdict["username"]
return json_error(
"No such 'user' with username '{0}'".format(username),
status=404
)
return json_response({
"nickname": user.username,
"updated": user.created.isoformat(),
"published": user.created.isoformat(),
"profile": user_profile,
}
return json_response(data)
})
@oauth_required
@csrf_exempt
@user_has_privilege(u'uploader')
def uploads(request):
def uploads_endpoint(request):
""" Endpoint for file uploads """
user = request.matchdict["username"]
requested_user = User.query.filter_by(username=user).first()
username = request.matchdict["username"]
requested_user = User.query.filter_by(username=username).first()
if requested_user is None:
return json_error("No such 'user' with id '{0}'".format(user), 404)
return json_error("No such 'user' with id '{0}'".format(username), 404)
if request.method == "POST":
# Ensure that the user is only able to upload to their own
@ -85,7 +108,9 @@ def uploads(request):
# Wrap the data in the werkzeug file wrapper
if "Content-Type" not in request.headers:
return json_error(
"Must supply 'Content-Type' header to upload media.")
"Must supply 'Content-Type' header to upload media."
)
mimetype = request.headers["Content-Type"]
filename = mimetypes.guess_all_extensions(mimetype)
filename = 'unknown' + filename[0] if filename else filename
@ -104,156 +129,179 @@ def uploads(request):
@oauth_required
@csrf_exempt
def feed(request):
def feed_endpoint(request):
""" Handles the user's outbox - /api/user/<username>/feed """
user = request.matchdict["username"]
requested_user = User.query.filter_by(username=user).first()
username = request.matchdict["username"]
requested_user = User.query.filter_by(username=username).first()
# check if the user exists
if requested_user is None:
return json_error("No such 'user' with id '{0}'".format(user), 404)
return json_error("No such 'user' with id '{0}'".format(username), 404)
if request.data:
data = json.loads(request.data)
else:
data = {"verb": None, "object": {}}
# We need to check that the user they're posting to is
# the person that they are.
if request.method in ["POST", "PUT"] and \
requested_user.id != request.user.id:
return json_error(
"Not able to post to another users feed.",
status=403
)
if request.method == "POST" and data["verb"] == "post":
obj = data.get("object", None)
if obj is None:
return json_error("Could not find 'object' element.")
if request.method in ["POST", "PUT"]:
# Validate that the activity is valid
if "verb" not in data or "object" not in data:
return json_error("Invalid activity provided.")
if obj.get("objectType", None) == "comment":
# post a comment
if not request.user.has_privilege(u'commenter'):
# Check that the verb is valid
if data["verb"] not in ["post", "update"]:
return json_error("Verb not yet implemented", 501)
# We need to check that the user they're posting to is
# the person that they are.
if requested_user.id != request.user.id:
return json_error(
"Not able to post to another users feed.",
status=403
)
# Handle new posts
if data["verb"] == "post":
obj = data.get("object", None)
if obj is None:
return json_error("Could not find 'object' element.")
if obj.get("objectType", None) == "comment":
# post a comment
if not request.user.has_privilege(u'commenter'):
return json_error(
"Privilege 'commenter' required to comment.",
status=403
)
comment = MediaComment(author=request.user.id)
comment.unserialize(data["object"])
comment.save()
data = {
"verb": "post",
"object": comment.serialize(request)
}
return json_response(data)
elif obj.get("objectType", None) == "image":
# Posting an image to the feed
media_id = int(data["object"]["id"])
media = MediaEntry.query.filter_by(id=media_id).first()
if media is None:
return json_response(
"No such 'image' with id '{0}'".format(media_id),
status=404
)
if media.uploader != request.user.id:
return json_error(
"Privilege 'commenter' required to comment.",
status=403
)
if not media.unserialize(data["object"]):
return json_error(
"Invalid 'image' with id '{0}'".format(media_id)
)
media.save()
api_add_to_feed(request, media)
return json_response({
"verb": "post",
"object": media.serialize(request)
})
elif obj.get("objectType", None) is None:
# They need to tell us what type of object they're giving us.
return json_error("No objectType specified.")
else:
# Oh no! We don't know about this type of object (yet)
object_type = obj.get("objectType", None)
return json_error(
"Privilege 'commenter' required to comment.",
status=403
"Unknown object type '{0}'.".format(object_type)
)
comment = MediaComment(author=request.user.id)
comment.unserialize(data["object"])
comment.save()
data = {"verb": "post", "object": comment.serialize(request)}
return json_response(data)
# Updating existing objects
if data["verb"] == "update":
# Check we've got a valid object
obj = data.get("object", None)
elif obj.get("objectType", None) == "image":
# Posting an image to the feed
media_id = int(data["object"]["id"])
media = MediaEntry.query.filter_by(id=media_id).first()
if media is None:
return json_response(
"No such 'image' with id '{0}'".format(id=media_id),
status=404
)
if obj is None:
return json_error("Could not find 'object' element.")
if not media.unserialize(data["object"]):
return json_error(
"Invalid 'image' with id '{0}'".format(media_id)
)
if "objectType" not in obj:
return json_error("No objectType specified.")
media.save()
api_add_to_feed(request, media)
if "id" not in obj:
return json_error("Object ID has not been specified.")
return json_response({
"verb": "post",
"object": media.serialize(request)
})
obj_id = obj["id"]
elif obj.get("objectType", None) is None:
# They need to tell us what type of object they're giving us.
return json_error("No objectType specified.")
else:
# Oh no! We don't know about this type of object (yet)
object_type = obj.get("objectType", None)
return json_error("Unknown object type '{0}'.".format(object_type))
# Now try and find object
if obj["objectType"] == "comment":
if not request.user.has_privilege(u'commenter'):
return json_error(
"Privilege 'commenter' required to comment.",
status=403
)
elif request.method in ["PUT", "POST"] and data["verb"] == "update":
# Check we've got a valid object
obj = data.get("object", None)
comment = MediaComment.query.filter_by(id=obj_id).first()
if comment is None:
return json_error(
"No such 'comment' with id '{0}'.".format(obj_id)
)
if obj is None:
return json_error("Could not find 'object' element.")
# Check that the person trying to update the comment is
# the author of the comment.
if comment.author != request.user.id:
return json_error(
"Only author of comment is able to update comment.",
status=403
)
if "objectType" not in obj:
return json_error("No objectType specified.")
if not comment.unserialize(data["object"]):
return json_error(
"Invalid 'comment' with id '{0}'".format(obj_id)
)
if "id" not in obj:
return json_error("Object ID has not been specified.")
comment.save()
obj_id = obj["id"]
activity = {
"verb": "update",
"object": comment.serialize(request),
}
return json_response(activity)
# Now try and find object
if obj["objectType"] == "comment":
if not request.user.has_privilege(u'commenter'):
return json_error(
"Privilege 'commenter' required to comment.",
status=403
)
elif obj["objectType"] == "image":
image = MediaEntry.query.filter_by(id=obj_id).first()
if image is None:
return json_error(
"No such 'image' with the id '{0}'.".format(obj_id)
)
comment = MediaComment.query.filter_by(id=obj_id).first()
if comment is None:
return json_error(
"No such 'comment' with id '{0}'.".format(obj_id)
)
# Check that the person trying to update the comment is
# the author of the comment.
if image.uploader != request.user.id:
return json_error(
"Only uploader of image is able to update image.",
status=403
)
# Check that the person trying to update the comment is
# the author of the comment.
if comment.author != request.user.id:
return json_error(
"Only author of comment is able to update comment.",
status=403
)
if not image.unserialize(obj):
return json_error(
"Invalid 'image' with id '{0}'".format(obj_id)
)
image.save()
if not comment.unserialize(data["object"]):
return json_error(
"Invalid 'comment' with id '{0}'".format(obj_id)
)
comment.save()
activity = {
"verb": "update",
"object": comment.serialize(request),
}
return json_response(activity)
elif obj["objectType"] == "image":
image = MediaEntry.query.filter_by(id=obj_id).first()
if image is None:
return json_error(
"No such 'image' with the id '{0}'.".format(obj_id)
)
# Check that the person trying to update the comment is
# the author of the comment.
if image.uploader != request.user.id:
return json_error(
"Only uploader of image is able to update image.",
status=403
)
if not image.unserialize(obj):
return json_error(
"Invalid 'image' with id '{0}'".format(obj_id)
)
image.save()
activity = {
"verb": "update",
"object": image.serialize(request),
}
return json_response(activity)
activity = {
"verb": "update",
"object": image.serialize(request),
}
return json_response(activity)
elif request.method != "GET":
return json_error(
@ -299,9 +347,9 @@ def feed(request):
item = {
"verb": "post",
"object": media.serialize(request),
"actor": request.user.serialize(request),
"actor": media.get_uploader.serialize(request),
"content": "{0} posted a picture".format(request.user.username),
"id": 1,
"id": media.id,
}
item["updated"] = item["object"]["updated"]
item["published"] = item["object"]["published"]
@ -312,7 +360,7 @@ def feed(request):
return json_response(feed)
@oauth_required
def object(request, raw_obj=False):
def object_endpoint(request):
""" Lookup for a object type """
object_type = request.matchdict["objectType"]
try:
@ -333,46 +381,41 @@ def object(request, raw_obj=False):
media = MediaEntry.query.filter_by(id=object_id).first()
if media is None:
error = "Can't find '{0}' with ID '{1}'".format(
object_type,
object_id
)
return json_error(
"Can't find '{0}' with ID '{1}'".format(object_type, object_id),
status=404
)
if raw_obj:
return media
return json_response(media.serialize(request))
@oauth_required
def object_comments(request):
""" Looks up for the comments on a object """
media = object(request, raw_obj=True)
response = media
if isinstance(response, MediaEntry):
comments = response.serialize(request)
comments = comments.get("replies", {
"totalItems": 0,
"items": [],
"url": request.urlgen(
"mediagoblin.federation.object.comments",
objectType=media.objectType,
uuid=media.id,
qualified=True
)
})
media = MediaEntry.query.filter_by(id=request.matchdict["id"]).first()
if media is None:
return json_error("Can't find '{0}' with ID '{1}'".format(
request.matchdict["objectType"],
request.matchdict["id"]
), 404)
comments["displayName"] = "Replies to {0}".format(comments["url"])
comments["links"] = {
"first": comments["url"],
"self": comments["url"],
}
response = json_response(comments)
comments = response.serialize(request)
comments = comments.get("replies", {
"totalItems": 0,
"items": [],
"url": request.urlgen(
"mediagoblin.federation.object.comments",
objectType=media.objectType,
id=media.id,
qualified=True
)
})
return response
comments["displayName"] = "Replies to {0}".format(comments["url"])
comments["links"] = {
"first": comments["url"],
"self": comments["url"],
}
return json_response(comments)
##
# Well known

6
mediagoblin/init/celery/__init__.py
View File

@ -28,7 +28,9 @@ _log = logging.getLogger(__name__)
MANDATORY_CELERY_IMPORTS = [
'mediagoblin.processing.task',
'mediagoblin.notifications.task']
'mediagoblin.notifications.task',
'mediagoblin.submit.task',
]
DEFAULT_SETTINGS_MODULE = 'mediagoblin.init.celery.dummy_settings_module'
@ -65,7 +67,7 @@ def get_celery_settings_dict(app_config, global_config,
frequency = int(frequency)
celery_settings['CELERYBEAT_SCHEDULE'] = {
'garbage-collection': {
'task': 'mediagoblin.federation.task.garbage_collection',
'task': 'mediagoblin.submit.task.garbage_collection',
'schedule': datetime.timedelta(minutes=frequency),
}
}

1
mediagoblin/oauth/views.py
View File

@ -339,4 +339,3 @@ def access_token(request):
av = AccessTokenEndpoint(request_validator)
tokens = av.create_access_token(request, {})
return form_response(tokens)

12
mediagoblin/submit/lib.py
View File

@ -266,7 +266,9 @@ def api_upload_request(request, file_data, entry):
""" This handles a image upload request """
# Use the same kind of method from mediagoblin/submit/views:submit_start
entry.title = file_data.filename
entry.generate_slug()
# This will be set later but currently we just don't have enough information
entry.slug = None
queue_file = prepare_queue_task(request.app, entry, file_data.filename)
with queue_file:
@ -278,15 +280,13 @@ def api_upload_request(request, file_data, entry):
def api_add_to_feed(request, entry):
""" Add media to Feed """
if entry.title:
# Shame we have to do this here but we didn't have the data in
# api_upload_request as no filename is usually specified.
entry.slug = None
entry.generate_slug()
feed_url = request.urlgen(
'mediagoblin.user_pages.atom_feed',
qualified=True, user=request.user.username)
qualified=True, user=request.user.username
)
run_process_media(entry, feed_url)
add_comment_subscription(request.user, entry)
return json_response(entry.serialize(request))
return json_response(entry.serialize(request))

118
mediagoblin/tests/test_api.py
View File

@ -39,6 +39,7 @@ class TestAPI(object):
username="otheruser",
privileges=[u'active', u'uploader', u'commenter']
)
self.active_user = self.user
def _activity_to_feed(self, test_app, activity, headers=None):
""" Posts an activity to the user's feed """
@ -47,10 +48,9 @@ class TestAPI(object):
else:
headers = {"Content-Type": "application/json"}
with mock.patch("mediagoblin.decorators.oauth_required",
new_callable=self.mocked_oauth_required):
with self.mock_oauth():
response = test_app.post(
"/api/user/{0}/feed".format(self.user.username),
"/api/user/{0}/feed".format(self.active_user.username),
json.dumps(activity),
headers=headers
)
@ -66,10 +66,9 @@ class TestAPI(object):
}
with mock.patch("mediagoblin.decorators.oauth_required",
new_callable=self.mocked_oauth_required):
with self.mock_oauth():
response = test_app.post(
"/api/user/{0}/uploads".format(self.user.username),
"/api/user/{0}/uploads".format(self.active_user.username),
data,
headers=headers
)
@ -86,12 +85,11 @@ class TestAPI(object):
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()
request.user = User.query.filter_by(id=self.active_user.id).first()
return controller(request, *args, **kwargs)
def oauth_required(c):
@ -99,6 +97,13 @@ class TestAPI(object):
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
@ -128,9 +133,7 @@ class TestAPI(object):
"Content-Length": str(len(data))
}
with mock.patch("mediagoblin.decorators.oauth_required",
new_callable=self.mocked_oauth_required):
with self.mock_oauth():
# Will be self.user trying to upload as self.other_user
with pytest.raises(AppError) as excinfo:
test_app.post(
@ -154,8 +157,7 @@ class TestAPI(object):
"Content-Type": "application/json",
}
with mock.patch("mediagoblin.decorators.oauth_required",
new_callable=self.mocked_oauth_required):
with self.mock_oauth():
with pytest.raises(AppError) as excinfo:
test_app.post(
"/api/user/{0}/feed".format(self.other_user.username),
@ -187,8 +189,7 @@ class TestAPI(object):
media.save()
# Now lets try and edit the image as self.user, this should produce a 403 error.
with mock.patch("mediagoblin.decorators.oauth_required",
new_callable=self.mocked_oauth_required):
with self.mock_oauth():
with pytest.raises(AppError) as excinfo:
test_app.post(
"/api/user/{0}/feed".format(self.user.username),
@ -216,8 +217,7 @@ class TestAPI(object):
activity = {"verb": "update", "object": image}
with mock.patch("mediagoblin.decorators.oauth_required",
new_callable=self.mocked_oauth_required):
with self.mock_oauth():
response = test_app.post(
"/api/user/{0}/feed".format(self.user.username),
json.dumps(activity),
@ -251,8 +251,7 @@ class TestAPI(object):
"Content-Length": str(len(data)),
}
with mock.patch("mediagoblin.decorators.oauth_required",
new_callable=self.mocked_oauth_required):
with self.mock_oauth():
with pytest.raises(AppError) as excinfo:
test_app.post(
"/api/user/{0}/uploads".format(self.user.username),
@ -279,8 +278,7 @@ class TestAPI(object):
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):
with self.mock_oauth():
request = test_app.get(object_uri)
image = json.loads(request.body)
@ -345,8 +343,7 @@ class TestAPI(object):
"Content-Type": "application/json",
}
with mock.patch("mediagoblin.decorators.oauth_required",
new_callable=self.mocked_oauth_required):
with self.mock_oauth():
with pytest.raises(AppError) as excinfo:
test_app.post(
"/api/user/{0}/feed".format(self.other_user.username),
@ -382,6 +379,7 @@ class TestAPI(object):
comment_id = comment_data["object"]["id"]
comment = MediaComment.query.filter_by(id=comment_id).first()
comment.author = self.other_user.id
comment.save()
# Update the comment as someone else.
comment_data["object"]["content"] = "Yep"
@ -390,8 +388,7 @@ class TestAPI(object):
"object": comment_data["object"]
}
with mock.patch("mediagoblin.decorators.oauth_required",
new_callable=self.mocked_oauth_required):
with self.mock_oauth():
with pytest.raises(AppError) as excinfo:
test_app.post(
"/api/user/{0}/feed".format(self.user.username),
@ -404,8 +401,7 @@ class TestAPI(object):
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):
with self.mock_oauth():
response = test_app.get(uri)
profile = json.loads(response.body)
@ -416,9 +412,77 @@ class TestAPI(object):
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)
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:
response = test_app.get("/api/whoami")
assert "401 UNAUTHORIZED" in excinfo.value.message
def test_read_feed(self, test_app):
""" Test able to read objects from the feed """
response, data = self._upload_image(test_app, GOOD_JPG)
response, data = self._post_image_to_feed(test_app, 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)
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]["actor"]
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.message
def test_object_endpoint(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)
assert response.status_code == 200
assert object_id == data["id"]
assert "url" in data
assert "links" in data
assert data["objectType"] == "image"

3
mediagoblin/tests/test_celery_setup.py
View File

@ -48,7 +48,8 @@ def test_setup_celery_from_config():
assert isinstance(fake_celery_module.CELERYD_ETA_SCHEDULER_PRECISION, float)
assert fake_celery_module.CELERY_RESULT_PERSISTENT is True
assert fake_celery_module.CELERY_IMPORTS == [
'foo.bar.baz', 'this.is.an.import', 'mediagoblin.processing.task', 'mediagoblin.notifications.task']
'foo.bar.baz', 'this.is.an.import', 'mediagoblin.processing.task', \
'mediagoblin.notifications.task', 'mediagoblin.submit.task']
assert fake_celery_module.CELERY_RESULT_BACKEND == 'database'
assert fake_celery_module.CELERY_RESULT_DBURI == (
'sqlite:///' +

2
mediagoblin/tests/tools.py
View File

@ -33,7 +33,6 @@ from mediagoblin.db.base import Session
from mediagoblin.meddleware import BaseMeddleware
from mediagoblin.auth import gen_password_hash
from mediagoblin.gmg_commands.dbupdate import run_dbupdate
from mediagoblin.oauth.views import OAUTH_ALPHABET
from mediagoblin.tools.crypto import random_string
from datetime import datetime
@ -346,4 +345,3 @@ def fixture_add_comment_report(comment=None, reported_user=None,
Session.expunge(comment_report)
return comment_report