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
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:],
},
# 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
@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,
return wrapper
return user_has_privilege_decorator
-
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",
# MediaTypes
from mediagoblin.media_types.image import MEDIA_TYPE as IMAGE_MEDIA_TYPE
+# 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(request, raw=False):
+def profile_endpoint(request):
""" This is /api/user/<username>/profile - This will give profile info """
- user = request.matchdict["username"]
- requested_user = User.query.filter_by(username=user).first()
+ 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
# 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
@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'):
- 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(id=media_id),
- status=404
- )
-
- 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("Unknown object type '{0}'.".format(object_type))
-
- elif request.method in ["PUT", "POST"] and data["verb"] == "update":
- # Check we've got a valid object
- obj = data.get("object", None)
-
- if obj is None:
- return json_error("Could not find 'object' element.")
-
- if "objectType" not in obj:
- return json_error("No objectType specified.")
-
- if "id" not in obj:
- return json_error("Object ID has not been specified.")
-
- obj_id = obj["id"]
-
- # 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
- )
+ # Check that the verb is valid
+ if data["verb"] not in ["post", "update"]:
+ return json_error("Verb not yet implemented", 501)
- 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 comment.author != request.user.id:
- return json_error(
- "Only author of comment is able to update comment.",
- status=403
- )
-
- 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
- )
+ # 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
+ )
- if not image.unserialize(obj):
+ # 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(
- "Invalid 'image' with id '{0}'".format(obj_id)
+ "Unknown object type '{0}'.".format(object_type)
)
- image.save()
- activity = {
- "verb": "update",
- "object": image.serialize(request),
- }
- return json_response(activity)
+ # Updating existing objects
+ if data["verb"] == "update":
+ # Check we've got a valid object
+ obj = data.get("object", None)
+
+ if obj is None:
+ return json_error("Could not find 'object' element.")
+
+ if "objectType" not in obj:
+ return json_error("No objectType specified.")
+
+ if "id" not in obj:
+ return json_error("Object ID has not been specified.")
+
+ obj_id = obj["id"]
+
+ # 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
+ )
+
+ 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 comment.author != request.user.id:
+ return json_error(
+ "Only author of comment is able to update comment.",
+ status=403
+ )
+
+ 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)
elif request.method != "GET":
return json_error(
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"]
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:
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
- )
- })
-
- comments["displayName"] = "Replies to {0}".format(comments["url"])
- comments["links"] = {
- "first": comments["url"],
- "self": comments["url"],
- }
- response = json_response(comments)
+ 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 = 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
MANDATORY_CELERY_IMPORTS = [
'mediagoblin.processing.task',
- 'mediagoblin.notifications.task']
+ 'mediagoblin.notifications.task',
+ 'mediagoblin.submit.task',
+]
DEFAULT_SETTINGS_MODULE = 'mediagoblin.init.celery.dummy_settings_module'
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),
}
}
av = AccessTokenEndpoint(request_validator)
tokens = av.create_access_token(request, {})
return form_response(tokens)
-
""" 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:
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))
\ No newline at end of file
+ return json_response(entry.serialize(request))
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 """
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
)
}
- 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
)
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):
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
"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(
"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),
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),
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),
"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),
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)
"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),
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"
"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),
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)
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"
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:///' +
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
Session.expunge(comment_report)
return comment_report
-
projects / mediagoblin.git / commitdiff