/celery.db
/kombu.db
/server-log.txt
+*.egg/
# pyconfigure/automake generated files
/Makefile
# The legacy of buildout
.installed.cfg
+
+# Virtualenv, tox
+venv*
+.tox/
--- /dev/null
+# A generic, single database configuration.
+
+[alembic]
+# path to migration scripts
+script_location = %(here)s/mediagoblin/db/migrations
+
+# template used to generate migration files
+# file_template = %%(rev)s_%%(slug)s
+
+# max length of characters to apply to the
+# "slug" field
+#truncate_slug_length = 40
+
+# set to 'true' to run the environment during
+# the 'revision' command, regardless of autogenerate
+# revision_environment = false
+
+# set to 'true' to allow .pyc and .pyo files without
+# a source .py file to be detected as revisions in the
+# versions/ directory
+# sourceless = false
+
+sqlalchemy.url = sqlite:///mediagoblin.db
+
+
+# Logging configuration
+[loggers]
+keys = root,sqlalchemy,alembic
+
+[handlers]
+keys = console
+
+[formatters]
+keys = generic
+
+[logger_root]
+level = WARN
+handlers = console
+qualname =
+
+[logger_sqlalchemy]
+level = WARN
+handlers =
+qualname = sqlalchemy.engine
+
+[logger_alembic]
+level = INFO
+handlers =
+qualname = alembic
+
+[handler_console]
+class = StreamHandler
+args = (sys.stderr,)
+level = NOTSET
+formatter = generic
+
+[formatter_generic]
+format = %(levelname)-5.5s [%(name)s] %(message)s
+datefmt = %H:%M:%S
local_bin="./bin"
case "$selfname" in
lazyserver.sh)
- starter_cmd=paster
+ starter_cmd=gunicorn
ini_prefix=paste
;;
lazycelery.sh)
if [ "$1" = "-h" ]; then
echo "$0 [-h] [-c filename.ini] [ARGS_to_${starter_cmd} ...]"
echo ""
- echo " For example:"
- echo " $0 -c fcgi.ini port_number=23371"
- echo " or: $0 --server-name=fcgi --log-file=paste.log"
+ echo " For Gunicorn settings, see at:"
+ echo " http://docs.gunicorn.org/en/19.0/settings.html"
echo ""
echo " The configfile defaults to ${ini_prefix}_local.ini,"
echo " if that is readable, otherwise ${ini_prefix}.ini."
export CELERY_ALWAYS_EAGER=true
case "$selfname" in
lazyserver.sh)
- $starter serve "$ini_file" "$@" --reload
+ $starter --paste "$ini_file" $@
;;
lazycelery.sh)
MEDIAGOBLIN_CONFIG="${ini_file}" \
--- /dev/null
+import functools
+import warnings
+
+import six
+
+if six.PY3:
+ from email.mime.text import MIMEText
+else:
+ from email.MIMEText import MIMEText
+
+
+def encode_to_utf8(method):
+ def wrapper(self):
+ if six.PY2 and isinstance(method(self), six.text_type):
+ return method(self).encode('utf-8')
+ return method(self)
+ functools.update_wrapper(wrapper, method, ['__name__', '__doc__'])
+ return wrapper
+
+
+# based on django.utils.encoding.python_2_unicode_compatible
+def py2_unicode(klass):
+ if six.PY2:
+ if '__str__' not in klass.__dict__:
+ warnings.warn("@py2_unicode cannot be applied "
+ "to %s because it doesn't define __str__()." %
+ klass.__name__)
+ klass.__unicode__ = klass.__str__
+ klass.__str__ = encode_to_utf8(klass.__unicode__)
+ if '__repr__' in klass.__dict__:
+ klass.__repr__ = encode_to_utf8(klass.__repr__)
+ return klass
from werkzeug.wrappers import Request
from werkzeug.exceptions import HTTPException
from werkzeug.routing import RequestRedirect
+from werkzeug.wsgi import SharedDataMiddleware
from mediagoblin import meddleware, __version__
from mediagoblin.db.util import check_db_up_to_date
if not mediagoblin_config:
raise IOError("Usable mediagoblin config not found.")
+ del app_config['config']
mgoblin_app = MediaGoblinApp(mediagoblin_config)
+ mgoblin_app.call_backend = SharedDataMiddleware(mgoblin_app.call_backend,
+ exports=app_config)
mgoblin_app = hook_transform('wrap_wsgi', mgoblin_app)
return mgoblin_app
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'):
# 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 unicode_literals
+
+import logging
+import os
+
+from alembic import command
+from alembic.config import Config
+from alembic.migration import MigrationContext
+
+from mediagoblin.db.base import Base
from mediagoblin.tools.common import simple_printer
from sqlalchemy import Table
from sqlalchemy.sql import select
+log = logging.getLogger(__name__)
+
+
class TableAlreadyExists(Exception):
pass
+class AlembicMigrationManager(object):
+
+ def __init__(self, session):
+ root_dir = os.path.abspath(os.path.dirname(os.path.dirname(
+ os.path.dirname(__file__))))
+ alembic_cfg_path = os.path.join(root_dir, 'alembic.ini')
+ self.alembic_cfg = Config(alembic_cfg_path)
+ self.session = session
+
+ def get_current_revision(self):
+ context = MigrationContext.configure(self.session.bind)
+ return context.get_current_revision()
+
+ def upgrade(self, version):
+ return command.upgrade(self.alembic_cfg, version or 'head')
+
+ def downgrade(self, version):
+ if isinstance(version, int) or version is None or version.isdigit():
+ version = 'base'
+ return command.downgrade(self.alembic_cfg, version)
+
+ def stamp(self, revision):
+ return command.stamp(self.alembic_cfg, revision=revision)
+
+ def init_tables(self):
+ Base.metadata.create_all(self.session.bind)
+ # load the Alembic configuration and generate the
+ # version table, "stamping" it with the most recent rev:
+ command.stamp(self.alembic_cfg, 'head')
+
+ def init_or_migrate(self, version=None):
+ if self.get_current_revision() is None:
+ log.info('Initializing tables and stamping it with '
+ 'the most recent migration...')
+ self.init_tables()
+ else:
+ self.upgrade(version)
+
+
class MigrationManager(object):
"""
Migration handling tool.
- migration_registry: where we should find all migrations to
run
"""
- self.name = unicode(name)
+ self.name = name
self.models = models
self.foundations = foundations
self.session = session
for migration_number, migration_func in migrations_to_run:
self.printer(
u' + Running migration %s, "%s"... ' % (
- migration_number, migration_func.func_name))
+ migration_number, migration_func.__name__))
migration_func(self.session)
self.set_current_migration(migration_number)
self.printer('done.\n')
import datetime
import uuid
+import six
+
from sqlalchemy import (MetaData, Table, Column, Boolean, SmallInteger,
Integer, Unicode, UnicodeText, DateTime,
ForeignKey, Date, Index)
from sqlalchemy.exc import ProgrammingError
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql import and_
-from migrate.changeset.constraint import UniqueConstraint
+from sqlalchemy.schema import UniqueConstraint
from mediagoblin.db.extratypes import JSONEncoded, MutationDict
from mediagoblin.db.migration_tools import (
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))
--- /dev/null
+Generic single-database configuration.
\ No newline at end of file
--- /dev/null
+from __future__ import with_statement
+from alembic import context
+from sqlalchemy import engine_from_config, pool
+from logging.config import fileConfig
+
+# this is the Alembic Config object, which provides
+# access to the values within the .ini file in use.
+config = context.config
+
+# Interpret the config file for Python logging.
+# This line sets up loggers basically.
+fileConfig(config.config_file_name)
+
+# add your model's MetaData object here
+# for 'autogenerate' support
+# from myapp import mymodel
+# target_metadata = mymodel.Base.metadata
+target_metadata = None
+
+# other values from the config, defined by the needs of env.py,
+# can be acquired:
+# my_important_option = config.get_main_option("my_important_option")
+# ... etc.
+
+def run_migrations_offline():
+ """Run migrations in 'offline' mode.
+
+ This configures the context with just a URL
+ and not an Engine, though an Engine is acceptable
+ here as well. By skipping the Engine creation
+ we don't even need a DBAPI to be available.
+
+ Calls to context.execute() here emit the given string to the
+ script output.
+
+ """
+ url = config.get_main_option("sqlalchemy.url")
+ context.configure(url=url, target_metadata=target_metadata)
+
+ with context.begin_transaction():
+ context.run_migrations()
+
+def run_migrations_online():
+ """Run migrations in 'online' mode.
+
+ In this scenario we need to create an Engine
+ and associate a connection with the context.
+
+ """
+ engine = engine_from_config(
+ config.get_section(config.config_ini_section),
+ prefix='sqlalchemy.',
+ poolclass=pool.NullPool)
+
+ connection = engine.connect()
+ context.configure(
+ connection=connection,
+ target_metadata=target_metadata
+ )
+
+ try:
+ with context.begin_transaction():
+ context.run_migrations()
+ finally:
+ connection.close()
+
+if context.is_offline_mode():
+ run_migrations_offline()
+else:
+ run_migrations_online()
+
--- /dev/null
+"""${message}
+
+Revision ID: ${up_revision}
+Revises: ${down_revision}
+Create Date: ${create_date}
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = ${repr(up_revision)}
+down_revision = ${repr(down_revision)}
+
+from alembic import op
+import sqlalchemy as sa
+${imports if imports else ""}
+
+def upgrade():
+ ${upgrades if upgrades else "pass"}
+
+
+def downgrade():
+ ${downgrades if downgrades else "pass"}
TODO: indexes on foreignkeys, where useful.
"""
+from __future__ import print_function
+
import logging
import datetime
from mediagoblin.tools.files import delete_media_files
from mediagoblin.tools.common import import_component
-# It's actually kind of annoying how sqlalchemy-migrate does this, if
-# I understand it right, but whatever. Anyway, don't remove this :P
-#
-# We could do migration calls more manually instead of relying on
-# this import-based meddling...
-from migrate import changeset
+import six
_log = logging.getLogger(__name__)
-
class User(Base, UserMixin):
"""
TODO: We should consider moving some rarely used fields
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 {}
- for key, value in kwargs.iteritems():
+ for key, value in six.iteritems(kwargs):
file_metadata[key] = value
media_file.file_metadata = file_metadata
media_data.get_media_entry = self
else:
# Update old media data
- for field, value in kwargs.iteritems():
+ for field, value in six.iteritems(kwargs):
setattr(media_data, field, value)
@memoized_property
# Delete all related files/attachments
try:
delete_media_files(self)
- except OSError, error:
+ except OSError as error:
# Returns list of files we failed to delete
_log.error('No such files from the user "{1}" to delete: '
'{0}'.format(str(error), self.get_uploader))
if __name__ == '__main__':
from sys import argv
- print repr(argv)
+ print(repr(argv))
if len(argv) == 2:
uri = argv[1]
else:
from sqlalchemy import create_engine, event
import logging
+import six
+
from mediagoblin.db.base import Base, Session
from mediagoblin import mg_globals
def __init__(self, engine):
self.engine = engine
- for k, v in Base._decl_class_registry.iteritems():
+ for k, v in six.iteritems(Base._decl_class_registry):
setattr(self, k, v)
def commit(self):
from functools import wraps
-from urlparse import urljoin
from werkzeug.exceptions import Forbidden, NotFound
from oauthlib.oauth1 import ResourceEndpoint
+from six.moves.urllib.parse import urljoin
+
from mediagoblin import mg_globals as mgg
from mediagoblin import messages
from mediagoblin.db.models import MediaEntry, User, MediaComment, AccessToken
# 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()
return redirect_obj(request, media)
if len(form.media_metadata) == 0:
- for identifier, value in media.media_metadata.iteritems():
+ for identifier, value in six.iteritems(media.media_metadata):
if identifier == "@context": continue
form.media_metadata.append_entry({
'identifier':identifier,
return json_error("No such 'user' with id '{0}'".format(username), 404)
if request.data:
- data = json.loads(request.data)
+ data = json.loads(request.data.decode())
else:
data = {"verb": None, "object": {}}
import argparse
import os
+import six
+
from mediagoblin.tools.common import import_component
'setup': 'mediagoblin.gmg_commands.deletemedia:parser_setup',
'func': 'mediagoblin.gmg_commands.deletemedia:deletemedia',
'help': 'Delete media entries'},
+ 'serve': {
+ 'setup': 'mediagoblin.gmg_commands.serve:parser_setup',
+ 'func': 'mediagoblin.gmg_commands.serve:serve',
+ 'help': 'PasteScript replacement'},
'batchaddmedia': {
'setup': 'mediagoblin.gmg_commands.batchaddmedia:parser_setup',
'func': 'mediagoblin.gmg_commands.batchaddmedia:batchaddmedia',
"otherwise mediagoblin.ini"))
subparsers = parser.add_subparsers(help='sub-command help')
- for command_name, command_struct in SUBCOMMAND_MAP.iteritems():
+ for command_name, command_struct in six.iteritems(SUBCOMMAND_MAP):
if 'help' in command_struct:
subparser = subparsers.add_parser(
command_name, help=command_struct['help'])
# 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(
mg_app=app,
user=user,
- submitted_file=file(abs_filename, 'r'), filename=filename,
+ submitted_file=open(abs_filename, 'r'), filename=filename,
title=maybe_unicodeify(args.title),
description=maybe_unicodeify(args.description),
license=maybe_unicodeify(args.license),
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.")
from sqlalchemy.orm import sessionmaker
from mediagoblin.db.open import setup_connection_and_db_from_config
-from mediagoblin.db.migration_tools import MigrationManager
+from mediagoblin.db.migration_tools import MigrationManager, AlembicMigrationManager
from mediagoblin.init import setup_global_and_app_config
from mediagoblin.tools.common import import_component
return managed_dbdata
+def run_alembic_migrations(db, app_config, global_config):
+ """Initializes a database and runs all Alembic migrations."""
+ Session = sessionmaker(bind=db.engine)
+ manager = AlembicMigrationManager(Session())
+ manager.init_or_migrate()
+
+
def run_dbupdate(app_config, global_config):
"""
Initialize or migrate the database as specified by the config file.
# Set up the database
db = setup_connection_and_db_from_config(app_config, migrations=True)
- #Run the migrations
+ # Run the migrations
run_all_migrations(db, app_config, global_config)
+ run_alembic_migrations(db, app_config, global_config)
def run_all_migrations(db, app_config, global_config):
"""
# Gather information from all media managers / projects
dbdatas = gather_database_data(
- global_config.get('plugins', {}).keys())
+ list(global_config.get('plugins', {}).keys()))
Session = sessionmaker(bind=db.engine)
# 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 sys
from mediagoblin.gmg_commands import util as commands_util
for media in medias:
found_medias.add(media.id)
media.delete()
- print 'Media ID %d has been deleted.' % media.id
+ print('Media ID %d has been deleted.' % media.id)
for media in media_ids - found_medias:
- print 'Can\'t find a media with ID %d.' % media
- print 'Done.'
+ print('Can\'t find a media with ID %d.' % media)
+ print('Done.')
sys.exit(0)
#
# 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 argparse
import os
manager = get_processing_manager_for_type(media_type)
except ProcessingManagerDoesNotExist:
entry = MediaEntry.query.filter_by(id=args.id_or_type).first()
- print 'No such processing manager for {0}'.format(entry.media_type)
+ print('No such processing manager for {0}'.format(entry.media_type))
if args.state:
processors = manager.list_all_processors_by_state(args.state)
else:
processors = manager.list_eligible_processors(media_entry)
- print "Available processors:"
- print "====================="
- print ""
+ print("Available processors:")
+ print("=====================")
+ print("")
if args.action_help:
for processor in processors:
- print processor.name
- print "-" * len(processor.name)
+ print(processor.name)
+ print("-" * len(processor.name))
parser = processor.generate_parser()
parser.print_help()
- print ""
+ print("")
else:
for processor in processors:
if processor.description:
- print " - %s: %s" % (processor.name, processor.description)
+ print(" - %s: %s" % (processor.name, processor.description))
else:
- print " - %s" % processor.name
+ print(" - %s" % processor.name)
def run(args, media_id=None):
processor_class = manager.get_processor(
args.reprocess_command, media_entry)
except ProcessorDoesNotExist:
- print 'No such processor "%s" for media with id "%s"' % (
- args.reprocess_command, media_entry.id)
+ print('No such processor "%s" for media with id "%s"' % (
+ args.reprocess_command, media_entry.id))
return
except ProcessorNotEligible:
- print 'Processor "%s" exists but media "%s" is not eligible' % (
- args.reprocess_command, media_entry.id)
+ print('Processor "%s" exists but media "%s" is not eligible' % (
+ args.reprocess_command, media_entry.id))
return
reprocess_parser = processor_class.generate_parser()
except ProcessingManagerDoesNotExist:
entry = MediaEntry.query.filter_by(id=media_id).first()
- print 'No such processing manager for {0}'.format(entry.media_type)
+ print('No such processing manager for {0}'.format(entry.media_type))
def bulk_run(args):
processor_class = manager.get_processor(
'resize', media_entry)
except ProcessorDoesNotExist:
- print 'No such processor "%s" for media with id "%s"' % (
- 'resize', media_entry.id)
+ print('No such processor "%s" for media with id "%s"' % (
+ 'resize', media_entry.id))
return
except ProcessorNotEligible:
- print 'Processor "%s" exists but media "%s" is not eligible' % (
- 'resize', media_entry.id)
+ print('Processor "%s" exists but media "%s" is not eligible' % (
+ 'resize', media_entry.id))
return
reprocess_parser = processor_class.generate_parser()
reprocess_info=reprocess_request)
except ProcessingManagerDoesNotExist:
- print 'No such processing manager for {0}'.format(entry.media_type)
+ print('No such processing manager for {0}'.format(entry.media_type))
def initial(args):
media_entry,
reprocess_action='initial')
except ProcessingManagerDoesNotExist:
- print 'No such processing manager for {0}'.format(entry.media_type)
+ print('No such processing manager for {0}'.format(entry.media_type))
def reprocess(args):
--- /dev/null
+# GNU 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
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# 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
+
+from paste.deploy import loadapp, loadserver
+
+
+class ServeCommand(object):
+
+ def loadserver(self, server_spec, name, relative_to, **kwargs):
+ return loadserver(server_spec, name=name, relative_to=relative_to,
+ **kwargs)
+
+ def loadapp(self, app_spec, name, relative_to, **kwargs):
+ return loadapp(app_spec, name=name, relative_to=relative_to, **kwargs)
+
+ def daemonize(self):
+ # TODO: pass to gunicorn if available
+ pass
+
+ def restart_with_reloader(self):
+ pass
+
+ def restart_with_monitor(self, reloader=False):
+ pass
+
+ def run(self):
+ print('Running...')
+
+
+def parser_setup(subparser):
+ subparser.add_argument('config', metavar='CONFIG_FILE')
+ subparser.add_argument('command',
+ choices=['start', 'stop', 'restart', 'status'],
+ nargs='?', default='start')
+ subparser.add_argument('-n', '--app-name',
+ dest='app_name',
+ metavar='NAME',
+ help="Load the named application (default main)")
+ subparser.add_argument('-s', '--server',
+ dest='server',
+ metavar='SERVER_TYPE',
+ help="Use the named server.")
+ subparser.add_argument('--reload',
+ dest='reload',
+ action='store_true',
+ help="Use auto-restart file monitor")
+
+
+def serve(args):
+ serve_cmd = ServeCommand() # TODO: pass args to it
+ serve_cmd.run()
# 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 six
+
from mediagoblin.gmg_commands import util as commands_util
from mediagoblin import auth
from mediagoblin import mg_globals
).count()
if users_with_username:
- print u'Sorry, a user with that name already exists.'
+ print(u'Sorry, a user with that name already exists.')
else:
# Create the user
entry = db.User()
- entry.username = 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(
entry.all_privileges = default_privileges
entry.save()
- print "User created (and email marked as verified)"
+ print(u"User created (and email marked as verified)")
def makeadmin_parser_setup(subparser):
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.Privilege.privilege_name==u'admin').one()
)
user.save()
- print 'The user is now Admin'
+ print(u'The user is now Admin')
else:
- print 'The user doesn\'t exist'
+ print(u'The user doesn\'t exist')
def changepw_parser_setup(subparser):
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()
- print 'Password successfully changed'
+ print(u'Password successfully changed')
else:
- print 'The user doesn\'t exist'
+ print(u'The user doesn\'t exist')
def deleteuser_parser_setup(subparser):
username=unicode(args.username.lower())).first()
if user:
user.delete()
- print 'The user %s has been deleted' % args.username
+ print('The user %s has been deleted' % args.username)
else:
- print 'The user %s doesn\'t exist' % args.username
+ print('The user %s doesn\'t exist' % args.username)
import datetime
import logging
+import six
+
from celery import Celery
from mediagoblin.tools.pluginapi import hook_runall
celery_settings = {}
# Add all celery settings from config
- for key, value in celery_conf.iteritems():
+ for key, value in six.iteritems(celery_conf):
celery_settings[key] = value
# TODO: use default result stuff here if it exists
__import__(settings_module)
this_module = sys.modules[settings_module]
- for key, value in celery_settings.iteritems():
+ for key, value in six.iteritems(celery_settings):
setattr(this_module, key, value)
if set_environ:
import Image
import logging
+import six
+
from mediagoblin import mg_globals as mgg
from mediagoblin.processing import (
create_pub_filepath, FilenameBuilder,
orig_file.seek(0)
def store_unicode_file(self):
- with file(self.process_filename, 'rb') as orig_file:
+ with open(self.process_filename, 'rb') as orig_file:
self._detect_charset(orig_file)
unicode_filepath = create_pub_filepath(self.entry,
'ascii-portable.txt')
# 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'))
self.entry.media_files['unicode'] = unicode_filepath
def generate_thumb(self, font=None, thumb_size=None):
- with file(self.process_filename, 'rb') as orig_file:
+ with open(self.process_filename, 'rb') as orig_file:
# If no font kwarg, check config
if not font:
font = self.ascii_config.get('thumbnail_font', None)
thumb = converter._create_image(
orig_file.read())
- with file(tmp_thumb, 'w') as thumb_file:
+ with open(tmp_thumb, 'w') as thumb_file:
thumb.thumbnail(
thumb_size,
Image.ANTIALIAS)
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()
# 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
+
try:
from PIL import Image
except ImportError:
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)
# Copy the new file to the conversion subdir, then remotely.
tmp_resized_filename = os.path.join(workdir, target_name)
- with file(tmp_resized_filename, 'w') as resized_file:
+ with open(tmp_resized_filename, 'wb') as resized_file:
resized.save(resized_file, quality=quality)
store_public(entry, keyname, tmp_resized_filename, target_name)
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)
clean = clean_exif(result)
useful = get_useful(clean)
- print pp.pprint(
- clean)
+ print(pp.pprint(clean))
try:
proc = Popen([unoconv, '--show'], stderr=PIPE)
output = proc.stderr.read()
- except OSError, e:
+ except OSError:
_log.warn(_('unoconv failing to run, check log file'))
return False
- if 'ERROR' in output:
+ if b'ERROR' in output:
return False
return True
_log.debug('pdfinfo could not read the pdf file.')
raise BadMediaFail()
+ lines = [l.decode() for l in lines]
info_dict = dict([[part.strip() for part in l.strip().split(':', 1)]
for l in lines if ':' in l])
pass
-class ThreeDee():
+class ThreeDee(object):
"""
3D model parser base class. Derrived classes are used for basic
analysis of 3D models, and are not intended to be used for 3D
import pkg_resources
import threading
+import six
#############################
# General mediagoblin globals
"""
from mediagoblin import mg_globals
- for key, value in kwargs.iteritems():
+ for key, value in six.iteritems(kwargs):
if not hasattr(mg_globals, key):
raise AssertionError("Global %s not known" % key)
setattr(mg_globals, key, value)
# 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
from mediagoblin.db.models import User, Privilege, UserBan
from mediagoblin.db.base import Session
from datetime import datetime
from mediagoblin.tools.translate import lazy_pass_to_ugettext as _
+
def take_punitive_actions(request, form, report, user):
- message_body =''
+ message_body = ''
# The bulk of this action is running through all of the different
# punitive actions that a moderator could take.
filters['reporter_id'] = form.reporter.data
filters = dict((k, v)
- for k, v in filters.iteritems() if v)
+ for k, v in six.iteritems(filters) if v)
return filters
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import datetime
-import string
+
+import six
from oauthlib.oauth1.rfc5849.utils import UNICODE_ASCII_CHARACTER_SET
from oauthlib.oauth1 import (RequestTokenEndpoint, AuthorizationEndpoint,
contacts = data.get("contacts", None)
if contacts is not None:
- if type(contacts) is not unicode:
+ if not isinstance(contacts, six.text_type):
error = "Contacts must be a string of space-seporated email addresses."
return json_response({"error": error}, status=400)
redirect_uris = data.get("redirect_uris", None)
if redirect_uris is not None:
- if type(redirect_uris) is not unicode:
+ if not isinstance(redirect_uris, six.text_type):
error = "redirect_uris must be space-seporated URLs."
return json_response({"error": error}, status=400)
import json
from functools import wraps
-from urlparse import urljoin
from werkzeug.exceptions import Forbidden
from werkzeug.wrappers import Response
+
+from six.moves.urllib.parse import urljoin
+
from mediagoblin import mg_globals
from mediagoblin.tools.pluginapi import PluginManager
from mediagoblin.storage.filestorage import BasicFileStorage
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', '')),
- tags_string=unicode(request.form.get('tags', '')),
+ title=six.text_type(request.form.get('title')),
+ description=six.text_type(request.form.get('description')),
+ license=six.text_type(request.form.get('license', '')),
+ tags_string=six.text_type(request.form.get('tags', '')),
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
# TODO: This is the *only* thing using Response() here, should that
# not simply use json_response()?
- return Response(json.dumps(user_data))
+ return Response(json.dumps(user_data, sort_keys=True))
def get_entries(request):
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:
import ldap
import logging
+import six
+
from mediagoblin.tools import pluginapi
_log = logging.getLogger(__name__)
return email
def login(self, username, password):
- for k, v in self.ldap_settings.iteritems():
+ for k, v in six.iteritems(self.ldap_settings):
try:
self._connect(v)
user_dn = v['LDAP_USER_DN_TEMPLATE'].format(username=username)
#
# 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'):
import wtforms
-from urlparse import urlparse
+from six.moves.urllib.parse import urlparse
from mediagoblin.tools.extlib.wtf_html5 import URLField
from mediagoblin.tools.translate import lazy_pass_to_ugettext as _
from mediagoblin.plugins.oauth.tools import generate_identifier, \
generate_secret, generate_token, generate_code, generate_refresh_token
-# Don't remove this, I *think* it applies sqlalchemy-migrate functionality onto
-# the models.
-from migrate import changeset
-
class OAuthClient(Base):
__tablename__ = 'oauth__client'
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))
import logging
-from urllib import urlencode
+from six.moves.urllib.parse import urlencode
+
+import six
from werkzeug.exceptions import BadRequest
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()
def _fill_element_dict(el, data, as_attr=()):
- for k, v in data.iteritems():
+ for k, v in six.iteritems(data):
if k in as_attr:
if not isinstance(v, six.string_types):
v = str(v)
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)
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import logging
-import urllib
-import urllib2
+
+from six.moves.urllib import request, parse
import celery
from celery.registry import tasks
hubparameters = {
'hub.mode': 'publish',
'hub.url': feed_url}
- hubdata = urllib.urlencode(hubparameters)
+ hubdata = parse.urlencode(hubparameters)
hubheaders = {
"Content-type": "application/x-www-form-urlencoded",
"Connection": "close"}
for huburl in mgg.app_config["push_urls"]:
- hubrequest = urllib2.Request(huburl, hubdata, hubheaders)
+ hubrequest = request.Request(huburl, hubdata, hubheaders)
try:
- hubresponse = urllib2.urlopen(hubrequest)
- except (urllib2.HTTPError, urllib2.URLError) as exc:
+ hubresponse = request.urlopen(hubrequest)
+ except (request.HTTPError, request.URLError) as exc:
# We retry by default 3 times before failing
_log.info("PuSH url %r gave error %r", huburl, exc)
try:
# 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 absolute_import
+
import shutil
import uuid
+import six
+
from werkzeug.utils import secure_filename
from mediagoblin.tools import common
shutil.copy(self.get_local_path(filepath), dest_path)
else:
with self.get_file(filepath, 'rb') as source_file:
- with file(dest_path, 'wb') as dest_file:
+ with open(dest_path, 'wb') as dest_file:
# Copy from remote storage in 4M chunks
shutil.copyfileobj(source_file, dest_file, length=4*1048576)
your storage system.
"""
with self.get_file(filepath, 'wb') as dest_file:
- with file(filename, 'rb') as source_file:
+ with open(filename, 'rb') as source_file:
# Copy to storage system in 4M chunks
shutil.copyfileobj(source_file, dest_file, length=4*1048576)
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:
"""
# This construct is needed, because dict(config) does
# not replace the variables in the config items.
- config_params = dict(config_section.iteritems())
+ config_params = dict(six.iteritems(config_section))
if 'storage_class' in config_params:
storage_class = config_params['storage_class']
storage_class = common.import_component(storage_class)
return storage_class(**config_params)
-import filestorage
+from . import filestorage
"""
# Override this method, using the "stream" iterator for efficient streaming
with self.get_file(filepath, 'rb') as source_file:
- with file(dest_path, 'wb') as dest_file:
+ with open(dest_path, 'wb') as dest_file:
for data in source_file:
dest_file.write(data)
# TODO: Fixing write() still seems worthwhile though.
_log.debug('Sending {0} to cloudfiles...'.format(filepath))
with self.get_file(filepath, 'wb') as dest_file:
- with file(filename, 'rb') as source_file:
+ with open(filename, 'rb') as source_file:
# Copy to storage system in 4096 byte chunks
dest_file.send(source_file)
# 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 os
+import shutil
+
+import six.moves.urllib.parse as urlparse
+
from mediagoblin.storage import (
StorageInterface,
clean_listy_filepath,
NoWebServing)
-import os
-import shutil
-import urlparse
-
class BasicFileStorage(StorageInterface):
"""
# 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.storage import StorageInterface, clean_listy_filepath
v = table.get(None)
if v:
res.append(" " * len(indent) + repr(indent) + ": " + repr(v))
- for k, v in table.iteritems():
+ for k, v in six.iteritems(table):
if k == None:
continue
res.append(" " * len(indent) + repr(k) + ":")
import uuid
from os.path import splitext
+import six
+
from werkzeug.utils import secure_filename
from werkzeug.datastructures import FileStorage
"""
Get the upload_limit and max_file_size for this user
"""
- if user.upload_limit >= 0:
+ if user.upload_limit is not None and user.upload_limit >= 0: # TODO: debug this
upload_limit = user.upload_limit
else:
upload_limit = mg_globals.app_config.get('upload_limit', None)
# 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()
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import json
-import mock
+try:
+ import mock
+except ImportError:
+ import unittest.mock as mock
import pytest
from webtest import AppError
headers=headers
)
- return response, json.loads(response.body)
+ return response, json.loads(response.body.decode())
def _upload_image(self, test_app, image):
""" Uploads and image to MediaGoblin via pump.io API """
data,
headers=headers
)
- image = json.loads(response.body)
+ image = json.loads(response.body.decode())
return response, image
headers=headers
)
- assert "403 FORBIDDEN" in excinfo.value.message
+ 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 """
headers=headers
)
- assert "403 FORBIDDEN" in excinfo.value.message
+ assert "403 FORBIDDEN" in excinfo.value.args[0]
def test_only_able_to_update_own_image(self, test_app):
""" Test's that the uploader is the only person who can update an image """
headers=headers
)
- assert "403 FORBIDDEN" in excinfo.value.message
+ 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 """
headers={"Content-Type": "application/json"}
)
- image = json.loads(response.body)["object"]
+ 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()
)
# Assert that we've got a 403
- assert "403 FORBIDDEN" in excinfo.value.message
+ assert "403 FORBIDDEN" in excinfo.value.args[0]
def test_object_endpoint(self, test_app):
""" Tests that object can be looked up at endpoint """
with self.mock_oauth():
request = test_app.get(object_uri)
- image = json.loads(request.body)
+ image = json.loads(request.body.decode())
entry = MediaEntry.query.filter_by(id=image["id"]).first()
assert request.status_code == 200
headers=headers
)
- assert "403 FORBIDDEN" in excinfo.value.message
+ 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. """
headers=headers
)
- assert "403 FORBIDDEN" in excinfo.value.message
+ 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)
+ profile = json.loads(response.body.decode())
assert response.status_code == 200
uri = "/api/user/{0}/".format(self.user.username)
with self.mock_oauth():
response = test_app.get(uri)
- user = json.loads(response.body)
+ user = json.loads(response.body.decode())
assert response.status_code == 200
with pytest.raises(AppError) as excinfo:
response = test_app.get("/api/whoami")
- assert "401 UNAUTHORIZED" in excinfo.value.message
+ assert "401 UNAUTHORIZED" in excinfo.value.args[0]
def test_read_feed(self, test_app):
""" Test able to read objects from the feed """
uri = "/api/user/{0}/feed".format(self.active_user.username)
with self.mock_oauth():
response = test_app.get(uri)
- feed = json.loads(response.body)
+ feed = json.loads(response.body.decode())
assert response.status_code == 200
with pytest.raises(AppError) as excinfo:
self._post_image_to_feed(test_app, data)
- assert "403 FORBIDDEN" in excinfo.value.message
+ assert "403 FORBIDDEN" in excinfo.value.args[0]
- def test_object_endpoint(self, test_app):
+ 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)
with self.mock_oauth():
response = test_app.get(data["object"]["links"]["self"]["href"])
- data = json.loads(response.body)
+ data = json.loads(response.body.decode())
assert response.status_code == 200
#
# 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 urlparse
+
import pkg_resources
import pytest
+import six
+
+import six.moves.urllib.parse as urlparse
+
from mediagoblin import mg_globals
from mediagoblin.db.models import User
from mediagoblin.tests.tools import get_app, fixture_add_user
## 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
assert message['To'] == 'angrygrrl@example.org'
email_context = template.TEMPLATE_TEST_CONTEXT[
'mediagoblin/auth/verification_email.txt']
- assert email_context['verification_url'] in message.get_payload(decode=True)
+ assert email_context['verification_url'].encode('ascii') in message.get_payload(decode=True)
path = urlparse.urlsplit(email_context['verification_url'])[2]
get_params = urlparse.urlsplit(email_context['verification_url'])[3]
email_context = template.TEMPLATE_TEST_CONTEXT[
'mediagoblin/plugins/basic_auth/fp_verification_email.txt']
#TODO - change the name of verification_url to something forgot-password-ish
- assert email_context['verification_url'] in message.get_payload(decode=True)
+ assert email_context['verification_url'].encode('ascii') in message.get_payload(decode=True)
path = urlparse.urlsplit(email_context['verification_url'])[2]
get_params = urlparse.urlsplit(email_context['verification_url'])[3]
# 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
# -----------------
#
# 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 urlparse
+
+import six.moves.urllib.parse as urlparse
from mediagoblin.db.models import User
from mediagoblin.plugins.basic_auth import tools as auth_tools
# 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 urlparse, os, pytest
+import six
+import six.moves.urllib.parse as urlparse
+import pytest
from mediagoblin import mg_globals
from mediagoblin.db.models import User, MediaEntry
assert message['To'] == 'new@example.com'
email_context = template.TEMPLATE_TEST_CONTEXT[
'mediagoblin/edit/verification.txt']
- assert email_context['verification_url'] in \
- message.get_payload(decode=True)
+ assert email_context['verification_url'].encode('ascii') in message.get_payload(decode=True)
path = urlparse.urlsplit(email_context['verification_url'])[2]
assert path == u'/edit/verify_email/'
old_metadata = new_metadata
new_metadata = media_entry.media_metadata
assert new_metadata == old_metadata
- assert ("u'On the worst day' is not a 'date-time'" in
- response.body)
+ context = template.TEMPLATE_TEST_CONTEXT[
+ 'mediagoblin/edit/metadata.html']
+ if six.PY2:
+ expected = "u'On the worst day' is not a 'date-time'"
+ else:
+ expected = "'On the worst day' is not a 'date-time'"
+ assert context['form'].errors[
+ 'media_metadata'][0]['identifier'][0] == expected
except ImportError:
import Image
+from collections import OrderedDict
+
from mediagoblin.tools.exif import exif_fix_image_orientation, \
extract_exif, clean_exif, get_gps_data, get_useful
from .resources import GOOD_JPG, EMPTY_JPG, BAD_JPG, GPS_JPG
assert gps == {}
# Do we have the "useful" tags?
- assert useful == {'EXIF CVAPattern': {'field_length': 8,
+
+ expected = OrderedDict({'EXIF CVAPattern': {'field_length': 8,
'field_offset': 26224,
'field_type': 7,
- 'printable': u'[0, 2, 0, 2, 1, 2, 0, 1]',
+ 'printable': '[0, 2, 0, 2, 1, 2, 0, 1]',
'tag': 41730,
'values': [0, 2, 0, 2, 1, 2, 0, 1]},
'EXIF ColorSpace': {'field_length': 2,
'field_offset': 476,
'field_type': 3,
- 'printable': u'sRGB',
+ 'printable': 'sRGB',
'tag': 40961,
'values': [1]},
'EXIF ComponentsConfiguration': {'field_length': 4,
'field_offset': 308,
'field_type': 7,
- 'printable': u'YCbCr',
+ 'printable': 'YCbCr',
'tag': 37121,
'values': [1, 2, 3, 0]},
'EXIF CompressedBitsPerPixel': {'field_length': 8,
'field_type': 5,
'printable': u'300',
'tag': 283,
- 'values': [[300, 1]]}}
+ 'values': [[300, 1]]}})
+
+ for k, v in useful.items():
+ assert v == expected[k]
def test_exif_image_orientation():
result)
# Are the dimensions correct?
- assert image.size == (428, 640)
+ assert image.size in ((428, 640), (640, 428))
# If this pixel looks right, the rest of the image probably will too.
assert_in(image.getdata()[10000],
import json
import pytest
-from urlparse import urlparse, parse_qs
+import six
+
+from six.moves.urllib.parse import parse_qs, urlparse
from mediagoblin import mg_globals
from mediagoblin.tools import processing
'client_id': client_id,
'client_secret': client_secret})
- response_data = json.loads(response.body)
+ response_data = json.loads(response.body.decode())
return response_data['access_token']
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
#
# 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 urlparse
+
import pkg_resources
import pytest
-import mock
+import six
+try:
+ import mock
+except ImportError:
+ import unittest.mock as mock
+
+import six.moves.urllib.parse as urlparse
from mediagoblin import mg_globals
from mediagoblin.db.base import Session
# 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 base64
+import json
import pytest
return template.TEMPLATE_TEST_CONTEXT[template_name]
def http_auth_headers(self):
- return {'Authorization': 'Basic {0}'.format(
- base64.b64encode(':'.join([
+ return {'Authorization': ('Basic {0}'.format(
+ base64.b64encode((':'.join([
self.user.username,
- self.user_password])))}
+ self.user_password])).encode('ascii')).decode()))}
def do_post(self, data, test_app, **kwargs):
url = kwargs.pop('url', '/api/submit')
'/api/test',
headers=self.http_auth_headers())
- assert response.body == \
- '{"username": "joapi", "email": "joapi@example.com"}'
+ assert json.loads(response.body.decode()) == {
+ "username": "joapi", "email": "joapi@example.com"}
def test_2_test_submission(self, test_app):
self.login(test_app)
jsonld_fail_1 = None
try:
jsonld_fail_1 = compact_and_validate(metadata_fail_1)
- except ValidationError, e:
+ except ValidationError as e:
assert e.message == "'All Rights Reserved.' is not a 'uri'"
assert jsonld_fail_1 == None
#,.,.,.,.,.,.,.,.,.,.,.,.,.,.,.,.,.,.,.,.,.,.,.,.,.,.,.,.,.,
jsonld_fail_2 = None
try:
jsonld_fail_2 = compact_and_validate(metadata_fail_2)
- except ValidationError, e:
+ except ValidationError as e:
assert e.message == "'The other day' is not a 'date-time'"
assert jsonld_fail_2 == None
# Maybe not every model needs a test, but some models have special
# methods, and so it makes sense to test them here.
+from __future__ import print_function
+
from mediagoblin.db.base import Session
from mediagoblin.db.models import MediaEntry, User, Privilege
from mediagoblin.tests import MGClientTestCase
from mediagoblin.tests.tools import fixture_add_user
-import mock
+try:
+ import mock
+except ImportError:
+ import unittest.mock as mock
import pytest
obj_in_session = 0
for obj in Session():
obj_in_session += 1
- print repr(obj)
+ print(repr(obj))
assert obj_in_session == 0
import pytest
-import urlparse
+import six.moves.urllib.parse as urlparse
from mediagoblin.tools import template, mail
self.logout()
self.login('otherperson', 'nosreprehto')
- self.test_app.get(media_uri_slug + '/c/{0}/'.format(comment_id))
+ self.test_app.get(media_uri_slug + 'c/{0}/'.format(comment_id))
notification = Notification.query.filter_by(id=notification_id).first()
assert notification.seen == True
- self.test_app.get(media_uri_slug + '/notifications/silence/')
+ self.test_app.get(media_uri_slug + 'notifications/silence/')
subscription = CommentSubscription.query.filter_by(id=subscription_id)\
.first()
# 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 cgi
-
import pytest
-from urlparse import parse_qs, urlparse
+
+from six.moves.urllib.parse import parse_qs, urlparse
from oauthlib.oauth1 import Client
headers["Content-Type"] = self.MIME_FORM
response = self.test_app.post(endpoint, headers=headers)
- response = cgi.parse_qs(response.body)
+ response = parse_qs(response.body.decode())
# each element is a list, reduce it to a string
for key, value in response.items():
import logging
import pytest
-from urlparse import parse_qs, urlparse
+import six
+
+from six.moves.urllib.parse import parse_qs, urlparse
from mediagoblin import mg_globals
from mediagoblin.tools import template, pluginapi
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))
assert token_res.status_int == 200
- token_data = json.loads(token_res.body)
+ token_data = json.loads(token_res.body.decode())
assert not 'error' in token_data
assert 'access_token' in token_data
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))
assert token_res.status_int == 200
- token_data = json.loads(token_res.body)
+ token_data = json.loads(token_res.body.decode())
assert 'error' in token_data
assert not 'access_token' in token_data
assert token_res.status_int == 200
- new_token_data = json.loads(token_res.body)
+ new_token_data = json.loads(token_res.body.decode())
assert not 'error' in new_token_data
assert 'access_token' in new_token_data
# 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 urlparse
import pkg_resources
import pytest
-import mock
+import six
+import six.moves.urllib.parse as urlparse
+try:
+ import mock
+except ImportError:
+ import unittest.mock as mock
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()
[DEFAULT]
debug = true
-[composite:main]
-use = egg:Paste#urlmap
-/ = mediagoblin
-/mgoblin_media/ = publicstore_serve
-/test_static/ = mediagoblin_static
-/theme_static/ = theme_static
-/plugin_static/ = plugin_static
-
-[app:mediagoblin]
+[app:main]
use = egg:mediagoblin#app
config = %(here)s/mediagoblin.ini
-
-[app:publicstore_serve]
-use = egg:Paste#static
-document_root = %(here)s/user_dev/media/public
-
-[app:mediagoblin_static]
-use = egg:Paste#static
-document_root = %(here)s/mediagoblin/static/
-
-[app:theme_static]
-use = egg:Paste#static
-document_root = %(here)s/user_dev/theme_static/
-cache_max_age = 86400
-
-[app:plugin_static]
-use = egg:Paste#static
-document_root = %(here)s/user_dev/plugin_static/
-cache_max_age = 86400
+/mgoblin_media = %(here)s/user_dev/media/public
+/test_static = %(here)s/mediagoblin/static
+/theme_static = %(here)s/user_dev/theme_static
+/plugin_static = %(here)s/user_dev/plugin_static
[celery]
CELERY_ALWAYS_EAGER = true
[server:main]
-use = egg:Paste#http
+use = egg:gunicorn
host = 127.0.0.1
port = 6543
# 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 collections
import tempfile
import shutil
import os
@pytest.mark.skipif("not check_prerequisites()")
def test_pdf():
- good_dict = {'pdf_version_major': 1, 'pdf_title': '',
+ good_dict = collections.OrderedDict({'pdf_version_major': 1, 'pdf_title': '',
'pdf_page_size_width': 612, 'pdf_author': '',
'pdf_keywords': '', 'pdf_pages': 10,
'pdf_producer': 'dvips + GNU Ghostscript 7.05',
'pdf_version_minor': 3,
'pdf_creator': 'LaTeX with hyperref package',
- 'pdf_page_size_height': 792}
+ 'pdf_page_size_height': 792})
assert pdf_info(GOOD) == good_dict
temp_dir = tempfile.mkdtemp()
create_pdf_thumb(GOOD, os.path.join(temp_dir, 'good_256_256.png'), 256, 256)
#
# 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 urlparse
+
import pkg_resources
import pytest
-import mock
+import six
+try:
+ import mock
+except ImportError:
+ import unittest.mock as mock
+
+import six.moves.urllib.parse as urlparse
pytest.importorskip("requests")
# 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()
def test_session(self):
resp = self.do_post("pwg.session.login",
{"username": u"nouser", "password": "wrong"})
- assert resp.body == XML_PREFIX \
- + '<rsp stat="fail"><err code="999" msg="Invalid username/password"/></rsp>'
+ assert resp.body == (XML_PREFIX + '<rsp stat="fail"><err code="999" msg="Invalid username/password"/></rsp>').encode('ascii')
resp = self.do_post("pwg.session.login",
{"username": self.username, "password": "wrong"})
- assert resp.body == XML_PREFIX \
- + '<rsp stat="fail"><err code="999" msg="Invalid username/password"/></rsp>'
+ assert resp.body == (XML_PREFIX + '<rsp stat="fail"><err code="999" msg="Invalid username/password"/></rsp>').encode('ascii')
resp = self.do_get("pwg.session.getStatus")
- assert resp.body == XML_PREFIX \
- + '<rsp stat="ok"><username>guest</username></rsp>'
+ assert resp.body == (XML_PREFIX + '<rsp stat="ok"><username>guest</username></rsp>').encode('ascii')
resp = self.do_post("pwg.session.login",
{"username": self.username, "password": self.password})
- assert resp.body == XML_PREFIX + '<rsp stat="ok">1</rsp>'
+ assert resp.body == (XML_PREFIX + '<rsp stat="ok">1</rsp>').encode('ascii')
resp = self.do_get("pwg.session.getStatus")
- assert resp.body == XML_PREFIX \
- + '<rsp stat="ok"><username>chris</username></rsp>'
+ assert resp.body == (XML_PREFIX + '<rsp stat="ok"><username>chris</username></rsp>').encode('ascii')
self.do_get("pwg.session.logout")
resp = self.do_get("pwg.session.getStatus")
- assert resp.body == XML_PREFIX \
- + '<rsp stat="ok"><username>guest</username></rsp>'
+ assert resp.body == (XML_PREFIX + '<rsp stat="ok"><username>guest</username></rsp>').encode('ascii')
assert pluginapi.hook_handle(
"nothing_handling", call_log, unhandled_okay=True) is None
assert call_log == []
-
+
# Multiple provided, go with the first!
call_log = []
assert pluginapi.hook_handle(
"""
# Specific thing passed into a page
result = context_modified_app.get("/modify_context/specific/")
- assert result.body.strip() == """Specific page!
+ assert result.body.strip() == b"""Specific page!
specific thing: in yer specificpage
global thing: globally appended!
# General test, should have global context variable only
result = context_modified_app.get("/modify_context/")
- assert result.body.strip() == """General page!
+ assert result.body.strip() == b"""General page!
global thing: globally appended!
lol: cats
junk_file_path = os.path.join(
linked_assets_dir.rstrip(os.path.sep),
'junk.txt')
- with file(junk_file_path, 'w') as junk_file:
+ with open(junk_file_path, 'w') as junk_file:
junk_file.write('barf')
os.unlink(plugin_link_dir)
# link dir exists, but is a non-symlink
os.unlink(plugin_link_dir)
- with file(plugin_link_dir, 'w') as clobber_file:
+ with open(plugin_link_dir, 'w') as clobber_file:
clobber_file.write('clobbered!')
result = run_assetlink().collection[0]
assert result == 'Could not link "staticstuff": %s exists and is not a symlink\n' % (
plugin_link_dir)
- with file(plugin_link_dir, 'r') as clobber_file:
+ with open(plugin_link_dir, 'r') as clobber_file:
assert clobber_file.read() == 'clobbered!'
Test that the staticdirect utilities pull up the right things
"""
result = json.loads(
- static_plugin_app.get('/staticstuff/').body)
+ static_plugin_app.get('/staticstuff/').body.decode())
assert len(result) == 2
assert result['mgoblin_bunny_pic'] == '/test_static/images/bunny_pic.png'
assert result['plugin_bunny_css'] == \
'/plugin_static/staticstuff/css/bunnify.css'
-
# 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
import pytest
from datetime import date, timedelta
from webtest import AppError
response = self.test_app.get('/')
assert response.status == "200 OK"
- assert "You are Banned" in response.body
+ assert b"You are Banned" in response.body
# Then test what happens when that ban has an expiration date which
# hasn't happened yet
#----------------------------------------------------------------------
response = self.test_app.get('/')
assert response.status == "200 OK"
- assert "You are Banned" in response.body
+ assert b"You are Banned" in response.body
# Then test what happens when that ban has an expiration date which
# has already happened
response = self.test_app.get('/')
assert response.status == "302 FOUND"
- assert not "You are Banned" in response.body
+ assert not b"You are Banned" in response.body
def testVariousPrivileges(self):
# The various actions that require privileges (ex. reporting,
#----------------------------------------------------------------------
with pytest.raises(AppError) as excinfo:
response = self.test_app.get('/submit/')
- assert 'Bad response: 403 FORBIDDEN' in str(excinfo)
+ excinfo = str(excinfo) if six.PY2 else str(excinfo).encode('ascii')
+ assert b'Bad response: 403 FORBIDDEN' in excinfo
with pytest.raises(AppError) as excinfo:
response = self.do_post({'upload_files':[('file',GOOD_JPG)],
'title':u'Normal Upload 1'},
url='/submit/')
- assert 'Bad response: 403 FORBIDDEN' in str(excinfo)
+ excinfo = str(excinfo) if six.PY2 else str(excinfo).encode('ascii')
+ assert b'Bad response: 403 FORBIDDEN' in excinfo
# Test that a user cannot comment without the commenter privilege
#----------------------------------------------------------------------
media_uri_slug = '/u/{0}/m/{1}/'.format(self.admin_user.username,
media_entry.slug)
response = self.test_app.get(media_uri_slug)
- assert not "Add a comment" in response.body
+ assert not b"Add a comment" in response.body
self.query_for_users()
with pytest.raises(AppError) as excinfo:
response = self.test_app.post(
media_uri_id + 'comment/add/',
{'comment_content': u'Test comment #42'})
- assert 'Bad response: 403 FORBIDDEN' in str(excinfo)
+ excinfo = str(excinfo) if six.PY2 else str(excinfo).encode('ascii')
+ assert b'Bad response: 403 FORBIDDEN' in excinfo
# Test that a user cannot report without the reporter privilege
#----------------------------------------------------------------------
with pytest.raises(AppError) as excinfo:
response = self.test_app.get(media_uri_slug+"report/")
- assert 'Bad response: 403 FORBIDDEN' in str(excinfo)
+ excinfo = str(excinfo) if six.PY2 else str(excinfo).encode('ascii')
+ assert b'Bad response: 403 FORBIDDEN' in excinfo
with pytest.raises(AppError) as excinfo:
response = self.do_post(
{'report_reason':u'Testing Reports #1',
'reporter_id':u'3'},
url=(media_uri_slug+"report/"))
- assert 'Bad response: 403 FORBIDDEN' in str(excinfo)
+ excinfo = str(excinfo) if six.PY2 else str(excinfo).encode('ascii')
+ assert b'Bad response: 403 FORBIDDEN' in excinfo
# Test that a user cannot access the moderation pages w/o moderator
# or admin privileges
#----------------------------------------------------------------------
with pytest.raises(AppError) as excinfo:
response = self.test_app.get("/mod/users/")
- assert 'Bad response: 403 FORBIDDEN' in str(excinfo)
+ excinfo = str(excinfo) if six.PY2 else str(excinfo).encode('ascii')
+ assert b'Bad response: 403 FORBIDDEN' in excinfo
with pytest.raises(AppError) as excinfo:
response = self.test_app.get("/mod/reports/")
- assert 'Bad response: 403 FORBIDDEN' in str(excinfo)
+ excinfo = str(excinfo) if six.PY2 else str(excinfo).encode('ascii')
+ assert b'Bad response: 403 FORBIDDEN' in excinfo
with pytest.raises(AppError) as excinfo:
response = self.test_app.get("/mod/media/")
- assert 'Bad response: 403 FORBIDDEN' in str(excinfo)
+ excinfo = str(excinfo) if six.PY2 else str(excinfo).encode('ascii')
+ assert b'Bad response: 403 FORBIDDEN' in excinfo
with pytest.raises(AppError) as excinfo:
response = self.test_app.get("/mod/users/1/")
- assert 'Bad response: 403 FORBIDDEN' in str(excinfo)
+ excinfo = str(excinfo) if six.PY2 else str(excinfo).encode('ascii')
+ assert b'Bad response: 403 FORBIDDEN' in excinfo
with pytest.raises(AppError) as excinfo:
response = self.test_app.get("/mod/reports/1/")
- assert 'Bad response: 403 FORBIDDEN' in str(excinfo)
+ excinfo = str(excinfo) if six.PY2 else str(excinfo).encode('ascii')
+ assert b'Bad response: 403 FORBIDDEN' in excinfo
self.query_for_users()
'targeted_user':self.admin_user.id},
url='/mod/reports/1/')
self.query_for_users()
- assert 'Bad response: 403 FORBIDDEN' in str(excinfo)
+ excinfo = str(excinfo) if six.PY2 else str(excinfo).encode('ascii')
+ assert b'Bad response: 403 FORBIDDEN' in excinfo
# 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"
# 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
+import pytest
+
+pytestmark = pytest.mark.skipif(six.PY3, reason='needs sqlalchemy.migrate')
+
import copy
from sqlalchemy import (
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql import select, insert
-from migrate import changeset
+if six.PY2:
+ from migrate import changeset
from mediagoblin.db.base import GMGTableBase
from mediagoblin.db.migration_tools import MigrationManager, RegisterMigration
for level in result:
- for exit_name, to_level in level['exits'].iteritems():
+ for exit_name, to_level in six.iteritems(level['exits']):
# Insert the level exit
db_conn.execute(
level_exits.insert().values(
import tempfile
import pytest
+import six
+
from werkzeug.utils import secure_filename
from mediagoblin import storage
storage.clean_listy_filepath(['../../', 'linooks.jpg'])
-class FakeStorageSystem():
+class FakeStorageSystem(object):
def __init__(self, foobie, blech, **kwargs):
self.foobie = foobie
self.blech = blech
'mediagoblin.tests.test_storage:FakeStorageSystem'})
assert this_storage.foobie == 'eiboof'
assert this_storage.blech == 'hcelb'
- assert unicode(this_storage.__class__) == \
- u'mediagoblin.tests.test_storage.FakeStorageSystem'
+ assert six.text_type(this_storage.__class__) == \
+ u"<class 'mediagoblin.tests.test_storage.FakeStorageSystem'>"
##########################
with this_storage.get_file(filepath, 'r') as our_file:
assert our_file.read() == 'First file'
assert os.path.exists(os.path.join(tmpdir, 'dir1/dir2/ourfile.txt'))
- with file(os.path.join(tmpdir, 'dir1/dir2/ourfile.txt'), 'r') as our_file:
+ with open(os.path.join(tmpdir, 'dir1/dir2/ourfile.txt'), 'r') as our_file:
assert our_file.read() == 'First file'
# Write to the same path but try to get a unique file.
with this_storage.get_file(new_filepath, 'r') as our_file:
assert our_file.read() == 'Second file'
assert os.path.exists(os.path.join(tmpdir, *new_filepath))
- with file(os.path.join(tmpdir, *new_filepath), 'r') as our_file:
+ with open(os.path.join(tmpdir, *new_filepath), 'r') as our_file:
assert our_file.read() == 'Second file'
# Read from an existing file
manually_written_file = os.makedirs(
os.path.join(tmpdir, 'testydir'))
- with file(os.path.join(tmpdir, 'testydir/testyfile.txt'), 'w') as testyfile:
+ with open(os.path.join(tmpdir, 'testydir/testyfile.txt'), 'w') as testyfile:
testyfile.write('testy file! so testy.')
with this_storage.get_file(['testydir', 'testyfile.txt']) as testyfile:
this_storage.copy_locally(filepath, new_file_dest)
this_storage.delete_file(filepath)
- assert file(new_file_dest).read() == 'Testing this file'
+ assert open(new_file_dest).read() == 'Testing this file'
os.remove(new_file_dest)
os.rmdir(dest_tmpdir)
def _test_copy_local_to_storage_works(tmpdir, this_storage):
local_filename = tempfile.mktemp()
- with file(local_filename, 'w') as tmpfile:
+ with open(local_filename, 'w') as tmpfile:
tmpfile.write('haha')
this_storage.copy_local_to_storage(
os.remove(local_filename)
- assert file(
+ assert open(
os.path.join(tmpdir, 'dir1/dir2/copiedto.txt'),
'r').read() == 'haha'
# 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 sys
-reload(sys)
-sys.setdefaultencoding('utf-8')
+import six
+
+if six.PY2: # this hack only work in Python 2
+ import sys
+ reload(sys)
+ sys.setdefaultencoding('utf-8')
-import urlparse
import os
import pytest
+import six.moves.urllib.parse as urlparse
+
from mediagoblin.tests.tools import fixture_add_user
from mediagoblin import mg_globals
from mediagoblin.db.models import MediaEntry, User
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']
# 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 mock
+try:
+ import mock
+except ImportError:
+ import unittest.mock as mock
import email
import pytest
import smtplib
import pkg_resources
+import six
+
from mediagoblin.tests.tools import get_app
from mediagoblin.tools import common, url, translate, mail, text, testing
assert message['From'] == "sender@mediagoblin.example.org"
assert message['To'] == "amanda@example.org, akila@example.org"
assert message['Subject'] == "Testing is so much fun!"
- assert message.get_payload(decode=True) == """HAYYY GUYS!
+ assert message.get_payload(decode=True) == b"""HAYYY GUYS!
I hope you like unit tests JUST AS MUCH AS I DO!"""
assert mbox_message['From'] == "sender@mediagoblin.example.org"
assert mbox_message['To'] == "amanda@example.org, akila@example.org"
assert mbox_message['Subject'] == "Testing is so much fun!"
- assert mbox_message.get_payload(decode=True) == """HAYYY GUYS!
+ assert mbox_message.get_payload(decode=True) == b"""HAYYY GUYS!
I hope you like unit tests JUST AS MUCH AS I DO!"""
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
# kill a workbench
this_workbench = self.workbench_manager.create()
tmpfile_name = this_workbench.joinpath('temp.txt')
- tmpfile = file(tmpfile_name, 'w')
+ tmpfile = open(tmpfile_name, 'w')
with tmpfile:
tmpfile.write('lollerskates')
import pkg_resources
import shutil
+import six
from paste.deploy import loadapp
from webtest import TestApp
"""
Very simply install fixtures in the database
"""
- for collection_name, collection_fixtures in fixtures.iteritems():
+ for collection_name, collection_fixtures in six.iteritems(fixtures):
collection = db[collection_name]
for fixture in collection_fixtures:
collection.insert(fixture)
{'id': 'foo',
'some_field': 'some_value'},]}
"""
- for collection_name, collection_data in expected.iteritems():
+ for collection_name, collection_data in six.iteritems(expected):
collection = db[collection_name]
for expected_document in collection_data:
document = collection.query.filter_by(id=expected_document['id']).first()
def create_key(key_dir, key_filepath):
global __itsda_secret
- old_umask = os.umask(077)
+ old_umask = os.umask(0o77)
key_file = None
try:
if not os.path.isdir(key_dir):
key = str(getrandbits(192))
key_file = tempfile.NamedTemporaryFile(dir=key_dir, suffix='.bin',
delete=False)
- key_file.write(key)
+ key_file.write(key.encode('ascii'))
key_file.flush()
os.rename(key_file.name, key_filepath)
key_file.close()
key_filepath = os.path.join(key_dir, 'itsdangeroussecret.bin')
try:
load_key(key_filepath)
- except IOError, error:
+ except IOError as error:
if error.errno != errno.ENOENT:
raise
create_key(key_dir, key_filepath)
# 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 exifread import process_file
from exifread.utils import Ratio
Returns EXIF tags found in file at ``filename``
"""
try:
- with file(filename) as image:
+ with open(filename, 'rb') as image:
return process_file(image, details=False)
except IOError:
raise BadMediaFail(_('Could not read the image file.'))
'Thumbnail JPEGInterchangeFormat']
return dict((key, _ifd_tag_to_dict(value)) for (key, value)
- in exif.iteritems() if key not in disabled_tags)
+ in six.iteritems(exif) if key not in disabled_tags)
def _ifd_tag_to_dict(tag):
'field_length': tag.field_length,
'values': None}
- if isinstance(tag.printable, str):
+ if isinstance(tag.printable, six.binary_type):
# Force it to be decoded as UTF-8 so that it'll fit into the DB
data['printable'] = tag.printable.decode('utf8', 'replace')
data['values'] = [_ratio_to_list(val) if isinstance(val, Ratio) else val
for val in tag.values]
else:
- if isinstance(tag.values, str):
+ if isinstance(tag.values, six.binary_type):
# Force UTF-8, so that it fits into the DB
data['values'] = tag.values.decode('utf8', 'replace')
else:
def get_useful(tags):
- return dict((key, tag) for (key, tag) in tags.iteritems())
+ from collections import OrderedDict
+ return OrderedDict((key, tag) for (key, tag) in six.iteritems(tags))
def get_gps_data(tags):
'latitude': tags['GPS GPSLatitude'],
'longitude': tags['GPS GPSLongitude']}
- for key, dat in dms_data.iteritems():
+ for key, dat in six.iteritems(dms_data):
gps_data[key] = (
lambda v:
float(v[0].num) / float(v[0].den) \
# 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, unicode_literals
+
import six
import smtplib
import sys
-from email.MIMEText import MIMEText
from mediagoblin import mg_globals, messages
+from mediagoblin._compat import MIMEText
from mediagoblin.tools import common
### ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
EMAIL_TEST_INBOX.append(message)
elif mg_globals.app_config['email_debug_mode']:
- print u"===== Email ====="
- print u"From address: %s" % message['From']
- print u"To addresses: %s" % message['To']
- print u"Subject: %s" % message['Subject']
- print u"-- Body: --"
- print message.get_payload(decode=True)
+ print("===== Email =====")
+ print("From address: %s" % message['From'])
+ print("To addresses: %s" % message['To'])
+ print("Subject: %s" % message['Subject'])
+ print("-- Body: --")
+ print(message.get_payload(decode=True))
return mhost.sendmail(from_addr, to_addrs, message.as_string())
if mg_globals.app_config['email_debug_mode']:
# DEBUG message, no need to translate
messages.add_message(request, messages.DEBUG,
- u"This instance is running in email debug mode. "
- u"The email will be on the console of the server process.")
+ "This instance is running in email debug mode. "
+ "The email will be on the console of the server process.")
# along with this program. If not, see <http://www.gnu.org/licenses/>.
+from io import open
import os
import copy
import json
os.path.sep.
"""
filename = resource_filename(package, os.path.sep.join(resource_path))
- return file(filename).read()
+ return open(filename, encoding="utf-8").read()
def load_resource_json(package, resource_path):
"""
import urllib
import copy
from math import ceil, floor
-from itertools import izip, count
+from itertools import count
from werkzeug.datastructures import MultiDict
+from six.moves import zip
+
PAGINATION_DEFAULT_PER_PAGE = 30
if jump_to_id:
cursor = copy.copy(self.cursor)
- for (doc, increment) in izip(cursor, count(0)):
+ for (doc, increment) in list(zip(cursor, count(0))):
if doc.id == jump_to_id:
self.page = 1 + int(floor(increment / self.per_page))
import json
import traceback
-from urllib2 import urlopen, Request, HTTPError
-from urllib import urlencode
+from six.moves.urllib import request, parse
_log = logging.getLogger(__name__)
data_parser: The parser function that is used to parse the `data`
argument
'''
- data_parser = kw.get('data_parser', urlencode)
+ data_parser = kw.get('data_parser', parse.urlencode)
headers = kw.get('headers', {})
- return Request(url, data_parser(data), headers=headers)
+ return request.Request(url, data_parser(data), headers=headers)
def json_processing_callback(entry):
data_parser=json.dumps)
try:
- urlopen(request)
+ request.urlopen(request)
_log.debug('Processing callback for {0} sent'.format(entry))
return True
- except HTTPError:
+ except request.HTTPError:
_log.error('Failed to send callback: {0}'.format(
traceback.format_exc()))
import json
+import six
import werkzeug.utils
from werkzeug.wrappers import Response as wz_Response
from mediagoblin.tools.template import render_template
'Access-Control-Allow-Origin': '*',
'Access-Control-Allow-Methods': 'POST, GET, OPTIONS',
'Access-Control-Allow-Headers': 'Content-Type, X-Requested-With'}
- for key, value in cors_headers.iteritems():
+ for key, value in six.iteritems(cors_headers):
response.headers.set(key, value)
return response
import logging
+import six
+
_log = logging.getLogger(__name__)
def __init__(self, domains):
self.domains = dict(
[(key, value.rstrip('/'))
- for key, value in domains.iteritems()])
+ for key, value in six.iteritems(domains)])
self.cache = {}
def __call__(self, filepath, domain=None):
# 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
import jinja2
from jinja2.ext import Extension
from mediagoblin.tools.timesince import timesince
from mediagoblin.meddleware.csrf import render_csrf_form_token
-
SETUP_JINJA_ENVS = {}
'jinja2.ext.i18n', 'jinja2.ext.autoescape',
TemplateHookExtension] + local_exts)
- template_env.install_gettext_callables(
- mg_globals.thread_scope.translations.ugettext,
- mg_globals.thread_scope.translations.ungettext)
+ if six.PY2:
+ template_env.install_gettext_callables(mg_globals.thread_scope.translations.ugettext,
+ mg_globals.thread_scope.translations.ungettext)
+ else:
+ template_env.install_gettext_callables(mg_globals.thread_scope.translations.gettext,
+ mg_globals.thread_scope.translations.ngettext)
# All templates will know how to ...
# ... fetch all waiting messages and remove them from the queue
import gettext
import pkg_resources
+import six
from babel import localedata
from babel.support import LazyProxy
"""
Like LazyProxy, except that it doesn't cache the value ;)
"""
- @property
- def value(self):
- return self._func(*self._args, **self._kwargs)
+ def __init__(self, func, *args, **kwargs):
+ super(ReallyLazyProxy, self).__init__(func, *args, **kwargs)
+ object.__setattr__(self, '_is_cache_enabled', False)
def __repr__(self):
return "<%s for %s(%r, %r)>" % (
The reason we can't have a global ugettext method is because
mg_globals gets swapped out by the application per-request.
"""
- return mg_globals.thread_scope.translations.ugettext(
- *args, **kwargs)
+ if six.PY2:
+ return mg_globals.thread_scope.translations.ugettext(*args, **kwargs)
+ return mg_globals.thread_scope.translations.gettext(*args, **kwargs)
def pass_to_ungettext(*args, **kwargs):
"""
The reason we can't have a global ugettext method is because
mg_globals gets swapped out by the application per-request.
"""
- return mg_globals.thread_scope.translations.ungettext(
- *args, **kwargs)
+ if six.PY2:
+ return mg_globals.thread_scope.translations.ungettext(*args, **kwargs)
+ return mg_globals.thread_scope.translations.ngettext(*args, **kwargs)
def lazy_pass_to_ugettext(*args, **kwargs):
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 shutil
import tempfile
+import six
+
+from mediagoblin._compat import py2_unicode
# Actual workbench stuff
# ----------------------
+
+@py2_unicode
class Workbench(object):
"""
Represent the directory for the workbench
"""
self.dir = dir
- def __unicode__(self):
- return unicode(self.dir)
-
def __str__(self):
- return str(self.dir)
+ return six.text_type(self.dir)
def __repr__(self):
try:
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))
debug = false
[pipeline:main]
-pipeline = errors routing
-
-[composite:routing]
-use = egg:Paste#urlmap
-/ = mediagoblin
-/mgoblin_media/ = publicstore_serve
-/mgoblin_static/ = mediagoblin_static
-/theme_static/ = theme_static
-/plugin_static/ = plugin_static
+# pipeline = errors mediagoblin
+pipeline = mediagoblin
[app:mediagoblin]
use = egg:mediagoblin#app
config = %(here)s/mediagoblin_local.ini %(here)s/mediagoblin.ini
+# static paths
+/mgoblin_media = %(here)s/user_dev/media/public
+/mgoblin_static = %(here)s/mediagoblin/static
+/theme_static = %(here)s/user_dev/theme_static
+/plugin_static = %(here)s/user_dev/plugin_static
[loggers]
keys = root
[formatter_generic]
format = %(asctime)s %(levelname)-7.7s [%(name)s] %(message)s
-[app:publicstore_serve]
-use = egg:Paste#static
-document_root = %(here)s/user_dev/media/public/
-cache_max_age = 604800
-
-[app:mediagoblin_static]
-use = egg:Paste#static
-document_root = %(here)s/mediagoblin/static/
-cache_max_age = 86400
-
-[app:theme_static]
-use = egg:Paste#static
-document_root = %(here)s/user_dev/theme_static/
-cache_max_age = 86400
-
-[app:plugin_static]
-use = egg:Paste#static
-document_root = %(here)s/user_dev/plugin_static/
-cache_max_age = 86400
-
[filter:errors]
use = egg:mediagoblin#errors
debug = false
# The server that is run by default.
# By default, should only be accessable locally
[server:main]
-use = egg:Paste#http
+use = egg:gunicorn
host = 127.0.0.1
port = 6543
+# Gunicorn settings. See http://docs.gunicorn.org/en/19.0/settings.html
+# for more information about configuring Gunicorn
+proc_name = gmg
+reload = true
+accesslog = -
#######################
# Helper server configs
# 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
+
from setuptools import setup, find_packages
+from io import open
import os
import re
+import sys
+
+PY2 = sys.version_info[0] == 2 # six is not installed yet
+
READMEFILE = "README"
VERSIONFILE = os.path.join("mediagoblin", "_version.py")
VSRE = r"^__version__ = ['\"]([^'\"]*)['\"]"
def get_version():
- verstrline = open(VERSIONFILE, "rt").read()
+ with open(VERSIONFILE, "rt") as fobj:
+ verstrline = fobj.read()
mo = re.search(VSRE, verstrline, re.M)
if mo:
return mo.group(1)
raise RuntimeError("Unable to find version string in %s." %
VERSIONFILE)
+py2_only_install_requires = []
+if PY2:
+ py2_only_install_requires.append('argparse') # only for < 2.7
+ py2_only_install_requires.append('PasteScript')
+ # newer sqlalchemy-migrate requires pbr which BREAKS EVERYTHING AND IS
+ # TERRIBLE AND IS THE END OF ALL THINGS
+ # I'd love to remove this restriction.
+ py2_only_install_requires.append('sqlalchemy-migrate<0.8')
+ # # Annoying. Please remove once we can! We only indirectly
+ # # use pbr, and currently it breaks things, presumably till
+ # # their next release.
+ # py2_only_install_requires.append('pbr==0.5.22')
+ py2_only_install_requires.append('mock') # mock is in the stdlib for 3.3+
+
+install_requires = [
+ 'gunicorn',
+ 'alembic==0.6.6',
+ 'python-dateutil',
+ 'wtforms',
+ 'py-bcrypt',
+ 'pytest>=2.3.1',
+ 'pytest-xdist',
+ 'werkzeug>=0.7',
+ 'celery>=3.0',
+ 'kombu',
+ 'jinja2',
+ 'Babel>=1.3',
+ 'webtest<2',
+ 'ConfigObj',
+ 'Markdown',
+ 'sqlalchemy<0.9.0, >0.8.0',
+ 'itsdangerous',
+ 'pytz',
+ # PLEASE change this when we can; a dependency is forcing us to set this
+ # specific number and it is breaking setup.py develop
+ 'six==1.5.2',
+ 'oauthlib',
+ 'unidecode',
+ 'jsonschema',
+ 'ExifRead', # TODO(berker): Install develop branch for Python 3
+ 'PasteDeploy',
+ 'requests',
+ 'pyld',
+ # This is optional:
+ # 'translitcodec',
+ # For now we're expecting that users will install this from
+ # their package managers.
+ # 'lxml',
+ # 'Pillow',
+] + py2_only_install_requires
+
+with open(READMEFILE, encoding="utf-8") as fobj:
+ long_description = fobj.read()
+
try:
setup(
name="mediagoblin",
zip_safe=False,
include_package_data = True,
# scripts and dependencies
- install_requires=[
- 'setuptools',
- 'python-dateutil',
- 'PasteScript',
- 'wtforms',
- 'py-bcrypt',
- 'pytest>=2.3.1',
- 'pytest-xdist',
- 'werkzeug>=0.7',
- 'celery>=3.0',
- 'kombu',
- 'jinja2',
- 'sphinx',
- 'Babel>=1.0',
- 'argparse',
- 'webtest<2',
- 'ConfigObj',
- 'Markdown',
- 'sqlalchemy<0.9.0, >0.8.0',
- # newer sqlalchemy-migrate requires pbr which BREAKS EVERYTHING AND IS
- # TERRIBLE AND IS THE END OF ALL THINGS
- # I'd love to remove this restriction.
- 'sqlalchemy-migrate<0.8',
- 'mock',
- 'itsdangerous',
- 'pytz',
- 'six>=1.4.1',
- 'oauthlib',
- 'unidecode',
- 'jsonschema',
- 'requests',
- 'pyld',
- 'ExifRead',
-
- # PLEASE change this when we can; a dependency is forcing us to set this
- # specific number and it is breaking setup.py develop
- 'six==1.5.2'
-
- ## Annoying. Please remove once we can! We only indirectly
- ## use pbr, and currently it breaks things, presumably till
- ## their next release.
- # 'pbr==0.5.22',
-
- ## This is optional!
- # 'translitcodec',
- ## For now we're expecting that users will install this from
- ## their package managers.
- # 'lxml',
- # 'PIL',
- ],
- # requires=['gst'],
+ install_requires=install_requires,
test_suite='nose.collector',
entry_points="""\
[console_scripts]
author='Free Software Foundation and contributors',
author_email='cwebber@gnu.org',
url="http://mediagoblin.org/",
- long_description=open(READMEFILE).read(),
+ long_description=long_description,
description='MediaGoblin is a web application for publishing all kinds of media',
classifiers=[
"Development Status :: 3 - Alpha",
"License :: OSI Approved :: GNU Affero General Public License v3 or later (AGPLv3+)",
"Operating System :: OS Independent",
"Programming Language :: Python",
+ 'Programming Language :: Python :: 2',
'Programming Language :: Python :: 2.6',
'Programming Language :: Python :: 2.7',
+ 'Programming Language :: Python :: 3',
+ 'Programming Language :: Python :: 3.3',
+ 'Programming Language :: Python :: 3.4',
"Topic :: Internet :: WWW/HTTP :: Dynamic Content"
],
)
-except TypeError, e:
+except TypeError as e:
+ import sys
+
# Check if the problem is caused by the sqlalchemy/setuptools conflict
msg_as_str = str(e)
if not (msg_as_str == 'dist must be a Distribution instance'):
raise
# If so, tell the user it is OK to just run the script again.
- print "\n\n---------- NOTE ----------"
- print "The setup.py command you ran failed."
- print ""
- print ("It is a known possible failure. Just run it again. It works the "
- "second time.")
- import sys
+ print("\n\n---------- NOTE ----------", file=sys.stderr)
+ print("The setup.py command you ran failed.\n", file=sys.stderr)
+ print("It is a known possible failure. Just run it again. It works the "
+ "second time.", file=sys.stderr)
sys.exit(1)
--- /dev/null
+[tox]
+envlist = py27, py33
+usedevelop = True
+sitepackages = False
+
+[testenv]
+whitelist_externals = sh
+commands = py.test ./mediagoblin/tests --boxed
+deps =
+ git+https://github.com/ianare/exif-py.git@develop
+ lxml
+ Pillow
projects / mediagoblin.git / commitdiff