heckyel/mediagoblin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote branch 'remotes/aaronw/bug444_fix_utils_py_redux'

Browse Source 
Conflicts:
	mediagoblin/util.py
This commit is contained in:
Christopher Allan Webber 2011-10-01 21:27:36 -05:00
commit b43b17fc26
37 changed files with 1018 additions and 864 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
mediagoblin/app.py
View File

@ -20,7 +20,9 @@ import urllib
import routes
from webob import Request, exc
from mediagoblin import routing, util, middleware
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,
@ -98,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]
@ -123,14 +125,14 @@ class MediaGoblinApp(object):
# Attach self as request.app
# Also attach a few utilities from request.app for convenience?
request.app = self
request.locale = util.get_locale_from_request(request)
request.locale = translate.get_locale_from_request(request)
request.template_env = util.get_jinja_env(
request.template_env = template.get_jinja_env(
self.template_loader, request.locale)
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:
@ -148,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

2
mediagoblin/auth/forms.py
View File

@ -17,7 +17,7 @@
import wtforms
import re
from mediagoblin.util import fake_ugettext_passthrough as _
from mediagoblin.tools.translate import fake_ugettext_passthrough as _
class RegistrationForm(wtforms.Form):

3
mediagoblin/auth/lib.py
View File

@ -19,7 +19,8 @@ import random
import bcrypt
from mediagoblin.util import send_email, render_template
from mediagoblin.tools.mail import send_email
from mediagoblin.tools.template import render_template
from mediagoblin import mg_globals

4
mediagoblin/auth/views.py
View File

@ -21,8 +21,8 @@ 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.util import pass_to_ugettext as _
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
from mediagoblin.auth import forms as auth_forms

2
mediagoblin/db/migrations.py
View File

@ -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

14
mediagoblin/db/models.py
View File

@ -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.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]
@ -242,7 +240,7 @@ class MediaEntry(Document):
pass
def generate_slug(self):
self['slug'] = util.slugify(self['title'])
self['slug'] = url.slugify(self['title'])
duplicate = mg_globals.database.media_entries.find_one(
{'slug': self['slug']})
@ -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):

2
mediagoblin/decorators.py
View File

@ -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

6
mediagoblin/edit/forms.py
View File

@ -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(

11
mediagoblin/edit/views.py
View File

@ -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

6
mediagoblin/gmg_commands/__init__.py
View File

@ -16,7 +16,7 @@
import argparse
from mediagoblin import util as mg_util
from mediagoblin.tools.common import import_component
SUBCOMMAND_MAP = {
@ -67,8 +67,8 @@ def main_cli():
else:
subparser = subparsers.add_parser(command_name)
setup_func = mg_util.import_component(command_struct['setup'])
exec_func = mg_util.import_component(command_struct['func'])
setup_func = import_component(command_struct['setup'])
exec_func = import_component(command_struct['func'])
setup_func(subparser)

3
mediagoblin/listings/views.py
View File

@ -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

2
mediagoblin/process_media/errors.py
View File

@ -14,7 +14,7 @@
# 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 mediagoblin.util import lazy_pass_to_ugettext as _
from mediagoblin.tools.translate import lazy_pass_to_ugettext as _
class BaseProcessingFail(Exception):
"""

4
mediagoblin/storage/__init__.py
View File

@ -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)

4
mediagoblin/submit/forms.py
View File

@ -17,8 +17,8 @@
import wtforms
from mediagoblin.util import tag_length_validator
from mediagoblin.util import fake_ugettext_passthrough as _
from mediagoblin.tools.text import tag_length_validator
from mediagoblin.tools.translate import fake_ugettext_passthrough as _
class SubmitStartForm(wtforms.Form):

7
mediagoblin/submit/views.py
View File

@ -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.util import pass_to_ugettext as _
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

106
mediagoblin/tests/test_auth.py
View File

@ -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 import util
from mediagoblin.tools import template, mail
########################
@ -76,16 +76,16 @@ def test_register_views(test_app):
test_app.get('/auth/register/')
# Make sure it rendered with the appropriate template
assert util.TEMPLATE_TEST_CONTEXT.has_key(
assert template.TEMPLATE_TEST_CONTEXT.has_key(
'mediagoblin/auth/register.html')
# Try to register without providing anything, should error
# --------------------------------------------------------
util.clear_test_template_context()
template.clear_test_template_context()
test_app.post(
'/auth/register/', {})
context = util.TEMPLATE_TEST_CONTEXT['mediagoblin/auth/register.html']
context = template.TEMPLATE_TEST_CONTEXT['mediagoblin/auth/register.html']
form = context['register_form']
assert form.username.errors == [u'This field is required.']
assert form.password.errors == [u'This field is required.']
@ -96,14 +96,14 @@ def test_register_views(test_app):
# --------------------------------------------------------
## too short
util.clear_test_template_context()
template.clear_test_template_context()
test_app.post(
'/auth/register/', {
'username': 'l',
'password': 'o',
'confirm_password': 'o',
'email': 'l'})
context = util.TEMPLATE_TEST_CONTEXT['mediagoblin/auth/register.html']
context = template.TEMPLATE_TEST_CONTEXT['mediagoblin/auth/register.html']
form = context['register_form']
assert form.username.errors == [
@ -112,12 +112,12 @@ def test_register_views(test_app):
u'Field must be between 6 and 30 characters long.']
## bad form
util.clear_test_template_context()
template.clear_test_template_context()
test_app.post(
'/auth/register/', {
'username': '@_@',
'email': 'lollerskates'})
context = util.TEMPLATE_TEST_CONTEXT['mediagoblin/auth/register.html']
context = template.TEMPLATE_TEST_CONTEXT['mediagoblin/auth/register.html']
form = context['register_form']
assert form.username.errors == [
@ -126,12 +126,12 @@ def test_register_views(test_app):
u'Invalid email address.']
## mismatching passwords
util.clear_test_template_context()
template.clear_test_template_context()
test_app.post(
'/auth/register/', {
'password': 'herpderp',
'confirm_password': 'derpherp'})
context = util.TEMPLATE_TEST_CONTEXT['mediagoblin/auth/register.html']
context = template.TEMPLATE_TEST_CONTEXT['mediagoblin/auth/register.html']
form = context['register_form']
assert form.password.errors == [
@ -142,7 +142,7 @@ def test_register_views(test_app):
# Successful register
# -------------------
util.clear_test_template_context()
template.clear_test_template_context()
response = test_app.post(
'/auth/register/', {
'username': 'happygirl',
@ -155,7 +155,7 @@ def test_register_views(test_app):
assert_equal(
urlparse.urlsplit(response.location)[2],
'/u/happygirl/')
assert util.TEMPLATE_TEST_CONTEXT.has_key(
assert template.TEMPLATE_TEST_CONTEXT.has_key(
'mediagoblin/user_pages/user.html')
## Make sure user is in place
@ -166,15 +166,15 @@ def test_register_views(test_app):
assert new_user['email_verified'] == False
## Make sure user is logged in
request = util.TEMPLATE_TEST_CONTEXT[
request = template.TEMPLATE_TEST_CONTEXT[
'mediagoblin/user_pages/user.html']['request']
assert request.session['user_id'] == unicode(new_user['_id'])
## Make sure we get email confirmation, and try verifying
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['To'] == 'happygrrl@example.org'
email_context = util.TEMPLATE_TEST_CONTEXT[
email_context = template.TEMPLATE_TEST_CONTEXT[
'mediagoblin/auth/verification_email.txt']
assert email_context['verification_url'] in message.get_payload(decode=True)
@ -190,12 +190,12 @@ def test_register_views(test_app):
new_user['verification_key']]
## Try verifying with bs verification key, shouldn't work
util.clear_test_template_context()
template.clear_test_template_context()
response = test_app.get(
"/auth/verify_email/?userid=%s&token=total_bs" % unicode(
new_user['_id']))
response.follow()
context = util.TEMPLATE_TEST_CONTEXT[
context = template.TEMPLATE_TEST_CONTEXT[
'mediagoblin/user_pages/user.html']
# assert context['verification_successful'] == True
# TODO: Would be good to test messages here when we can do so...
@ -206,10 +206,10 @@ def test_register_views(test_app):
assert new_user['email_verified'] == False
## Verify the email activation works
util.clear_test_template_context()
template.clear_test_template_context()
response = test_app.get("%s?%s" % (path, get_params))
response.follow()
context = util.TEMPLATE_TEST_CONTEXT[
context = template.TEMPLATE_TEST_CONTEXT[
'mediagoblin/user_pages/user.html']
# assert context['verification_successful'] == True
# TODO: Would be good to test messages here when we can do so...
@ -222,7 +222,7 @@ def test_register_views(test_app):
# Uniqueness checks
# -----------------
## We shouldn't be able to register with that user twice
util.clear_test_template_context()
template.clear_test_template_context()
response = test_app.post(
'/auth/register/', {
'username': 'happygirl',
@ -230,7 +230,7 @@ def test_register_views(test_app):
'confirm_password': 'iamsohappy2',
'email': 'happygrrl2@example.org'})
context = util.TEMPLATE_TEST_CONTEXT[
context = template.TEMPLATE_TEST_CONTEXT[
'mediagoblin/auth/register.html']
form = context['register_form']
assert form.username.errors == [
@ -240,7 +240,7 @@ def test_register_views(test_app):
### Oops, forgot the password
# -------------------
util.clear_test_template_context()
template.clear_test_template_context()
response = test_app.post(
'/auth/forgot_password/',
{'username': 'happygirl'})
@ -250,14 +250,14 @@ def test_register_views(test_app):
assert_equal(
urlparse.urlsplit(response.location)[2],
'/auth/forgot_password/email_sent/')
assert util.TEMPLATE_TEST_CONTEXT.has_key(
assert template.TEMPLATE_TEST_CONTEXT.has_key(
'mediagoblin/auth/fp_email_sent.html')
## Make sure link to change password is sent by email
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['To'] == 'happygrrl@example.org'
email_context = util.TEMPLATE_TEST_CONTEXT[
email_context = template.TEMPLATE_TEST_CONTEXT[
'mediagoblin/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)
@ -277,14 +277,14 @@ def test_register_views(test_app):
assert (new_user['fp_token_expire'] - datetime.datetime.now()).days == 9
## Try using a bs password-changing verification key, shouldn't work
util.clear_test_template_context()
template.clear_test_template_context()
response = test_app.get(
"/auth/forgot_password/verify/?userid=%s&token=total_bs" % unicode(
new_user['_id']), status=400)
assert response.status == '400 Bad Request'
## Try using an expired token to change password, shouldn't work
util.clear_test_template_context()
template.clear_test_template_context()
real_token_expiration = new_user['fp_token_expire']
new_user['fp_token_expire'] = datetime.datetime.now()
new_user.save()
@ -294,12 +294,12 @@ def test_register_views(test_app):
new_user.save()
## Verify step 1 of password-change works -- can see form to change password
util.clear_test_template_context()
template.clear_test_template_context()
response = test_app.get("%s?%s" % (path, get_params))
assert util.TEMPLATE_TEST_CONTEXT.has_key('mediagoblin/auth/change_fp.html')
assert template.TEMPLATE_TEST_CONTEXT.has_key('mediagoblin/auth/change_fp.html')
## Verify step 2.1 of password-change works -- report success to user
util.clear_test_template_context()
template.clear_test_template_context()
response = test_app.post(
'/auth/forgot_password/verify/', {
'userid': parsed_get_params['userid'],
@ -307,11 +307,11 @@ def test_register_views(test_app):
'confirm_password': 'iamveryveryhappy',
'token': parsed_get_params['token']})
response.follow()
assert util.TEMPLATE_TEST_CONTEXT.has_key(
assert template.TEMPLATE_TEST_CONTEXT.has_key(
'mediagoblin/auth/fp_changed_success.html')
## Verify step 2.2 of password-change works -- login w/ new password success
util.clear_test_template_context()
template.clear_test_template_context()
response = test_app.post(
'/auth/login/', {
'username': u'happygirl',
@ -322,7 +322,7 @@ def test_register_views(test_app):
assert_equal(
urlparse.urlsplit(response.location)[2],
'/')
assert util.TEMPLATE_TEST_CONTEXT.has_key(
assert template.TEMPLATE_TEST_CONTEXT.has_key(
'mediagoblin/root.html')
@ -341,61 +341,61 @@ def test_authentication_views(test_app):
# Get login
# ---------
test_app.get('/auth/login/')
assert util.TEMPLATE_TEST_CONTEXT.has_key(
assert template.TEMPLATE_TEST_CONTEXT.has_key(
'mediagoblin/auth/login.html')
# Failed login - blank form
# -------------------------
util.clear_test_template_context()
template.clear_test_template_context()
response = test_app.post('/auth/login/')
context = util.TEMPLATE_TEST_CONTEXT['mediagoblin/auth/login.html']
context = template.TEMPLATE_TEST_CONTEXT['mediagoblin/auth/login.html']
form = context['login_form']
assert form.username.errors == [u'This field is required.']
assert form.password.errors == [u'This field is required.']
# Failed login - blank user
# -------------------------
util.clear_test_template_context()
template.clear_test_template_context()
response = test_app.post(
'/auth/login/', {
'password': u'toast'})
context = util.TEMPLATE_TEST_CONTEXT['mediagoblin/auth/login.html']
context = template.TEMPLATE_TEST_CONTEXT['mediagoblin/auth/login.html']
form = context['login_form']
assert form.username.errors == [u'This field is required.']
# Failed login - blank password
# -----------------------------
util.clear_test_template_context()
template.clear_test_template_context()
response = test_app.post(
'/auth/login/', {
'username': u'chris'})
context = util.TEMPLATE_TEST_CONTEXT['mediagoblin/auth/login.html']
context = template.TEMPLATE_TEST_CONTEXT['mediagoblin/auth/login.html']
form = context['login_form']
assert form.password.errors == [u'This field is required.']
# Failed login - bad user
# -----------------------
util.clear_test_template_context()
template.clear_test_template_context()
response = test_app.post(
'/auth/login/', {
'username': u'steve',
'password': 'toast'})
context = util.TEMPLATE_TEST_CONTEXT['mediagoblin/auth/login.html']
context = template.TEMPLATE_TEST_CONTEXT['mediagoblin/auth/login.html']
assert context['login_failed']
# Failed login - bad password
# ---------------------------
util.clear_test_template_context()
template.clear_test_template_context()
response = test_app.post(
'/auth/login/', {
'username': u'chris',
'password': 'jam'})
context = util.TEMPLATE_TEST_CONTEXT['mediagoblin/auth/login.html']
context = template.TEMPLATE_TEST_CONTEXT['mediagoblin/auth/login.html']
assert context['login_failed']
# Successful login
# ----------------
util.clear_test_template_context()
template.clear_test_template_context()
response = test_app.post(
'/auth/login/', {
'username': u'chris',
@ -406,17 +406,17 @@ def test_authentication_views(test_app):
assert_equal(
urlparse.urlsplit(response.location)[2],
'/')
assert util.TEMPLATE_TEST_CONTEXT.has_key(
assert template.TEMPLATE_TEST_CONTEXT.has_key(
'mediagoblin/root.html')
# Make sure user is in the session
context = util.TEMPLATE_TEST_CONTEXT['mediagoblin/root.html']
context = template.TEMPLATE_TEST_CONTEXT['mediagoblin/root.html']
session = context['request'].session
assert session['user_id'] == unicode(test_user['_id'])
# Successful logout
# -----------------
util.clear_test_template_context()
template.clear_test_template_context()
response = test_app.get('/auth/logout/')
# Should be redirected to index page
@ -424,17 +424,17 @@ def test_authentication_views(test_app):
assert_equal(
urlparse.urlsplit(response.location)[2],
'/')
assert util.TEMPLATE_TEST_CONTEXT.has_key(
assert template.TEMPLATE_TEST_CONTEXT.has_key(
'mediagoblin/root.html')
# Make sure the user is not in the session
context = util.TEMPLATE_TEST_CONTEXT['mediagoblin/root.html']
context = template.TEMPLATE_TEST_CONTEXT['mediagoblin/root.html']
session = context['request'].session
assert session.has_key('user_id') == False
# User is redirected to custom URL if POST['next'] is set
# -------------------------------------------------------
util.clear_test_template_context()
template.clear_test_template_context()
response = test_app.post(
'/auth/login/', {
'username': u'chris',

4
mediagoblin/tests/test_messages.py
View File

@ -16,7 +16,7 @@
from mediagoblin.messages import fetch_messages, add_message
from mediagoblin.tests.tools import setup_fresh_app
from mediagoblin import util
from mediagoblin.tools import template
@setup_fresh_app
@ -28,7 +28,7 @@ def test_messages(test_app):
"""
# Aquire a request object
test_app.get('/')
context = util.TEMPLATE_TEST_CONTEXT['mediagoblin/root.html']
context = template.TEMPLATE_TEST_CONTEXT['mediagoblin/root.html']
request = context['request']
# The message queue should be empty

42
mediagoblin/tests/test_submission.py
View File

@ -22,7 +22,7 @@ from nose.tools import assert_equal, assert_true, assert_false
from mediagoblin.auth import lib as auth_lib
from mediagoblin.tests.tools import setup_fresh_app, get_test_app
from mediagoblin import mg_globals
from mediagoblin import util
from mediagoblin.tools import template, common
GOOD_JPG = pkg_resources.resource_filename(
'mediagoblin.tests', 'test_submission/good.jpg')
@ -63,20 +63,20 @@ class TestSubmission:
def test_missing_fields(self):
# Test blank form
# ---------------
util.clear_test_template_context()
template.clear_test_template_context()
response = self.test_app.post(
'/submit/', {})
context = util.TEMPLATE_TEST_CONTEXT['mediagoblin/submit/start.html']
context = template.TEMPLATE_TEST_CONTEXT['mediagoblin/submit/start.html']
form = context['submit_form']
assert form.file.errors == [u'You must provide a file.']
# Test blank file
# ---------------
util.clear_test_template_context()
template.clear_test_template_context()
response = self.test_app.post(
'/submit/', {
'title': 'test title'})
context = util.TEMPLATE_TEST_CONTEXT['mediagoblin/submit/start.html']
context = template.TEMPLATE_TEST_CONTEXT['mediagoblin/submit/start.html']
form = context['submit_form']
assert form.file.errors == [u'You must provide a file.']
@ -84,7 +84,7 @@ class TestSubmission:
def test_normal_uploads(self):
# Test JPG
# --------
util.clear_test_template_context()
template.clear_test_template_context()
response = self.test_app.post(
'/submit/', {
'title': 'Normal upload 1'
@ -96,12 +96,12 @@ class TestSubmission:
assert_equal(
urlparse.urlsplit(response.location)[2],
'/u/chris/')
assert util.TEMPLATE_TEST_CONTEXT.has_key(
assert template.TEMPLATE_TEST_CONTEXT.has_key(
'mediagoblin/user_pages/user.html')
# Test PNG
# --------
util.clear_test_template_context()
template.clear_test_template_context()
response = self.test_app.post(
'/submit/', {
'title': 'Normal upload 2'
@ -112,13 +112,13 @@ class TestSubmission:
assert_equal(
urlparse.urlsplit(response.location)[2],
'/u/chris/')
assert util.TEMPLATE_TEST_CONTEXT.has_key(
assert template.TEMPLATE_TEST_CONTEXT.has_key(
'mediagoblin/user_pages/user.html')
def test_tags(self):
# Good tag string
# --------
util.clear_test_template_context()
template.clear_test_template_context()
response = self.test_app.post(
'/submit/', {
'title': 'Balanced Goblin',
@ -128,7 +128,7 @@ class TestSubmission:
# New media entry with correct tags should be created
response.follow()
context = util.TEMPLATE_TEST_CONTEXT['mediagoblin/user_pages/user.html']
context = template.TEMPLATE_TEST_CONTEXT['mediagoblin/user_pages/user.html']
request = context['request']
media = request.db.MediaEntry.find({'title': 'Balanced Goblin'})[0]
assert_equal(media['tags'],
@ -137,7 +137,7 @@ class TestSubmission:
# Test tags that are too long
# ---------------
util.clear_test_template_context()
template.clear_test_template_context()
response = self.test_app.post(
'/submit/', {
'title': 'Balanced Goblin',
@ -146,14 +146,14 @@ class TestSubmission:
'file', GOOD_JPG)])
# Too long error should be raised
context = util.TEMPLATE_TEST_CONTEXT['mediagoblin/submit/start.html']
context = template.TEMPLATE_TEST_CONTEXT['mediagoblin/submit/start.html']
form = context['submit_form']
assert form.tags.errors == [
u'Tags must be shorter than 50 characters. Tags that are too long'\
': ffffffffffffffffffffffffffuuuuuuuuuuuuuuuuuuuuuuuuuu']
def test_delete(self):
util.clear_test_template_context()
template.clear_test_template_context()
response = self.test_app.post(
'/submit/', {
'title': 'Balanced Goblin',
@ -163,7 +163,7 @@ class TestSubmission:
# Post image
response.follow()
request = util.TEMPLATE_TEST_CONTEXT[
request = template.TEMPLATE_TEST_CONTEXT[
'mediagoblin/user_pages/user.html']['request']
media = request.db.MediaEntry.find({'title': 'Balanced Goblin'})[0]
@ -183,7 +183,7 @@ class TestSubmission:
response.follow()
request = util.TEMPLATE_TEST_CONTEXT[
request = template.TEMPLATE_TEST_CONTEXT[
'mediagoblin/user_pages/user.html']['request']
media = request.db.MediaEntry.find({'title': 'Balanced Goblin'})[0]
@ -202,7 +202,7 @@ class TestSubmission:
response.follow()
request = util.TEMPLATE_TEST_CONTEXT[
request = template.TEMPLATE_TEST_CONTEXT[
'mediagoblin/user_pages/user.html']['request']
# Does media entry still exist?
@ -213,14 +213,14 @@ class TestSubmission:
def test_malicious_uploads(self):
# Test non-suppoerted file with non-supported extension
# -----------------------------------------------------
util.clear_test_template_context()
template.clear_test_template_context()
response = self.test_app.post(
'/submit/', {
'title': 'Malicious Upload 1'
}, upload_files=[(
'file', EVIL_FILE)])
context = util.TEMPLATE_TEST_CONTEXT['mediagoblin/submit/start.html']
context = template.TEMPLATE_TEST_CONTEXT['mediagoblin/submit/start.html']
form = context['submit_form']
assert form.file.errors == ['The file doesn\'t seem to be an image!']
@ -230,7 +230,7 @@ class TestSubmission:
# Test non-supported file with .jpg extension
# -------------------------------------------
util.clear_test_template_context()
template.clear_test_template_context()
response = self.test_app.post(
'/submit/', {
'title': 'Malicious Upload 2'
@ -250,7 +250,7 @@ class TestSubmission:
# Test non-supported file with .png extension
# -------------------------------------------
util.clear_test_template_context()
template.clear_test_template_context()
response = self.test_app.post(
'/submit/', {
'title': 'Malicious Upload 3'

11
mediagoblin/tests/test_tags.py
View File

@ -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'}]

57
mediagoblin/tests/test_util.py
View File

@ -16,10 +16,9 @@
import email
from mediagoblin import util
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'])
@ -71,43 +70,43 @@ I hope you like unit tests JUST AS MUCH AS I DO!"""
I hope you like unit tests JUST AS MUCH AS I DO!"""
def test_slugify():
assert util.slugify('a walk in the park') == 'a-walk-in-the-park'
assert util.slugify('A Walk in the Park') == 'a-walk-in-the-park'
assert util.slugify('a walk in the park') == 'a-walk-in-the-park'
assert util.slugify('a walk in-the-park') == 'a-walk-in-the-park'
assert util.slugify('a w@lk in the park?') == 'a-w-lk-in-the-park'
assert util.slugify(u'a walk in the par\u0107') == 'a-walk-in-the-parc'
assert util.slugify(u'\u00E0\u0042\u00E7\u010F\u00EB\u0066') == 'abcdef'
assert url.slugify('a walk in the park') == 'a-walk-in-the-park'
assert url.slugify('A Walk in the Park') == 'a-walk-in-the-park'
assert url.slugify('a walk in the park') == 'a-walk-in-the-park'
assert url.slugify('a walk in-the-park') == 'a-walk-in-the-park'
assert url.slugify('a w@lk in the park?') == 'a-w-lk-in-the-park'
assert url.slugify(u'a walk in the par\u0107') == 'a-walk-in-the-parc'
assert url.slugify(u'\u00E0\u0042\u00E7\u010F\u00EB\u0066') == 'abcdef'
def test_locale_to_lower_upper():
"""
Test cc.i18n.util.locale_to_lower_upper()
"""
assert util.locale_to_lower_upper('en') == 'en'
assert util.locale_to_lower_upper('en_US') == 'en_US'
assert util.locale_to_lower_upper('en-us') == 'en_US'
assert translate.locale_to_lower_upper('en') == 'en'
assert translate.locale_to_lower_upper('en_US') == 'en_US'
assert translate.locale_to_lower_upper('en-us') == 'en_US'
# crazy renditions. Useful?
assert util.locale_to_lower_upper('en-US') == 'en_US'
assert util.locale_to_lower_upper('en_us') == 'en_US'
assert translate.locale_to_lower_upper('en-US') == 'en_US'
assert translate.locale_to_lower_upper('en_us') == 'en_US'
def test_locale_to_lower_lower():
"""
Test cc.i18n.util.locale_to_lower_lower()
"""
assert util.locale_to_lower_lower('en') == 'en'
assert util.locale_to_lower_lower('en_US') == 'en-us'
assert util.locale_to_lower_lower('en-us') == 'en-us'
assert translate.locale_to_lower_lower('en') == 'en'
assert translate.locale_to_lower_lower('en_US') == 'en-us'
assert translate.locale_to_lower_lower('en-us') == 'en-us'
# crazy renditions. Useful?
assert util.locale_to_lower_lower('en-US') == 'en-us'
assert util.locale_to_lower_lower('en_us') == 'en-us'
assert translate.locale_to_lower_lower('en-US') == 'en-us'
assert translate.locale_to_lower_lower('en_us') == 'en-us'
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>')

6
mediagoblin/tests/tools.py
View File

@ -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)

0
mediagoblin/tools/__init__.py Normal file
View File

37
mediagoblin/tools/common.py Normal file
View File

@ -0,0 +1,37 @@
# GNU MediaGoblin -- federated, autonomous media hosting
# Copyright (C) 2011 Free Software Foundation, Inc
#
# 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/>.
import sys
DISPLAY_IMAGE_FETCHING_ORDER = [u'medium', u'original', u'thumb']
global TESTS_ENABLED
TESTS_ENABLED = False
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

32
mediagoblin/tools/files.py Normal file
View File

@ -0,0 +1,32 @@
# 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 mediagoblin import mg_globals
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'])

120
mediagoblin/tools/mail.py Normal file
View File

@ -0,0 +1,120 @@
# 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/>.
import smtplib
from email.MIMEText import MIMEText
from mediagoblin import mg_globals
from mediagoblin.tools import common
### ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
### 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())

109
mediagoblin/tools/pagination.py Normal file
View File

@ -0,0 +1,109 @@
# 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/>.
import urllib
import copy
from math import ceil, floor
from itertools import izip, count
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)

37
mediagoblin/tools/request.py Normal file
View File

@ -0,0 +1,37 @@
# 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 mediagoblin.db.util import ObjectId
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

44
mediagoblin/tools/response.py Normal file
View File

@ -0,0 +1,44 @@
# 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 webob import Response, exc
from mediagoblin.tools.template import render_template
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 render_404(request):
"""
Render a 404.
"""
return render_to_response(
request, 'mediagoblin/404.html', {}, status=400)
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 '']))

116
mediagoblin/tools/template.py Normal file
View File

@ -0,0 +1,116 @@
# 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 math import ceil
import jinja2
from babel.localedata import exists
from babel.support import LazyProxy
from mediagoblin import mg_globals
from mediagoblin import messages
from mediagoblin.tools import common
from mediagoblin.tools.translate import setup_gettext
from mediagoblin.middleware.csrf import render_csrf_form_token
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
context['csrf_token'] = render_csrf_form_token(request)
rendered = template.render(context)
if common.TESTS_ENABLED:
TEMPLATE_TEST_CONTEXT[template_path] = context
return rendered
def clear_test_template_context():
global TEMPLATE_TEST_CONTEXT
TEMPLATE_TEST_CONTEXT = {}
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)

45
mediagoblin/tools/testing.py Normal file
View File

@ -0,0 +1,45 @@
# 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 mediagoblin.tools import common
from mediagoblin.tools.template import clear_test_template_context
from mediagoblin.tools.mail import EMAIL_TEST_INBOX, EMAIL_TEST_MBOX_INBOX
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()

117
mediagoblin/tools/text.py Normal file
View File

@ -0,0 +1,117 @@
# 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/>.
import wtforms
import markdown
from lxml.html.clean import Cleaner
from mediagoblin import mg_globals
from mediagoblin.tools import url
# 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))

167
mediagoblin/tools/translate.py Normal file
View File

@ -0,0 +1,167 @@
# 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/>.
import gettext
import pkg_resources
from babel.localedata import exists
from babel.support import LazyProxy
from mediagoblin import mg_globals
###################
# 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)
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

31
mediagoblin/tools/url.py Normal file
View File

@ -0,0 +1,31 @@
# GNU MediaGoblin -- federated, autonomous media hosting
# Copyright (C) 2011 Free Software Foundation, Inc
#
# 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/>.
import re
import translitcodec
_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))

2
mediagoblin/user_pages/forms.py
View File

@ -16,7 +16,7 @@
import wtforms
from mediagoblin.util import fake_ugettext_passthrough as _
from mediagoblin.tools.translate import fake_ugettext_passthrough as _
class MediaCommentForm(wtforms.Form):

9
mediagoblin/user_pages/views.py
View File

@ -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.util import pass_to_ugettext as _
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,

701
mediagoblin/util.py
View File

@ -1,701 +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 urllib
from math import ceil, floor
import copy
import wtforms
from babel.localedata import exists
from babel.support import LazyProxy
import jinja2
import translitcodec
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.middleware.csrf import render_csrf_form_token
from itertools import izip, count
DISPLAY_IMAGE_FETCHING_ORDER = [u'medium', u'original', u'thumb']
TESTS_ENABLED = False
def _activate_testing():
"""
Call this to activate testing in util.py
"""
global TESTS_ENABLED
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
context['csrf_token'] = render_csrf_form_token(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 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 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': 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'])

3
mediagoblin/views.py
View File

@ -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