I will be switch to use ``from __future__ import unicode_literals`` later.
import logging
+
+import six
import wtforms
from sqlalchemy import or_
user.save()
# log the user in
- request.session['user_id'] = unicode(user.id)
+ request.session['user_id'] = six.text_type(user.id)
request.session.save()
# send verification email
# 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 six
+
from itsdangerous import BadSignature
from mediagoblin import messages, mg_globals
# set up login in session
if login_form.stay_logged_in.data:
request.session['stay_logged_in'] = True
- request.session['user_id'] = unicode(user.id)
+ request.session['user_id'] = six.text_type(user.id)
request.session.save()
if request.form.get('next'):
import datetime
import uuid
+import six
+
from sqlalchemy import (MetaData, Table, Column, Boolean, SmallInteger,
Integer, Unicode, UnicodeText, DateTime,
ForeignKey, Date)
for row in db.execute(media_table.select()):
# no slug, try setting to an id
if not row.slug:
- append_garbage_till_unique(row, unicode(row.id))
+ append_garbage_till_unique(row, six.text_type(row.id))
# has "=" or ":" in it... we're getting rid of those
elif u"=" in row.slug or u":" in row.slug:
append_garbage_till_unique(
existing_slugs[row.creator].append(row.slug)
for row_id in slugs_to_change:
- new_slug = unicode(uuid.uuid4())
+ new_slug = six.text_type(uuid.uuid4())
db.execute(collection_table.update().
where(collection_table.c.id == row_id).
values(slug=new_slug))
return the value of the key.
"""
media_file = MediaFile.query.filter_by(media_entry=self.id,
- name=unicode(file_key)).first()
+ name=six.text_type(file_key)).first()
if media_file:
if metadata_key:
Update the file_metadata of a MediaFile.
"""
media_file = MediaFile.query.filter_by(media_entry=self.id,
- name=unicode(file_key)).first()
+ name=six.text_type(file_key)).first()
file_metadata = media_file.file_metadata or {}
# 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 six
+
from datetime import datetime
from itsdangerous import BadSignature
media.tags = convert_to_tag_list_of_dicts(
form.tags.data)
- media.license = unicode(form.license.data) or None
+ media.license = six.text_type(form.license.data) or None
media.slug = slug
media.save()
attachment_public_filepath \
= mg_globals.public_store.get_unique_filepath(
- ['media_entries', unicode(media.id), 'attachment',
+ ['media_entries', six.text_type(media.id), 'attachment',
public_filename])
attachment_public_file = mg_globals.public_store.get_file(
bio=user.bio)
if request.method == 'POST' and form.validate():
- user.url = unicode(form.url.data)
- user.bio = unicode(form.bio.data)
+ user.url = six.text_type(form.url.data)
+ user.bio = six.text_type(form.bio.data)
user.save()
form.slug.errors.append(
_(u'A collection with that slug already exists for this user.'))
else:
- collection.title = unicode(form.title.data)
- collection.description = unicode(form.description.data)
- collection.slug = unicode(form.slug.data)
+ collection.title = six.text_type(form.title.data)
+ collection.description = six.text_type(form.description.data)
+ collection.slug = six.text_type(form.slug.data)
collection.save()
# 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/>.
+from __future__ import print_function
+
import os
+import six
+
from mediagoblin.gmg_commands import util as commands_util
from mediagoblin.submit.lib import (
submit_media, get_upload_file_limits,
# get the user
user = app.db.User.query.filter_by(username=args.username.lower()).first()
if user is None:
- print "Sorry, no user by username '%s'" % args.username
+ print("Sorry, no user by username '%s'" % args.username)
return
# check for the file, if it exists...
filename = os.path.split(args.filename)[-1]
abs_filename = os.path.abspath(args.filename)
if not os.path.exists(abs_filename):
- print "Can't find a file with filename '%s'" % args.filename
+ print("Can't find a file with filename '%s'" % args.filename)
return
upload_limit, max_file_size = get_upload_file_limits(user)
if some_string is None:
return None
else:
- return unicode(some_string)
+ return six.text_type(some_string)
try:
submit_media(
tags_string=maybe_unicodeify(args.tags) or u"",
upload_limit=upload_limit, max_file_size=max_file_size)
except FileUploadLimit:
- print "This file is larger than the upload limits for this site."
+ print("This file is larger than the upload limits for this site.")
except UserUploadLimit:
- print "This file will put this user past their upload limits."
+ print("This file will put this user past their upload limits.")
except UserPastUploadLimit:
- print "This user is already past their upload limits."
+ print("This user is already past their upload limits.")
# 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 six
+
from mediagoblin.gmg_commands import util as commands_util
from mediagoblin import auth
from mediagoblin import mg_globals
else:
# Create the user
entry = db.User()
- entry.username = unicode(args.username.lower())
- entry.email = unicode(args.email)
+ entry.username = six.text_type(args.username.lower())
+ entry.email = six.text_type(args.email)
entry.pw_hash = auth.gen_password_hash(args.password)
default_privileges = [
db.Privilege.query.filter(
db = mg_globals.database
user = db.User.query.filter_by(
- username=unicode(args.username.lower())).one()
+ username=six.text_type(args.username.lower())).one()
if user:
user.all_privileges.append(
db.Privilege.query.filter(
db = mg_globals.database
user = db.User.query.filter_by(
- username=unicode(args.username.lower())).one()
+ username=six.text_type(args.username.lower())).one()
if user:
user.pw_hash = auth.gen_password_hash(args.password)
user.save()
import Image
import logging
+import six
+
from mediagoblin import mg_globals as mgg
from mediagoblin.processing import (
create_pub_filepath, FilenameBuilder,
# Encode the unicode instance to ASCII and replace any
# non-ASCII with an HTML entity (&#
unicode_file.write(
- unicode(orig_file.read().decode(
+ six.text_type(orig_file.read().decode(
self.charset)).encode(
'ascii',
'xmlcharrefreplace'))
from datetime import datetime
+import six
+
from werkzeug.exceptions import Forbidden
from mediagoblin.tools import pluginapi
if request.method=='POST' and form.validate():
_log.info("Here")
blog = request.db.Blog()
- blog.title = unicode(form.title.data)
- blog.description = unicode(cleaned_markdown_conversion((form.description.data)))
+ blog.title = six.text_type(form.title.data)
+ blog.description = six.text_type(cleaned_markdown_conversion((form.description.data)))
blog.author = request.user.id
blog.generate_slug()
'app_config': mg_globals.app_config})
else:
if request.method == 'POST' and form.validate():
- blog.title = unicode(form.title.data)
- blog.description = unicode(cleaned_markdown_conversion((form.description.data)))
+ blog.title = six.text_type(form.title.data)
+ blog.description = six.text_type(cleaned_markdown_conversion((form.description.data)))
blog.author = request.user.id
blog.generate_slug()
blogpost = request.db.MediaEntry()
blogpost.media_type = 'mediagoblin.media_types.blogpost'
- blogpost.title = unicode(form.title.data)
- blogpost.description = unicode(cleaned_markdown_conversion((form.description.data)))
+ blogpost.title = six.text_type(form.title.data)
+ blogpost.description = six.text_type(cleaned_markdown_conversion((form.description.data)))
blogpost.tags = convert_to_tag_list_of_dicts(form.tags.data)
- blogpost.license = unicode(form.license.data) or None
+ blogpost.license = six.text_type(form.license.data) or None
blogpost.uploader = request.user.id
blogpost.generate_slug()
form = blog_forms.BlogPostEditForm(request.form, **defaults)
if request.method == 'POST' and form.validate():
- blogpost.title = unicode(form.title.data)
- blogpost.description = unicode(cleaned_markdown_conversion((form.description.data)))
+ blogpost.title = six.text_type(form.title.data)
+ blogpost.description = six.text_type(cleaned_markdown_conversion((form.description.data)))
blogpost.tags = convert_to_tag_list_of_dicts(form.tags.data)
- blogpost.license = unicode(form.license.data)
+ blogpost.license = six.text_type(form.license.data)
set_blogpost_state(request, blogpost)
blogpost.generate_slug()
blogpost.save()
import logging
import argparse
+import six
+
from mediagoblin import mg_globals as mgg
from mediagoblin.processing import (
BadMediaFail, FilenameBuilder,
resize_filter = PIL_FILTERS[filter.upper()]
except KeyError:
raise Exception('Filter "{0}" not found, choose one of {1}'.format(
- unicode(filter),
+ six.text_type(filter),
u', '.join(PIL_FILTERS.keys())))
resized.thumbnail(new_size, resize_filter)
or im.size[1] > new_size[1]\
or exif_image_needs_rotation(exif_tags):
resize_image(
- entry, im, unicode(keyname), target_name,
+ entry, im, six.text_type(keyname), target_name,
tuple(new_size),
exif_tags, conversions_subdir,
quality, filter)
import json
import logging
+import six
+
from werkzeug.exceptions import BadRequest
from werkzeug.wrappers import Response
callback_url = request.form.get('callback_url')
if callback_url:
- callback_url = unicode(callback_url)
+ callback_url = six.text_type(callback_url)
try:
entry = submit_media(
mg_app=request.app, user=request.user,
submitted_file=request.files['file'],
filename=request.files['file'].filename,
- title=unicode(request.form.get('title')),
- description=unicode(request.form.get('description')),
- license=unicode(request.form.get('license', '')),
+ title=six.text_type(request.form.get('title')),
+ description=six.text_type(request.form.get('description')),
+ license=six.text_type(request.form.get('license', '')),
upload_limit=upload_limit, max_file_size=max_file_size,
callback_url=callback_url)
'''
if isinstance(e, InvalidFileType) or \
isinstance(e, FileTypeNotSupported):
- raise BadRequest(unicode(e))
+ raise BadRequest(six.text_type(e))
else:
raise
import bcrypt
import random
+import six
+
from mediagoblin import mg_globals
from mediagoblin.tools.crypto import get_timed_signer_url
from mediagoblin.tools.mail import send_email
if extra_salt:
raw_pass = u"%s:%s" % (extra_salt, raw_pass)
- return unicode(
+ return six.text_type(
bcrypt.hashpw(raw_pass.encode('utf-8'), bcrypt.gensalt()))
import logging
+import six
+
from werkzeug.exceptions import Unauthorized
from mediagoblin.auth.tools import check_login_simple
if not request.authorization:
return False
- user = check_login_simple(unicode(request.authorization['username']),
+ user = check_login_simple(six.text_type(request.authorization['username']),
request.authorization['password'])
if user:
#
# 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 six
+
from mediagoblin import mg_globals, messages
from mediagoblin.auth.tools import register_user
from mediagoblin.db.models import User
if user:
# set up login in session
- request.session['user_id'] = unicode(user.id)
+ request.session['user_id'] = six.text_type(user.id)
request.session.save()
if request.form.get('next'):
from functools import wraps
+import six
+
from mediagoblin.tools.response import json_response
def generate_identifier():
''' Generates a ``uuid.uuid4()`` '''
- return unicode(uuid.uuid4())
+ return six.text_type(uuid.uuid4())
def generate_token():
'''
# XXX: We might not want it to use bcrypt, since bcrypt takes its time to
# generate the result.
- return unicode(getrandbits(192))
+ return six.text_type(getrandbits(192))
from urllib import urlencode
+import six
+
from werkzeug.exceptions import BadRequest
from mediagoblin.tools.response import render_to_response, redirect, json_response
if request.method == 'POST' and form.validate():
client = OAuthClient()
- client.name = unicode(form.name.data)
- client.description = unicode(form.description.data)
- client.type = unicode(form.type.data)
+ client.name = six.text_type(form.name.data)
+ client.description = six.text_type(form.description.data)
+ client.type = six.text_type(form.type.data)
client.owner_id = request.user.id
- client.redirect_uri = unicode(form.redirect_uri.data)
+ client.redirect_uri = six.text_type(form.redirect_uri.data)
client.save()
import base64
import time
+import six
+
from openid.association import Association as OIDAssociation
from openid.store.interface import OpenIDStore
from openid.store import nonce
if not assoc:
assoc = Association()
- assoc.server_url = unicode(server_url)
+ assoc.server_url = six.text_type(server_url)
assoc.handle = association.handle
# django uses base64 encoding, python-openid uses a blob field for
# secret
- assoc.secret = unicode(base64.encodestring(association.secret))
+ assoc.secret = six.text_type(base64.encodestring(association.secret))
assoc.issued = association.issued
assoc.lifetime = association.lifetime
assoc.assoc_type = association.assoc_type
#
# 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 six
+
from openid.consumer import consumer
from openid.consumer.discover import DiscoveryFailure
from openid.extensions.sreg import SRegRequest, SRegResponse
if user:
# Set up login in session
- request.session['user_id'] = unicode(user.id)
+ request.session['user_id'] = six.text_type(user.id)
request.session.save()
if request.session.get('next'):
import logging
import requests
+import six
+
from werkzeug.exceptions import BadRequest
from mediagoblin import messages, mg_globals
user = query.user if query else None
if user:
- request.session['user_id'] = unicode(user.id)
+ request.session['user_id'] = six.text_type(user.id)
request.session['persona_login_email'] = email
request.session.save()
import logging
import re
+import six
+
from werkzeug.exceptions import MethodNotAllowed, BadRequest, NotImplemented
from werkzeug.wrappers import BaseResponse
mg_app=request.app, user=request.user,
submitted_file=request.files['image'],
filename=request.files['image'].filename,
- title=unicode(form.name.data),
- description=unicode(form.comment.data),
+ title=six.text_type(form.name.data),
+ description=six.text_type(form.comment.data),
upload_limit=upload_limit, max_file_size=max_file_size)
collection_id = form.category.data
import logging
import os
+import six
+
from mediagoblin import mg_globals as mgg
from mediagoblin.db.util import atomic_update
from mediagoblin.db.models import MediaEntry
def create_pub_filepath(entry, filename):
return mgg.public_store.get_unique_filepath(
['media_entries',
- unicode(entry.id),
+ six.text_type(entry.id),
filename])
atomic_update(mgg.database.MediaEntry,
{'id': entry_id},
{u'state': u'failed',
- u'fail_error': unicode(exc.exception_path),
+ u'fail_error': six.text_type(exc.exception_path),
u'fail_metadata': exc.metadata})
else:
_log.warn("No idea what happened here, but it failed: %r", exc)
A cleaned list of unicode objects.
"""
cleaned_filepath = [
- unicode(secure_filename(filepath))
+ six.text_type(secure_filename(filepath))
for filepath in listy_filepath]
if u'' in cleaned_filepath:
import uuid
from os.path import splitext
+import six
+
from werkzeug.utils import secure_filename
from werkzeug.datastructures import FileStorage
# If the filename contains non ascii generate a unique name
if not all(ord(c) < 128 for c in filename):
- filename = unicode(uuid.uuid4()) + splitext(filename)[-1]
+ filename = six.text_type(uuid.uuid4()) + splitext(filename)[-1]
# Sniff the submitted media to determine which
# media plugin should handle processing
# create entry and save in database
entry = new_upload_entry(user)
entry.media_type = media_type
- entry.title = (title or unicode(splitext(filename)[0]))
+ entry.title = (title or six.text_type(splitext(filename)[0]))
entry.description = description or u""
# (If we got it off the task's auto-generation, there'd be
# a risk of a race condition when we'd save after sending
# off the task)
- task_id = unicode(uuid.uuid4())
+ task_id = six.text_type(uuid.uuid4())
entry.queued_task_id = task_id
# Now store generate the queueing related filename
# 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 six
+
from mediagoblin import messages
import mediagoblin.mg_globals as mg_globals
mg_app=request.app, user=request.user,
submitted_file=request.files['file'],
filename=request.files['file'].filename,
- title=unicode(submit_form.title.data),
- description=unicode(submit_form.description.data),
- license=unicode(submit_form.license.data) or None,
+ title=six.text_type(submit_form.title.data),
+ description=six.text_type(submit_form.description.data),
+ license=six.text_type(submit_form.license.data) or None,
tags_string=submit_form.tags.data,
upload_limit=upload_limit, max_file_size=max_file_size,
urlgen=request.urlgen)
if request.method == 'POST' and submit_form.validate():
collection = request.db.Collection()
- collection.title = unicode(submit_form.title.data)
- collection.description = unicode(submit_form.description.data)
+ collection.title = six.text_type(submit_form.title.data)
+ collection.description = six.text_type(submit_form.description.data)
collection.creator = request.user.id
collection.generate_slug()
import pkg_resources
import pytest
+import six
+
import six.moves.urllib.parse as urlparse
from mediagoblin import mg_globals
## Make sure user is logged in
request = template.TEMPLATE_TEST_CONTEXT[
'mediagoblin/user_pages/user_nonactive.html']['request']
- assert request.session['user_id'] == unicode(new_user.id)
+ assert request.session['user_id'] == six.text_type(new_user.id)
## Make sure we get email confirmation, and try verifying
assert len(mail.EMAIL_TEST_INBOX) == 1
# Make sure user is in the session
context = template.TEMPLATE_TEST_CONTEXT['mediagoblin/root.html']
session = context['request'].session
- assert session['user_id'] == unicode(test_user.id)
+ assert session['user_id'] == six.text_type(test_user.id)
# Successful logout
# -----------------
import json
import pytest
+import six
+
from urlparse import urlparse, parse_qs
from mediagoblin import mg_globals
code = parse_qs(urlparse(redirect.location).query)['code'][0]
client = self.db.OAuthClient.query.filter(
- self.db.OAuthClient.identifier == unicode(client_id)).first()
+ self.db.OAuthClient.identifier == six.text_type(client_id)).first()
client_secret = client.secret
import pkg_resources
import pytest
import mock
+import six
import six.moves.urllib.parse as urlparse
# Make sure user is in the session
context = template.TEMPLATE_TEST_CONTEXT['mediagoblin/root.html']
session = context['request'].session
- assert session['user_id'] == unicode(test_user.id)
+ assert session['user_id'] == six.text_type(test_user.id)
_test_authentication()
import logging
import pytest
+import six
+
from urlparse import parse_qs, urlparse
from mediagoblin import mg_globals
code = self.get_code_from_redirect_uri(code_redirect.location)
client = self.db.OAuthClient.query.filter(
- self.db.OAuthClient.identifier == unicode(client_id)).first()
+ self.db.OAuthClient.identifier == six.text_type(client_id)).first()
token_res = self.test_app.get('/oauth-2/access_token?client_id={0}&\
code={1}&client_secret={2}'.format(client_id, code, client.secret))
code = self.get_code_from_redirect_uri(code_redirect.location)
client = self.db.OAuthClient.query.filter(
- self.db.OAuthClient.identifier == unicode(client_id)).first()
+ self.db.OAuthClient.identifier == six.text_type(client_id)).first()
token_res = self.test_app.get('/oauth-2/access_token?\
code={0}&client_secret={1}'.format(code, client.secret))
import pkg_resources
import pytest
import mock
+import six
openid_consumer = pytest.importorskip(
"openid.consumer.consumer")
# Make sure user is in the session
context = template.TEMPLATE_TEST_CONTEXT['mediagoblin/root.html']
session = context['request'].session
- assert session['user_id'] == unicode(test_user.id)
+ assert session['user_id'] == six.text_type(test_user.id)
_test_new_user()
import pkg_resources
import pytest
import mock
+import six
import six.moves.urllib.parse as urlparse
# Make sure user is in the session
context = template.TEMPLATE_TEST_CONTEXT['mediagoblin/root.html']
session = context['request'].session
- assert session['user_id'] == unicode(test_user.id)
+ assert session['user_id'] == six.text_type(test_user.id)
_test_registration()
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import pytest
+import six
from mediagoblin.tools import template
from mediagoblin.tests.tools import (fixture_add_user, fixture_media_entry,
response, context = self.do_post(
{'report_reason':u'Testing Media Report',
- 'reporter_id':unicode(allie_id)},url= media_uri_slug + "report/")
+ 'reporter_id':six.text_type(allie_id)},url= media_uri_slug + "report/")
assert response.status == "302 FOUND"
response, context = self.do_post({
'report_reason':u'Testing Comment Report',
- 'reporter_id':unicode(allie_id)},url= comment_uri_slug + "report/")
+ 'reporter_id':six.text_type(allie_id)},url= comment_uri_slug + "report/")
assert response.status == "302 FOUND"
import tempfile
import pytest
+import six
+
from werkzeug.utils import secure_filename
from mediagoblin import storage
'mediagoblin.tests.test_storage:FakeStorageSystem'})
assert this_storage.foobie == 'eiboof'
assert this_storage.blech == 'hcelb'
- assert unicode(this_storage.__class__) == \
+ assert six.text_type(this_storage.__class__) == \
u'mediagoblin.tests.test_storage.FakeStorageSystem'
import os
import pytest
+import six
import six.moves.urllib.parse as urlparse
BIG_BLUE, GOOD_PDF, GPS_JPG, MED_PNG, BIG_PNG
GOOD_TAG_STRING = u'yin,yang'
-BAD_TAG_STRING = unicode('rage,' + 'f' * 26 + 'u' * 26)
+BAD_TAG_STRING = six.text_type('rage,' + 'f' * 26 + 'u' * 26)
FORM_CONTEXT = ['mediagoblin/submit/start.html', 'submit_form']
REQUEST_CONTEXT = ['mediagoblin/user_pages/user.html', 'request']
import email
+import six
+
from mediagoblin.tools import common, url, translate, mail, text, testing
testing._activate_testing()
orig = u"Password"
set_thread_locale("es")
- p1 = unicode(proxy)
+ p1 = six.text_type(proxy)
p1_should = pass_to_ugettext(orig)
assert p1_should != orig, "Test useless, string not translated"
assert p1 == p1_should
set_thread_locale("sv")
- p2 = unicode(proxy)
+ p2 = six.text_type(proxy)
p2_should = pass_to_ugettext(orig)
assert p2_should != orig, "Test broken, string not translated"
assert p2 == p2_should
import re
from unidecode import unidecode
+import six
+
_punct_re = re.compile(r'[\t !"#:$%&\'()*\-/<=>?@\[\\\]^_`{|},.]+')
result = []
for word in _punct_re.split(text.lower()):
result.extend(unidecode(word).split())
- return unicode(delim.join(result))
+ return six.text_type(delim.join(result))
import datetime
import json
+import six
+
from mediagoblin import messages, mg_globals
from mediagoblin.db.models import (MediaEntry, MediaTag, Collection,
CollectionItem, User)
comment = request.db.MediaComment()
comment.media_entry = media.id
comment.author = request.user.id
- comment.content = unicode(request.form['comment_content'])
+ comment.content = six.text_type(request.form['comment_content'])
# Show error message if commenting is disabled.
if not mg_globals.app_config['allow_comments']:
if not request.is_xhr:
return render_404(request)
- comment = unicode(request.form['comment_content'])
+ comment = six.text_type(request.form['comment_content'])
cleancomment = { "content":cleaned_markdown_conversion(comment)}
return Response(json.dumps(cleancomment))
projects / mediagoblin.git / commitdiff