Finished splitting util.py into separate files.
This commit is contained in:
Aaron Williamson
parent
03ae172a60
commit
152a3bfaa3
@ -20,8 +20,9 @@ import urllib
|
||||
import routes
|
||||
from webob import Request, exc
|
||||
|
||||
from mediagoblin import routing, util, middleware
|
||||
from mediagoblin.tools import translate, template
|
||||
from mediagoblin import routing, middleware
|
||||
from mediagoblin.tools import common, translate, template, response
|
||||
from mediagoblin.tools import request as mg_request
|
||||
from mediagoblin.mg_globals import setup_globals
|
||||
from mediagoblin.init.celery import setup_celery_from_config
|
||||
from mediagoblin.init import (get_jinja_loader, get_staticdirector,
|
||||
@ -99,7 +100,7 @@ class MediaGoblinApp(object):
|
||||
setup_workbench()
|
||||
|
||||
# instantiate application middleware
|
||||
self.middleware = [util.import_component(m)(self)
|
||||
self.middleware = [common.import_component(m)(self)
|
||||
for m in middleware.ENABLED_MIDDLEWARE]
|
||||
|
||||
|
||||
@ -131,7 +132,7 @@ class MediaGoblinApp(object):
|
||||
request.db = self.db
|
||||
request.staticdirect = self.staticdirector
|
||||
|
||||
util.setup_user_in_request(request)
|
||||
mg_request.setup_user_in_request(request)
|
||||
|
||||
# No matching page?
|
||||
if route_match is None:
|
||||
@ -149,9 +150,9 @@ class MediaGoblinApp(object):
|
||||
|
||||
# Okay, no matches. 404 time!
|
||||
request.matchdict = {} # in case our template expects it
|
||||
return util.render_404(request)(environ, start_response)
|
||||
return response.render_404(request)(environ, start_response)
|
||||
|
||||
controller = util.import_component(route_match['controller'])
|
||||
controller = common.import_component(route_match['controller'])
|
||||
request.start_response = start_response
|
||||
|
||||
# get the response from the controller
|
||||
|
||||
@ -19,7 +19,7 @@ import random
|
||||
|
||||
import bcrypt
|
||||
|
||||
from mediagoblin.util import send_email
|
||||
from mediagoblin.tools.mail import send_email
|
||||
from mediagoblin.tools.template import render_template
|
||||
from mediagoblin import mg_globals
|
||||
|
||||
|
||||
@ -21,7 +21,7 @@ from webob import exc
|
||||
|
||||
from mediagoblin import messages
|
||||
from mediagoblin import mg_globals
|
||||
from mediagoblin.util import render_to_response, redirect, render_404
|
||||
from mediagoblin.tools.response import render_to_response, redirect, render_404
|
||||
from mediagoblin.tools.translate import pass_to_ugettext as _
|
||||
from mediagoblin.db.util import ObjectId, InvalidId
|
||||
from mediagoblin.auth import lib as auth_lib
|
||||
|
||||
@ -15,7 +15,7 @@
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
from mediagoblin.db.util import RegisterMigration
|
||||
from mediagoblin.util import cleaned_markdown_conversion
|
||||
from mediagoblin.tools.text import cleaned_markdown_conversion
|
||||
|
||||
|
||||
# Please see mediagoblin/tests/test_migrations.py for some examples of
|
||||
|
||||
@ -18,14 +18,12 @@ import datetime, uuid
|
||||
|
||||
from mongokit import Document
|
||||
|
||||
from mediagoblin import util
|
||||
from mediagoblin.auth import lib as auth_lib
|
||||
from mediagoblin import mg_globals
|
||||
from mediagoblin.db import migrations
|
||||
from mediagoblin.db.util import ASCENDING, DESCENDING, ObjectId
|
||||
from mediagoblin.util import Pagination
|
||||
from mediagoblin.util import DISPLAY_IMAGE_FETCHING_ORDER
|
||||
from mediagoblin.tools import url
|
||||
from mediagoblin.tools.pagination import Pagination
|
||||
from mediagoblin.tools import url, common
|
||||
|
||||
###################
|
||||
# Custom validators
|
||||
@ -220,7 +218,7 @@ class MediaEntry(Document):
|
||||
return self.db.MediaComment.find({
|
||||
'media_entry': self['_id']}).sort('created', DESCENDING)
|
||||
|
||||
def get_display_media(self, media_map, fetch_order=DISPLAY_IMAGE_FETCHING_ORDER):
|
||||
def get_display_media(self, media_map, fetch_order=common.DISPLAY_IMAGE_FETCHING_ORDER):
|
||||
"""
|
||||
Find the best media for display.
|
||||
|
||||
@ -234,7 +232,7 @@ class MediaEntry(Document):
|
||||
"""
|
||||
media_sizes = media_map.keys()
|
||||
|
||||
for media_size in DISPLAY_IMAGE_FETCHING_ORDER:
|
||||
for media_size in common.DISPLAY_IMAGE_FETCHING_ORDER:
|
||||
if media_size in media_sizes:
|
||||
return media_map[media_size]
|
||||
|
||||
@ -304,7 +302,7 @@ class MediaEntry(Document):
|
||||
Get the exception that's appropriate for this error
|
||||
"""
|
||||
if self['fail_error']:
|
||||
return util.import_component(self['fail_error'])
|
||||
return common.import_component(self['fail_error'])
|
||||
|
||||
|
||||
class MediaComment(Document):
|
||||
|
||||
@ -17,7 +17,7 @@
|
||||
|
||||
from webob import exc
|
||||
|
||||
from mediagoblin.util import redirect, render_404
|
||||
from mediagoblin.tools.response import redirect, render_404
|
||||
from mediagoblin.db.util import ObjectId, InvalidId
|
||||
|
||||
|
||||
|
||||
@ -14,12 +14,10 @@
|
||||
# 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 wtforms
|
||||
|
||||
from mediagoblin.util import tag_length_validator, TOO_LONG_TAG_WARNING
|
||||
from mediagoblin.util import fake_ugettext_passthrough as _
|
||||
|
||||
from mediagoblin.tools.text import tag_length_validator, TOO_LONG_TAG_WARNING
|
||||
from mediagoblin.tools.translate import fake_ugettext_passthrough as _
|
||||
|
||||
class EditForm(wtforms.Form):
|
||||
title = wtforms.TextField(
|
||||
|
||||
@ -25,14 +25,15 @@ from werkzeug.utils import secure_filename
|
||||
|
||||
from mediagoblin import messages
|
||||
from mediagoblin import mg_globals
|
||||
from mediagoblin.util import (
|
||||
render_to_response, redirect, clean_html, convert_to_tag_list_of_dicts,
|
||||
media_tags_as_string, cleaned_markdown_conversion)
|
||||
from mediagoblin.util import pass_to_ugettext as _
|
||||
|
||||
from mediagoblin.edit import forms
|
||||
from mediagoblin.edit.lib import may_edit_media
|
||||
from mediagoblin.decorators import require_active_login, get_user_media_entry
|
||||
|
||||
from mediagoblin.tools.response import render_to_response, redirect
|
||||
from mediagoblin.tools.translate import pass_to_ugettext as _
|
||||
from mediagoblin.tools.text import (
|
||||
clean_html, convert_to_tag_list_of_dicts,
|
||||
media_tags_as_string, cleaned_markdown_conversion)
|
||||
|
||||
@get_user_media_entry
|
||||
@require_active_login
|
||||
|
||||
@ -16,7 +16,8 @@
|
||||
|
||||
from mediagoblin.db.util import DESCENDING
|
||||
|
||||
from mediagoblin.util import Pagination, render_to_response
|
||||
from mediagoblin.tools.pagination import Pagination
|
||||
from mediagoblin.tools.response import render_to_response
|
||||
from mediagoblin.decorators import uses_pagination
|
||||
|
||||
from werkzeug.contrib.atom import AtomFeed
|
||||
|
||||
@ -21,7 +21,7 @@ import uuid
|
||||
|
||||
from werkzeug.utils import secure_filename
|
||||
|
||||
from mediagoblin import util
|
||||
from mediagoblin.tools import common
|
||||
|
||||
########
|
||||
# Errors
|
||||
@ -236,5 +236,5 @@ def storage_system_from_config(config_section):
|
||||
else:
|
||||
storage_class = 'mediagoblin.storage.filestorage:BasicFileStorage'
|
||||
|
||||
storage_class = util.import_component(storage_class)
|
||||
storage_class = common.import_component(storage_class)
|
||||
return storage_class(**config_params)
|
||||
|
||||
@ -17,7 +17,7 @@
|
||||
|
||||
import wtforms
|
||||
|
||||
from mediagoblin.util import tag_length_validator
|
||||
from mediagoblin.tools.text import tag_length_validator
|
||||
from mediagoblin.tools.translate import fake_ugettext_passthrough as _
|
||||
|
||||
|
||||
|
||||
@ -22,10 +22,9 @@ from cgi import FieldStorage
|
||||
from werkzeug.utils import secure_filename
|
||||
|
||||
from mediagoblin.db.util import ObjectId
|
||||
from mediagoblin.util import (
|
||||
render_to_response, redirect, cleaned_markdown_conversion, \
|
||||
convert_to_tag_list_of_dicts)
|
||||
from mediagoblin.tools.text import cleaned_markdown_conversion, convert_to_tag_list_of_dicts
|
||||
from mediagoblin.tools.translate import pass_to_ugettext as _
|
||||
from mediagoblin.tools.response import render_to_response, redirect
|
||||
from mediagoblin.decorators import require_active_login
|
||||
from mediagoblin.submit import forms as submit_forms, security
|
||||
from mediagoblin.process_media import process_media, mark_entry_failed
|
||||
|
||||
@ -22,7 +22,7 @@ from nose.tools import assert_equal
|
||||
from mediagoblin.auth import lib as auth_lib
|
||||
from mediagoblin.tests.tools import setup_fresh_app
|
||||
from mediagoblin import mg_globals
|
||||
from mediagoblin.tools import template
|
||||
from mediagoblin.tools import template, mail
|
||||
|
||||
|
||||
########################
|
||||
@ -171,8 +171,8 @@ def test_register_views(test_app):
|
||||
assert request.session['user_id'] == unicode(new_user['_id'])
|
||||
|
||||
## Make sure we get email confirmation, and try verifying
|
||||
assert len(template.EMAIL_TEST_INBOX) == 1
|
||||
message = template.EMAIL_TEST_INBOX.pop()
|
||||
assert len(mail.EMAIL_TEST_INBOX) == 1
|
||||
message = mail.EMAIL_TEST_INBOX.pop()
|
||||
assert message['To'] == 'happygrrl@example.org'
|
||||
email_context = template.TEMPLATE_TEST_CONTEXT[
|
||||
'mediagoblin/auth/verification_email.txt']
|
||||
@ -254,8 +254,8 @@ def test_register_views(test_app):
|
||||
'mediagoblin/auth/fp_email_sent.html')
|
||||
|
||||
## Make sure link to change password is sent by email
|
||||
assert len(template.EMAIL_TEST_INBOX) == 1
|
||||
message = template.EMAIL_TEST_INBOX.pop()
|
||||
assert len(mail.EMAIL_TEST_INBOX) == 1
|
||||
message = mail.EMAIL_TEST_INBOX.pop()
|
||||
assert message['To'] == 'happygrrl@example.org'
|
||||
email_context = template.TEMPLATE_TEST_CONTEXT[
|
||||
'mediagoblin/auth/fp_verification_email.txt']
|
||||
|
||||
@ -15,9 +15,8 @@
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
from mediagoblin.tests.tools import setup_fresh_app
|
||||
from mediagoblin import util
|
||||
from mediagoblin import mg_globals
|
||||
|
||||
from mediagoblin.tools import text
|
||||
|
||||
@setup_fresh_app
|
||||
def test_list_of_dicts_conversion(test_app):
|
||||
@ -28,23 +27,23 @@ def test_list_of_dicts_conversion(test_app):
|
||||
function performs the reverse operation when populating a form to edit tags.
|
||||
"""
|
||||
# Leading, trailing, and internal whitespace should be removed and slugified
|
||||
assert util.convert_to_tag_list_of_dicts('sleep , 6 AM, chainsaw! ') == [
|
||||
assert text.convert_to_tag_list_of_dicts('sleep , 6 AM, chainsaw! ') == [
|
||||
{'name': u'sleep', 'slug': u'sleep'},
|
||||
{'name': u'6 AM', 'slug': u'6-am'},
|
||||
{'name': u'chainsaw!', 'slug': u'chainsaw'}]
|
||||
|
||||
# If the user enters two identical tags, record only one of them
|
||||
assert util.convert_to_tag_list_of_dicts('echo,echo') == [{'name': u'echo',
|
||||
assert text.convert_to_tag_list_of_dicts('echo,echo') == [{'name': u'echo',
|
||||
'slug': u'echo'}]
|
||||
|
||||
# Make sure converting the list of dicts to a string works
|
||||
assert util.media_tags_as_string([{'name': u'yin', 'slug': u'yin'},
|
||||
assert text.media_tags_as_string([{'name': u'yin', 'slug': u'yin'},
|
||||
{'name': u'yang', 'slug': u'yang'}]) == \
|
||||
u'yin,yang'
|
||||
|
||||
# If the tag delimiter is a space then we expect different results
|
||||
mg_globals.app_config['tags_delimiter'] = u' '
|
||||
assert util.convert_to_tag_list_of_dicts('unicorn ceramic nazi') == [
|
||||
assert text.convert_to_tag_list_of_dicts('unicorn ceramic nazi') == [
|
||||
{'name': u'unicorn', 'slug': u'unicorn'},
|
||||
{'name': u'ceramic', 'slug': u'ceramic'},
|
||||
{'name': u'nazi', 'slug': u'nazi'}]
|
||||
|
||||
@ -16,10 +16,9 @@
|
||||
|
||||
import email
|
||||
|
||||
from mediagoblin import util
|
||||
from mediagoblin.tools import url, translate
|
||||
from mediagoblin.tools import common, url, translate, mail, text, testing
|
||||
|
||||
util._activate_testing()
|
||||
testing._activate_testing()
|
||||
|
||||
|
||||
def _import_component_testing_method(silly_string):
|
||||
@ -28,7 +27,7 @@ def _import_component_testing_method(silly_string):
|
||||
|
||||
|
||||
def test_import_component():
|
||||
imported_func = util.import_component(
|
||||
imported_func = common.import_component(
|
||||
'mediagoblin.tests.test_util:_import_component_testing_method')
|
||||
result = imported_func('hooobaladoobala')
|
||||
expected = u"'hooobaladoobala' is the silliest string I've ever seen"
|
||||
@ -36,10 +35,10 @@ def test_import_component():
|
||||
|
||||
|
||||
def test_send_email():
|
||||
util._clear_test_inboxes()
|
||||
mail._clear_test_inboxes()
|
||||
|
||||
# send the email
|
||||
util.send_email(
|
||||
mail.send_email(
|
||||
"sender@mediagoblin.example.org",
|
||||
["amanda@example.org", "akila@example.org"],
|
||||
"Testing is so much fun!",
|
||||
@ -48,8 +47,8 @@ def test_send_email():
|
||||
I hope you like unit tests JUST AS MUCH AS I DO!""")
|
||||
|
||||
# check the main inbox
|
||||
assert len(util.EMAIL_TEST_INBOX) == 1
|
||||
message = util.EMAIL_TEST_INBOX.pop()
|
||||
assert len(mail.EMAIL_TEST_INBOX) == 1
|
||||
message = mail.EMAIL_TEST_INBOX.pop()
|
||||
assert message['From'] == "sender@mediagoblin.example.org"
|
||||
assert message['To'] == "amanda@example.org, akila@example.org"
|
||||
assert message['Subject'] == "Testing is so much fun!"
|
||||
@ -58,8 +57,8 @@ I hope you like unit tests JUST AS MUCH AS I DO!""")
|
||||
I hope you like unit tests JUST AS MUCH AS I DO!"""
|
||||
|
||||
# Check everything that the FakeMhost.sendmail() method got is correct
|
||||
assert len(util.EMAIL_TEST_MBOX_INBOX) == 1
|
||||
mbox_dict = util.EMAIL_TEST_MBOX_INBOX.pop()
|
||||
assert len(mail.EMAIL_TEST_MBOX_INBOX) == 1
|
||||
mbox_dict = mail.EMAIL_TEST_MBOX_INBOX.pop()
|
||||
assert mbox_dict['from'] == "sender@mediagoblin.example.org"
|
||||
assert mbox_dict['to'] == ["amanda@example.org", "akila@example.org"]
|
||||
mbox_message = email.message_from_string(mbox_dict['message'])
|
||||
@ -107,7 +106,7 @@ def test_locale_to_lower_lower():
|
||||
|
||||
def test_html_cleaner():
|
||||
# Remove images
|
||||
result = util.clean_html(
|
||||
result = text.clean_html(
|
||||
'<p>Hi everybody! '
|
||||
'<img src="http://example.org/huge-purple-barney.png" /></p>\n'
|
||||
'<p>:)</p>')
|
||||
@ -118,7 +117,7 @@ def test_html_cleaner():
|
||||
'</div>')
|
||||
|
||||
# Remove evil javascript
|
||||
result = util.clean_html(
|
||||
result = text.clean_html(
|
||||
'<p><a href="javascript:nasty_surprise">innocent link!</a></p>')
|
||||
assert result == (
|
||||
'<p><a href="">innocent link!</a></p>')
|
||||
|
||||
@ -21,7 +21,7 @@ import os, shutil
|
||||
from paste.deploy import loadapp
|
||||
from webtest import TestApp
|
||||
|
||||
from mediagoblin import util
|
||||
from mediagoblin.tools import testing
|
||||
from mediagoblin.init.config import read_mediagoblin_config
|
||||
from mediagoblin.decorators import _make_safe
|
||||
from mediagoblin.db.open import setup_connection_and_db_from_config
|
||||
@ -59,7 +59,7 @@ def get_test_app(dump_old_app=True):
|
||||
suicide_if_bad_celery_environ()
|
||||
|
||||
# Make sure we've turned on testing
|
||||
util._activate_testing()
|
||||
testing._activate_testing()
|
||||
|
||||
# Leave this imported as it sets up celery.
|
||||
from mediagoblin.init.celery import from_tests
|
||||
@ -117,7 +117,7 @@ def setup_fresh_app(func):
|
||||
"""
|
||||
def wrapper(*args, **kwargs):
|
||||
test_app = get_test_app()
|
||||
util.clear_test_buckets()
|
||||
testing.clear_test_buckets()
|
||||
return func(test_app, *args, **kwargs)
|
||||
|
||||
return _make_safe(wrapper, func)
|
||||
|
||||
@ -18,10 +18,11 @@ from webob import exc
|
||||
|
||||
from mediagoblin import messages, mg_globals
|
||||
from mediagoblin.db.util import DESCENDING, ObjectId
|
||||
from mediagoblin.util import (
|
||||
Pagination, render_to_response, redirect, cleaned_markdown_conversion,
|
||||
render_404, delete_media_files)
|
||||
from mediagoblin.tools.text import cleaned_markdown_conversion
|
||||
from mediagoblin.tools.response import render_to_response, render_404, redirect
|
||||
from mediagoblin.tools.translate import pass_to_ugettext as _
|
||||
from mediagoblin.tools.pagination import Pagination
|
||||
from mediagoblin.tools.files import delete_media_files
|
||||
from mediagoblin.user_pages import forms as user_forms
|
||||
|
||||
from mediagoblin.decorators import (uses_pagination, get_user_media_entry,
|
||||
|
||||
@ -1,699 +0,0 @@
|
||||
# GNU MediaGoblin -- federated, autonomous media hosting
|
||||
# Copyright (C) 2011 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 division
|
||||
|
||||
from email.MIMEText import MIMEText
|
||||
#import gettext
|
||||
#import pkg_resources
|
||||
import smtplib
|
||||
import sys
|
||||
#import re
|
||||
#import translitcodec
|
||||
import urllib
|
||||
from math import ceil, floor
|
||||
import copy
|
||||
import wtforms
|
||||
|
||||
#from babel.localedata import exists
|
||||
#from babel.support import LazyProxy
|
||||
#import jinja2
|
||||
from webob import Response, exc
|
||||
from lxml.html.clean import Cleaner
|
||||
import markdown
|
||||
from wtforms.form import Form
|
||||
|
||||
from mediagoblin import mg_globals
|
||||
#from mediagoblin import messages
|
||||
from mediagoblin.db.util import ObjectId
|
||||
from mediagoblin.tools import url
|
||||
from mediagoblin.tools import common
|
||||
from mediagoblin.tools.template import TEMPLATE_TEST_CONTEXT, render_template
|
||||
|
||||
from itertools import izip, count
|
||||
|
||||
DISPLAY_IMAGE_FETCHING_ORDER = [u'medium', u'original', u'thumb']
|
||||
|
||||
def _activate_testing():
|
||||
"""
|
||||
Call this to activate testing in util.py
|
||||
"""
|
||||
|
||||
common.TESTS_ENABLED = True
|
||||
|
||||
def clear_test_buckets():
|
||||
"""
|
||||
We store some things for testing purposes that should be cleared
|
||||
when we want a "clean slate" of information for our next round of
|
||||
tests. Call this function to wipe all that stuff clean.
|
||||
|
||||
Also wipes out some other things we might redefine during testing,
|
||||
like the jinja envs.
|
||||
"""
|
||||
global SETUP_JINJA_ENVS
|
||||
SETUP_JINJA_ENVS = {}
|
||||
|
||||
global EMAIL_TEST_INBOX
|
||||
global EMAIL_TEST_MBOX_INBOX
|
||||
EMAIL_TEST_INBOX = []
|
||||
EMAIL_TEST_MBOX_INBOX = []
|
||||
|
||||
clear_test_template_context()
|
||||
|
||||
|
||||
# SETUP_JINJA_ENVS = {}
|
||||
|
||||
|
||||
# def get_jinja_env(template_loader, locale):
|
||||
# """
|
||||
# Set up the Jinja environment,
|
||||
|
||||
# (In the future we may have another system for providing theming;
|
||||
# for now this is good enough.)
|
||||
# """
|
||||
# setup_gettext(locale)
|
||||
|
||||
# # If we have a jinja environment set up with this locale, just
|
||||
# # return that one.
|
||||
# if SETUP_JINJA_ENVS.has_key(locale):
|
||||
# return SETUP_JINJA_ENVS[locale]
|
||||
|
||||
# template_env = jinja2.Environment(
|
||||
# loader=template_loader, autoescape=True,
|
||||
# extensions=['jinja2.ext.i18n', 'jinja2.ext.autoescape'])
|
||||
|
||||
# template_env.install_gettext_callables(
|
||||
# mg_globals.translations.ugettext,
|
||||
# mg_globals.translations.ungettext)
|
||||
|
||||
# # All templates will know how to ...
|
||||
# # ... fetch all waiting messages and remove them from the queue
|
||||
# # ... construct a grid of thumbnails or other media
|
||||
# template_env.globals['fetch_messages'] = messages.fetch_messages
|
||||
# template_env.globals['gridify_list'] = gridify_list
|
||||
# template_env.globals['gridify_cursor'] = gridify_cursor
|
||||
|
||||
# if exists(locale):
|
||||
# SETUP_JINJA_ENVS[locale] = template_env
|
||||
|
||||
# return template_env
|
||||
|
||||
|
||||
# # We'll store context information here when doing unit tests
|
||||
# TEMPLATE_TEST_CONTEXT = {}
|
||||
|
||||
|
||||
# def render_template(request, template_path, context):
|
||||
# """
|
||||
# Render a template with context.
|
||||
|
||||
# Always inserts the request into the context, so you don't have to.
|
||||
# Also stores the context if we're doing unit tests. Helpful!
|
||||
# """
|
||||
# template = request.template_env.get_template(
|
||||
# template_path)
|
||||
# context['request'] = request
|
||||
# rendered = template.render(context)
|
||||
|
||||
# if TESTS_ENABLED:
|
||||
# TEMPLATE_TEST_CONTEXT[template_path] = context
|
||||
|
||||
# return rendered
|
||||
|
||||
|
||||
def clear_test_template_context():
|
||||
global TEMPLATE_TEST_CONTEXT
|
||||
TEMPLATE_TEST_CONTEXT = {}
|
||||
|
||||
|
||||
def render_to_response(request, template, context, status=200):
|
||||
"""Much like Django's shortcut.render()"""
|
||||
return Response(
|
||||
render_template(request, template, context),
|
||||
status=status)
|
||||
|
||||
|
||||
def redirect(request, *args, **kwargs):
|
||||
"""Returns a HTTPFound(), takes a request and then urlgen params"""
|
||||
|
||||
querystring = None
|
||||
if kwargs.get('querystring'):
|
||||
querystring = kwargs.get('querystring')
|
||||
del kwargs['querystring']
|
||||
|
||||
return exc.HTTPFound(
|
||||
location=''.join([
|
||||
request.urlgen(*args, **kwargs),
|
||||
querystring if querystring else '']))
|
||||
|
||||
|
||||
def setup_user_in_request(request):
|
||||
"""
|
||||
Examine a request and tack on a request.user parameter if that's
|
||||
appropriate.
|
||||
"""
|
||||
if not request.session.has_key('user_id'):
|
||||
request.user = None
|
||||
return
|
||||
|
||||
user = None
|
||||
user = request.app.db.User.one(
|
||||
{'_id': ObjectId(request.session['user_id'])})
|
||||
|
||||
if not user:
|
||||
# Something's wrong... this user doesn't exist? Invalidate
|
||||
# this session.
|
||||
request.session.invalidate()
|
||||
|
||||
request.user = user
|
||||
|
||||
|
||||
def import_component(import_string):
|
||||
"""
|
||||
Import a module component defined by STRING. Probably a method,
|
||||
class, or global variable.
|
||||
|
||||
Args:
|
||||
- import_string: a string that defines what to import. Written
|
||||
in the format of "module1.module2:component"
|
||||
"""
|
||||
module_name, func_name = import_string.split(':', 1)
|
||||
__import__(module_name)
|
||||
module = sys.modules[module_name]
|
||||
func = getattr(module, func_name)
|
||||
return func
|
||||
|
||||
# _punct_re = re.compile(r'[\t !"#$%&\'()*\-/<=>?@\[\\\]^_`{|},.]+')
|
||||
|
||||
# def slugify(text, delim=u'-'):
|
||||
# """
|
||||
# Generates an ASCII-only slug. Taken from http://flask.pocoo.org/snippets/5/
|
||||
# """
|
||||
# result = []
|
||||
# for word in _punct_re.split(text.lower()):
|
||||
# word = word.encode('translit/long')
|
||||
# if word:
|
||||
# result.append(word)
|
||||
# return unicode(delim.join(result))
|
||||
|
||||
### ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
||||
### Special email test stuff begins HERE
|
||||
### ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
||||
|
||||
# We have two "test inboxes" here:
|
||||
#
|
||||
# EMAIL_TEST_INBOX:
|
||||
# ----------------
|
||||
# If you're writing test views, you'll probably want to check this.
|
||||
# It contains a list of MIMEText messages.
|
||||
#
|
||||
# EMAIL_TEST_MBOX_INBOX:
|
||||
# ----------------------
|
||||
# This collects the messages from the FakeMhost inbox. It's reslly
|
||||
# just here for testing the send_email method itself.
|
||||
#
|
||||
# Anyway this contains:
|
||||
# - from
|
||||
# - to: a list of email recipient addresses
|
||||
# - message: not just the body, but the whole message, including
|
||||
# headers, etc.
|
||||
#
|
||||
# ***IMPORTANT!***
|
||||
# ----------------
|
||||
# Before running tests that call functions which send email, you should
|
||||
# always call _clear_test_inboxes() to "wipe" the inboxes clean.
|
||||
|
||||
EMAIL_TEST_INBOX = []
|
||||
EMAIL_TEST_MBOX_INBOX = []
|
||||
|
||||
|
||||
class FakeMhost(object):
|
||||
"""
|
||||
Just a fake mail host so we can capture and test messages
|
||||
from send_email
|
||||
"""
|
||||
def login(self, *args, **kwargs):
|
||||
pass
|
||||
|
||||
def sendmail(self, from_addr, to_addrs, message):
|
||||
EMAIL_TEST_MBOX_INBOX.append(
|
||||
{'from': from_addr,
|
||||
'to': to_addrs,
|
||||
'message': message})
|
||||
|
||||
def _clear_test_inboxes():
|
||||
global EMAIL_TEST_INBOX
|
||||
global EMAIL_TEST_MBOX_INBOX
|
||||
EMAIL_TEST_INBOX = []
|
||||
EMAIL_TEST_MBOX_INBOX = []
|
||||
|
||||
### ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
||||
### </Special email test stuff>
|
||||
### ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
||||
|
||||
def send_email(from_addr, to_addrs, subject, message_body):
|
||||
"""
|
||||
Simple email sending wrapper, use this so we can capture messages
|
||||
for unit testing purposes.
|
||||
|
||||
Args:
|
||||
- from_addr: address you're sending the email from
|
||||
- to_addrs: list of recipient email addresses
|
||||
- subject: subject of the email
|
||||
- message_body: email body text
|
||||
"""
|
||||
if common.TESTS_ENABLED or mg_globals.app_config['email_debug_mode']:
|
||||
mhost = FakeMhost()
|
||||
elif not mg_globals.app_config['email_debug_mode']:
|
||||
mhost = smtplib.SMTP(
|
||||
mg_globals.app_config['email_smtp_host'],
|
||||
mg_globals.app_config['email_smtp_port'])
|
||||
|
||||
# SMTP.__init__ Issues SMTP.connect implicitly if host
|
||||
if not mg_globals.app_config['email_smtp_host']: # e.g. host = ''
|
||||
mhost.connect() # We SMTP.connect explicitly
|
||||
|
||||
if mg_globals.app_config['email_smtp_user'] \
|
||||
or mg_globals.app_config['email_smtp_pass']:
|
||||
mhost.login(
|
||||
mg_globals.app_config['email_smtp_user'],
|
||||
mg_globals.app_config['email_smtp_pass'])
|
||||
|
||||
message = MIMEText(message_body.encode('utf-8'), 'plain', 'utf-8')
|
||||
message['Subject'] = subject
|
||||
message['From'] = from_addr
|
||||
message['To'] = ', '.join(to_addrs)
|
||||
|
||||
if common.TESTS_ENABLED:
|
||||
EMAIL_TEST_INBOX.append(message)
|
||||
|
||||
if 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)
|
||||
|
||||
return mhost.sendmail(from_addr, to_addrs, message.as_string())
|
||||
|
||||
|
||||
# ###################
|
||||
# # Translation tools
|
||||
# ###################
|
||||
|
||||
|
||||
# TRANSLATIONS_PATH = pkg_resources.resource_filename(
|
||||
# 'mediagoblin', 'i18n')
|
||||
|
||||
|
||||
# def locale_to_lower_upper(locale):
|
||||
# """
|
||||
# Take a locale, regardless of style, and format it like "en-us"
|
||||
# """
|
||||
# if '-' in locale:
|
||||
# lang, country = locale.split('-', 1)
|
||||
# return '%s_%s' % (lang.lower(), country.upper())
|
||||
# elif '_' in locale:
|
||||
# lang, country = locale.split('_', 1)
|
||||
# return '%s_%s' % (lang.lower(), country.upper())
|
||||
# else:
|
||||
# return locale.lower()
|
||||
|
||||
|
||||
# def locale_to_lower_lower(locale):
|
||||
# """
|
||||
# Take a locale, regardless of style, and format it like "en_US"
|
||||
# """
|
||||
# if '_' in locale:
|
||||
# lang, country = locale.split('_', 1)
|
||||
# return '%s-%s' % (lang.lower(), country.lower())
|
||||
# else:
|
||||
# return locale.lower()
|
||||
|
||||
|
||||
# def get_locale_from_request(request):
|
||||
# """
|
||||
# Figure out what target language is most appropriate based on the
|
||||
# request
|
||||
# """
|
||||
# request_form = request.GET or request.POST
|
||||
|
||||
# if request_form.has_key('lang'):
|
||||
# return locale_to_lower_upper(request_form['lang'])
|
||||
|
||||
# accept_lang_matches = request.accept_language.best_matches()
|
||||
|
||||
# # Your routing can explicitly specify a target language
|
||||
# matchdict = request.matchdict or {}
|
||||
|
||||
# if matchdict.has_key('locale'):
|
||||
# target_lang = matchdict['locale']
|
||||
# elif request.session.has_key('target_lang'):
|
||||
# target_lang = request.session['target_lang']
|
||||
# # Pull the first acceptable language
|
||||
# elif accept_lang_matches:
|
||||
# target_lang = accept_lang_matches[0]
|
||||
# # Fall back to English
|
||||
# else:
|
||||
# target_lang = 'en'
|
||||
|
||||
# return locale_to_lower_upper(target_lang)
|
||||
|
||||
|
||||
# A super strict version of the lxml.html cleaner class
|
||||
HTML_CLEANER = Cleaner(
|
||||
scripts=True,
|
||||
javascript=True,
|
||||
comments=True,
|
||||
style=True,
|
||||
links=True,
|
||||
page_structure=True,
|
||||
processing_instructions=True,
|
||||
embedded=True,
|
||||
frames=True,
|
||||
forms=True,
|
||||
annoying_tags=True,
|
||||
allow_tags=[
|
||||
'div', 'b', 'i', 'em', 'strong', 'p', 'ul', 'ol', 'li', 'a', 'br'],
|
||||
remove_unknown_tags=False, # can't be used with allow_tags
|
||||
safe_attrs_only=True,
|
||||
add_nofollow=True, # for now
|
||||
host_whitelist=(),
|
||||
whitelist_tags=set([]))
|
||||
|
||||
|
||||
def clean_html(html):
|
||||
# clean_html barfs on an empty string
|
||||
if not html:
|
||||
return u''
|
||||
|
||||
return HTML_CLEANER.clean_html(html)
|
||||
|
||||
|
||||
def convert_to_tag_list_of_dicts(tag_string):
|
||||
"""
|
||||
Filter input from incoming string containing user tags,
|
||||
|
||||
Strips trailing, leading, and internal whitespace, and also converts
|
||||
the "tags" text into an array of tags
|
||||
"""
|
||||
taglist = []
|
||||
if tag_string:
|
||||
|
||||
# Strip out internal, trailing, and leading whitespace
|
||||
stripped_tag_string = u' '.join(tag_string.strip().split())
|
||||
|
||||
# Split the tag string into a list of tags
|
||||
for tag in stripped_tag_string.split(
|
||||
mg_globals.app_config['tags_delimiter']):
|
||||
|
||||
# Ignore empty or duplicate tags
|
||||
if tag.strip() and tag.strip() not in [t['name'] for t in taglist]:
|
||||
|
||||
taglist.append({'name': tag.strip(),
|
||||
'slug': url.slugify(tag.strip())})
|
||||
return taglist
|
||||
|
||||
|
||||
def media_tags_as_string(media_entry_tags):
|
||||
"""
|
||||
Generate a string from a media item's tags, stored as a list of dicts
|
||||
|
||||
This is the opposite of convert_to_tag_list_of_dicts
|
||||
"""
|
||||
media_tag_string = ''
|
||||
if media_entry_tags:
|
||||
media_tag_string = mg_globals.app_config['tags_delimiter'].join(
|
||||
[tag['name'] for tag in media_entry_tags])
|
||||
return media_tag_string
|
||||
|
||||
TOO_LONG_TAG_WARNING = \
|
||||
u'Tags must be shorter than %s characters. Tags that are too long: %s'
|
||||
|
||||
def tag_length_validator(form, field):
|
||||
"""
|
||||
Make sure tags do not exceed the maximum tag length.
|
||||
"""
|
||||
tags = convert_to_tag_list_of_dicts(field.data)
|
||||
too_long_tags = [
|
||||
tag['name'] for tag in tags
|
||||
if len(tag['name']) > mg_globals.app_config['tags_max_length']]
|
||||
|
||||
if too_long_tags:
|
||||
raise wtforms.ValidationError(
|
||||
TOO_LONG_TAG_WARNING % (mg_globals.app_config['tags_max_length'], \
|
||||
', '.join(too_long_tags)))
|
||||
|
||||
|
||||
MARKDOWN_INSTANCE = markdown.Markdown(safe_mode='escape')
|
||||
|
||||
def cleaned_markdown_conversion(text):
|
||||
"""
|
||||
Take a block of text, run it through MarkDown, and clean its HTML.
|
||||
"""
|
||||
# Markdown will do nothing with and clean_html can do nothing with
|
||||
# an empty string :)
|
||||
if not text:
|
||||
return u''
|
||||
|
||||
return clean_html(MARKDOWN_INSTANCE.convert(text))
|
||||
|
||||
|
||||
# SETUP_GETTEXTS = {}
|
||||
|
||||
# def setup_gettext(locale):
|
||||
# """
|
||||
# Setup the gettext instance based on this locale
|
||||
# """
|
||||
# # Later on when we have plugins we may want to enable the
|
||||
# # multi-translations system they have so we can handle plugin
|
||||
# # translations too
|
||||
|
||||
# # TODO: fallback nicely on translations from pt_PT to pt if not
|
||||
# # available, etc.
|
||||
# if SETUP_GETTEXTS.has_key(locale):
|
||||
# this_gettext = SETUP_GETTEXTS[locale]
|
||||
# else:
|
||||
# this_gettext = gettext.translation(
|
||||
# 'mediagoblin', TRANSLATIONS_PATH, [locale], fallback=True)
|
||||
# if exists(locale):
|
||||
# SETUP_GETTEXTS[locale] = this_gettext
|
||||
|
||||
# mg_globals.setup_globals(
|
||||
# translations=this_gettext)
|
||||
|
||||
|
||||
# # Force en to be setup before anything else so that
|
||||
# # mg_globals.translations is never None
|
||||
# setup_gettext('en')
|
||||
|
||||
|
||||
# def pass_to_ugettext(*args, **kwargs):
|
||||
# """
|
||||
# Pass a translation on to the appropriate ugettext method.
|
||||
|
||||
# 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.translations.ugettext(
|
||||
# *args, **kwargs)
|
||||
|
||||
|
||||
# def lazy_pass_to_ugettext(*args, **kwargs):
|
||||
# """
|
||||
# Lazily pass to ugettext.
|
||||
|
||||
# This is useful if you have to define a translation on a module
|
||||
# level but you need it to not translate until the time that it's
|
||||
# used as a string.
|
||||
# """
|
||||
# return LazyProxy(pass_to_ugettext, *args, **kwargs)
|
||||
|
||||
|
||||
# def pass_to_ngettext(*args, **kwargs):
|
||||
# """
|
||||
# Pass a translation on to the appropriate ngettext method.
|
||||
|
||||
# The reason we can't have a global ngettext method is because
|
||||
# mg_globals gets swapped out by the application per-request.
|
||||
# """
|
||||
# return mg_globals.translations.ngettext(
|
||||
# *args, **kwargs)
|
||||
|
||||
|
||||
# def lazy_pass_to_ngettext(*args, **kwargs):
|
||||
# """
|
||||
# Lazily pass to ngettext.
|
||||
|
||||
# This is useful if you have to define a translation on a module
|
||||
# level but you need it to not translate until the time that it's
|
||||
# used as a string.
|
||||
# """
|
||||
# return LazyProxy(pass_to_ngettext, *args, **kwargs)
|
||||
|
||||
|
||||
# def fake_ugettext_passthrough(string):
|
||||
# """
|
||||
# Fake a ugettext call for extraction's sake ;)
|
||||
|
||||
# In wtforms there's a separate way to define a method to translate
|
||||
# things... so we just need to mark up the text so that it can be
|
||||
# extracted, not so that it's actually run through gettext.
|
||||
# """
|
||||
# return string
|
||||
|
||||
|
||||
PAGINATION_DEFAULT_PER_PAGE = 30
|
||||
|
||||
class Pagination(object):
|
||||
"""
|
||||
Pagination class for mongodb queries.
|
||||
|
||||
Initialization through __init__(self, cursor, page=1, per_page=2),
|
||||
get actual data slice through __call__().
|
||||
"""
|
||||
|
||||
def __init__(self, page, cursor, per_page=PAGINATION_DEFAULT_PER_PAGE,
|
||||
jump_to_id=False):
|
||||
"""
|
||||
Initializes Pagination
|
||||
|
||||
Args:
|
||||
- page: requested page
|
||||
- per_page: number of objects per page
|
||||
- cursor: db cursor
|
||||
- jump_to_id: ObjectId, sets the page to the page containing the object
|
||||
with _id == jump_to_id.
|
||||
"""
|
||||
self.page = page
|
||||
self.per_page = per_page
|
||||
self.cursor = cursor
|
||||
self.total_count = self.cursor.count()
|
||||
self.active_id = None
|
||||
|
||||
if jump_to_id:
|
||||
cursor = copy.copy(self.cursor)
|
||||
|
||||
for (doc, increment) in izip(cursor, count(0)):
|
||||
if doc['_id'] == jump_to_id:
|
||||
self.page = 1 + int(floor(increment / self.per_page))
|
||||
|
||||
self.active_id = jump_to_id
|
||||
break
|
||||
|
||||
|
||||
def __call__(self):
|
||||
"""
|
||||
Returns slice of objects for the requested page
|
||||
"""
|
||||
return self.cursor.skip(
|
||||
(self.page - 1) * self.per_page).limit(self.per_page)
|
||||
|
||||
@property
|
||||
def pages(self):
|
||||
return int(ceil(self.total_count / float(self.per_page)))
|
||||
|
||||
@property
|
||||
def has_prev(self):
|
||||
return self.page > 1
|
||||
|
||||
@property
|
||||
def has_next(self):
|
||||
return self.page < self.pages
|
||||
|
||||
def iter_pages(self, left_edge=2, left_current=2,
|
||||
right_current=5, right_edge=2):
|
||||
last = 0
|
||||
for num in xrange(1, self.pages + 1):
|
||||
if num <= left_edge or \
|
||||
(num > self.page - left_current - 1 and \
|
||||
num < self.page + right_current) or \
|
||||
num > self.pages - right_edge:
|
||||
if last + 1 != num:
|
||||
yield None
|
||||
yield num
|
||||
last = num
|
||||
|
||||
def get_page_url_explicit(self, base_url, get_params, page_no):
|
||||
"""
|
||||
Get a page url by adding a page= parameter to the base url
|
||||
"""
|
||||
new_get_params = copy.copy(get_params or {})
|
||||
new_get_params['page'] = page_no
|
||||
return "%s?%s" % (
|
||||
base_url, urllib.urlencode(new_get_params))
|
||||
|
||||
def get_page_url(self, request, page_no):
|
||||
"""
|
||||
Get a new page url based of the request, and the new page number.
|
||||
|
||||
This is a nice wrapper around get_page_url_explicit()
|
||||
"""
|
||||
return self.get_page_url_explicit(
|
||||
request.path_info, request.GET, page_no)
|
||||
|
||||
|
||||
# def gridify_list(this_list, num_cols=5):
|
||||
# """
|
||||
# Generates a list of lists where each sub-list's length depends on
|
||||
# the number of columns in the list
|
||||
# """
|
||||
# grid = []
|
||||
|
||||
# # Figure out how many rows we should have
|
||||
# num_rows = int(ceil(float(len(this_list)) / num_cols))
|
||||
|
||||
# for row_num in range(num_rows):
|
||||
# slice_min = row_num * num_cols
|
||||
# slice_max = (row_num + 1) * num_cols
|
||||
|
||||
# row = this_list[slice_min:slice_max]
|
||||
|
||||
# grid.append(row)
|
||||
|
||||
# return grid
|
||||
|
||||
|
||||
# def gridify_cursor(this_cursor, num_cols=5):
|
||||
# """
|
||||
# Generates a list of lists where each sub-list's length depends on
|
||||
# the number of columns in the list
|
||||
# """
|
||||
# return gridify_list(list(this_cursor), num_cols)
|
||||
|
||||
|
||||
def render_404(request):
|
||||
"""
|
||||
Render a 404.
|
||||
"""
|
||||
return render_to_response(
|
||||
request, 'mediagoblin/404.html', {}, status=400)
|
||||
|
||||
def delete_media_files(media):
|
||||
"""
|
||||
Delete all files associated with a MediaEntry
|
||||
|
||||
Arguments:
|
||||
- media: A MediaEntry document
|
||||
"""
|
||||
for listpath in media['media_files'].itervalues():
|
||||
mg_globals.public_store.delete_file(
|
||||
listpath)
|
||||
|
||||
for attachment in media['attachment_files']:
|
||||
mg_globals.public_store.delete_file(
|
||||
attachment['filepath'])
|
||||
@ -15,7 +15,8 @@
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
from mediagoblin import mg_globals
|
||||
from mediagoblin.util import render_to_response, Pagination
|
||||
from mediagoblin.tools.pagination import Pagination
|
||||
from mediagoblin.tools.response import render_to_response
|
||||
from mediagoblin.db.util import DESCENDING
|
||||
from mediagoblin.decorators import uses_pagination
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user