return self.content
def serialize(self, request):
+ href = request.urlgen(
+ "mediagoblin.federation.object",
+ object_type=self.object_type,
+ id=self.id,
+ qualified=True
+ )
published = UTC.localize(self.published)
updated = UTC.localize(self.updated)
obj = {
- "id": self.id,
+ "id": href,
"actor": self.get_actor.serialize(request),
"verb": self.verb,
"published": published.isoformat(),
ActivityMixin
from mediagoblin.tools.files import delete_media_files
from mediagoblin.tools.common import import_component
+from mediagoblin.tools.routing import extract_url_arguments
import six
from pytz import UTC
def serialize(self, request, show_comments=True):
""" Unserialize MediaEntry to object """
+ href = request.urlgen(
+ "mediagoblin.federation.object",
+ object_type=self.object_type,
+ id=self.id,
+ qualified=True
+ )
author = self.get_uploader
published = UTC.localize(self.created)
updated = UTC.localize(self.created)
context = {
- "id": self.id,
+ "id": href,
"author": author.serialize(request),
"objectType": self.object_type,
"url": self.url_for_self(request.urlgen, qualified=True),
},
"links": {
"self": {
- "href": request.urlgen(
- "mediagoblin.federation.object",
- object_type=self.object_type,
- id=self.id,
- qualified=True
- ),
+ "href": href,
},
}
def serialize(self, request):
""" Unserialize to python dictionary for API """
+ href = request.urlgen(
+ "mediagoblin.federation.object",
+ object_type=self.object_type,
+ id=self.id,
+ qualified=True
+ )
media = MediaEntry.query.filter_by(id=self.media_entry).first()
author = self.get_author
context = {
- "id": self.id,
+ "id": href,
"objectType": self.object_type,
"content": self.content,
"inReplyTo": media.serialize(request, show_comments=False),
return context
- def unserialize(self, data):
+ def unserialize(self, data, request):
""" Takes API objects and unserializes on existing comment """
# Do initial checks to verify the object is correct
required_attributes = ["content", "inReplyTo"]
# Validate that the ID is correct
try:
- media_id = int(data["inReplyTo"]["id"])
+ media_id = int(extract_url_arguments(
+ url=data["inReplyTo"]["id"],
+ urlmap=request.app.url_map
+ )["id"])
except ValueError:
return False
)
def serialize(self, request):
+ href = request.urlgen(
+ "mediagoblin.federation.object",
+ object_type=self.object_type,
+ id=self.id,
+ qualified=True
+ )
published = UTC.localize(self.published)
updated = UTC.localize(self.updated)
return {
- "id": self.id,
+ "id": href,
"displayName": self.name,
"published": published.isoformat(),
"updated": updated.isoformat(),
-# GNU MediaGoblin -- federated, autonomous media hosting
+# GN 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
from mediagoblin.decorators import oauth_required, require_active_login
from mediagoblin.federation.decorators import user_has_privilege
from mediagoblin.db.models import User, MediaEntry, MediaComment, Activity
+from mediagoblin.tools.routing import extract_url_arguments
from mediagoblin.tools.response import redirect, json_response, json_error, \
render_404, render_to_response
from mediagoblin.meddleware.csrf import csrf_exempt
)
comment = MediaComment(author=request.user.id)
- comment.unserialize(data["object"])
+ comment.unserialize(data["object"], request)
comment.save()
data = {
"verb": "post",
elif obj.get("objectType", None) == "image":
# Posting an image to the feed
- media_id = int(data["object"]["id"])
+ media_id = int(extract_url_arguments(
+ url=data["object"]["id"],
+ urlmap=request.app.url_map
+ )["id"])
+
media = MediaEntry.query.filter_by(id=media_id).first()
if media is None:
if "id" not in obj:
return json_error("Object ID has not been specified.")
- obj_id = obj["id"]
+ obj_id = int(extract_url_arguments(
+ url=obj["id"],
+ urlmap=request.app.url_map
+ )["id"])
# Now try and find object
if obj["objectType"] == "comment":
""" Lookup for a object type """
object_type = request.matchdict["object_type"]
try:
- object_id = int(request.matchdict["id"])
+ object_id = request.matchdict["id"]
except ValueError:
error = "Invalid object ID '{0}' for '{1}'".format(
request.matchdict["id"],
from .resources import GOOD_JPG
from mediagoblin import mg_globals
from mediagoblin.db.models import User, MediaEntry, MediaComment
+from mediagoblin.tools.routing import extract_url_arguments
from mediagoblin.tests.tools import fixture_add_user
from mediagoblin.moderation.tools import take_away_privileges
# 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(id=data["object"]["id"]).first()
+ id = int(data["object"]["id"].split("/")[-1])
+ media = MediaEntry.query.filter_by(id=id).first()
media.uploader = self.other_user.id
media.save()
image = json.loads(response.body.decode())["object"]
# Check everything has been set on the media correctly
- media = MediaEntry.query.filter_by(id=image["id"]).first()
+ id = int(image["id"].split("/")[-1])
+ media = MediaEntry.query.filter_by(id=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 int(image["id"].split("/")[-1]) == media.id
assert image["displayName"] == title
assert image["content"] == description
assert image["license"] == license
request = test_app.get(object_uri)
image = json.loads(request.body.decode())
- entry = MediaEntry.query.filter_by(id=image["id"]).first()
+ entry_id = int(image["id"].split("/")[-1])
+ entry = MediaEntry.query.filter_by(id=entry_id).first()
assert request.status_code == 200
- assert entry.id == image["id"]
assert "image" in image
assert "fullImage" in image
assert response.status_code == 200
# Find the objects in the database
- media = MediaEntry.query.filter_by(id=data["object"]["id"]).first()
+ media_id = int(data["object"]["id"].split("/")[-1])
+ media = MediaEntry.query.filter_by(id=media_id).first()
comment = media.get_comments()[0]
# Tests that it matches in the database
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):
response, comment_data = self._activity_to_feed(test_app, activity)
# change who uploaded the comment as it's easier than changing
- comment_id = comment_data["object"]["id"]
+ comment_id = int(comment_data["object"]["id"].split("/")[-1])
comment = MediaComment.query.filter_by(id=comment_id).first()
comment.author = self.other_user.id
comment.save()
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import logging
+import urlparse
import six
from werkzeug.routing import Map, Rule
for endpoint, url, controller in routes:
url = "%s/%s" % (mountpoint.rstrip('/'), url.lstrip('/'))
add_route(endpoint, url, controller)
+
+def extract_url_arguments(url, urlmap):
+ """
+ This extracts the URL arguments from a given URL
+ """
+ parsed_url = urlparse.urlparse(url)
+ map_adapter = urlmap.bind(
+ server_name=parsed_url.netloc,
+ script_name=parsed_url.path,
+ url_scheme=parsed_url.scheme,
+ path_info=parsed_url.path
+ )
+
+ return map_adapter.match()[1]
projects / mediagoblin.git / commitdiff