Run::
- ./bin/nosetests
+ CELERY_CONFIG_MODULE=mediagoblin.celery_setup.from_tests ./bin/nosetests
Running a shell
raise ImproperlyConfigured(
"One of direct_remote_path or direct_remote_paths must be provided")
- if asbool(os.environ.get('CELERY_ALWAYS_EAGER')):
- setup_celery_from_config(
- app_config, global_config,
- force_celery_always_eager=True)
- else:
- setup_celery_from_config(app_config, global_config)
+ if not asbool(app_config.get('celery_setup_elsewhere')):
+ if asbool(os.environ.get('CELERY_ALWAYS_EAGER')):
+ setup_celery_from_config(
+ app_config, global_config,
+ force_celery_always_eager=True)
+ else:
+ setup_celery_from_config(app_config, global_config)
mgoblin_app = MediaGoblinApp(
connection, db,
'Password',
[wtforms.validators.Required(),
wtforms.validators.Length(min=6, max=30),
- wtforms.validators.EqualTo('confirm_password')])
+ wtforms.validators.EqualTo(
+ 'confirm_password',
+ 'Passwords must match.')])
confirm_password = wtforms.PasswordField(
'Confirm password',
[wtforms.validators.Required()])
import bcrypt
+from mediagoblin.util import send_email, render_template
+from mediagoblin import globals as mgoblin_globals
+
def bcrypt_check_password(raw_pass, stored_hash, extra_salt=None):
"""
randplus_hashed_pass = bcrypt.hashpw(hashed_pass, rand_salt)
randplus_stored_hash == randplus_hashed_pass
+
+
+EMAIL_VERIFICATION_TEMPLATE = (
+ u"http://{host}{uri}?"
+ u"userid={userid}&token={verification_key}")
+
+def send_verification_email(user, request):
+ """
+ Send the verification email to users to activate their accounts.
+
+ Args:
+ - user: a user object
+ - request: the request
+ """
+ rendered_email = render_template(
+ request, 'mediagoblin/auth/verification_email.txt',
+ {'username': user['username'],
+ 'verification_url': EMAIL_VERIFICATION_TEMPLATE.format(
+ host=request.host,
+ uri=request.urlgen('mediagoblin.auth.verify_email'),
+ userid=unicode(user['_id']),
+ verification_key=user['verification_key'])})
+
+ # TODO: There is no error handling in place
+ send_email(
+ mgoblin_globals.email_sender_address,
+ [user['email']],
+ # TODO
+ # Due to the distributed nature of GNU MediaGoblin, we should
+ # find a way to send some additional information about the
+ # specific GNU MediaGoblin instance in the subject line. For
+ # example "GNU MediaGoblin @ Wandborg - [...]".
+ 'GNU MediaGoblin - Verify your email!',
+ rendered_email)
Route('mediagoblin.auth.register', '/register/',
controller='mediagoblin.auth.views:register'),
Route('mediagoblin.auth.register_success', '/register/success/',
- controller='mediagoblin.auth.views:register_success'),
+ template='mediagoblin/auth/register_success.html',
+ controller='mediagoblin.views:simple_template_render'),
Route('mediagoblin.auth.login', '/login/',
controller='mediagoblin.auth.views:login'),
Route('mediagoblin.auth.logout', '/logout/',
Route('mediagoblin.auth.verify_email', '/verify_email/',
controller='mediagoblin.auth.views:verify_email'),
Route('mediagoblin.auth.verify_email_notice', '/verification_required/',
- controller='mediagoblin.auth.views:verify_email_notice'),
+ template='mediagoblin/auth/verification_needed.html',
+ controller='mediagoblin.views:simple_template_render'),
Route('mediagoblin.auth.resend_verification', '/resend_verification/',
controller='mediagoblin.auth.views:resend_activation'),
Route('mediagoblin.auth.resend_verification_success',
'/resend_verification_success/',
- controller='mediagoblin.auth.views:resend_activation_success')]
+ template='mediagoblin/auth/resent_verification_email.html',
+ controller='mediagoblin.views:simple_template_render')]
import uuid
-from webob import Response, exc
+from webob import exc
+from mediagoblin.util import render_to_response, redirect
from mediagoblin.db.util import ObjectId
from mediagoblin.auth import lib as auth_lib
from mediagoblin.auth import forms as auth_forms
-from mediagoblin.util import send_email
-from mediagoblin import globals as mgoblin_globals
+from mediagoblin.auth.lib import send_verification_email
def register(request):
request.POST['password'])
entry.save(validate=True)
- email_template = request.template_env.get_template(
- 'mediagoblin/auth/verification_email.txt')
-
- # TODO: There is no error handling in place
- send_email(
- mgoblin_globals.email_sender_address,
- [entry['email']],
- # TODO
- # Due to the distributed nature of GNU MediaGoblin, we should
- # find a way to send some additional information about the
- # specific GNU MediaGoblin instance in the subject line. For
- # example "GNU MediaGoblin @ Wandborg - [...]".
- 'GNU MediaGoblin - Verify email',
- email_template.render(
- username=entry['username'],
- verification_url='http://{host}{uri}?userid={userid}&token={verification_key}'.format(
- host=request.host,
- uri=request.urlgen('mediagoblin.auth.verify_email'),
- userid=unicode(entry['_id']),
- verification_key=entry['verification_key'])))
-
- # Redirect to register_success
- return exc.HTTPFound(
- location=request.urlgen("mediagoblin.auth.register_success"))
-
- # render
- template = request.template_env.get_template(
- 'mediagoblin/auth/register.html')
- return Response(
- template.render(
- {'request': request,
- 'register_form': register_form}))
+ send_verification_email(entry, request)
+ return redirect(request, "mediagoblin.auth.register_success")
-def register_success(request):
- template = request.template_env.get_template(
- 'mediagoblin/auth/register_success.html')
- return Response(
- template.render(
- {'request': request}))
+ return render_to_response(
+ request,
+ 'mediagoblin/auth/register.html',
+ {'register_form': register_form})
def login(request):
if request.POST.get('next'):
return exc.HTTPFound(location=request.POST['next'])
else:
- return exc.HTTPFound(
- location=request.urlgen("index"))
+ return redirect(request, "index")
else:
# Prevent detecting who's on this system by testing login
auth_lib.fake_login_attempt()
login_failed = True
- # render
- template = request.template_env.get_template(
- 'mediagoblin/auth/login.html')
- return Response(
- template.render(
- {'request': request,
- 'login_form': login_form,
- 'next': request.GET.get('next') or request.POST.get('next'),
- 'login_failed': login_failed}))
+ return render_to_response(
+ request,
+ 'mediagoblin/auth/login.html',
+ {'login_form': login_form,
+ 'next': request.GET.get('next') or request.POST.get('next'),
+ 'login_failed': login_failed})
def logout(request):
# Maybe deleting the user_id parameter would be enough?
request.session.delete()
- return exc.HTTPFound(
- location=request.urlgen("index"))
+ return redirect(request, "index")
def verify_email(request):
user = request.db.User.find_one(
{'_id': ObjectId(unicode(request.GET['userid']))})
- verification_successful = bool
-
if user and user['verification_key'] == unicode(request.GET['token']):
user['status'] = u'active'
user['email_verified'] = True
else:
verification_successful = False
- template = request.template_env.get_template(
- 'mediagoblin/auth/verify_email.html')
- return Response(
- template.render(
- {'request': request,
- 'user': user,
- 'verification_successful': verification_successful}))
-
-def verify_email_notice(request):
- """
- Verify warning view.
-
- When the user tries to do some action that requires their account
- to be verified beforehand, this view is called upon!
- """
-
- template = request.template_env.get_template(
- 'mediagoblin/auth/verification_needed.html')
- return Response(
- template.render(
- {'request': request}))
+ return render_to_response(
+ request,
+ 'mediagoblin/auth/verify_email.html',
+ {'user': user,
+ 'verification_successful': verification_successful})
def resend_activation(request):
request.user['verification_key'] = unicode(uuid.uuid4())
request.user.save()
- # Copied shamelessly from the register view above.
-
- email_template = request.template_env.get_template(
- 'mediagoblin/auth/verification_email.txt')
-
- # TODO: There is no error handling in place
- send_email(
- mgoblin_globals.email_sender_address,
- [request.user['email']],
- # TODO
- # Due to the distributed nature of GNU MediaGoblin, we should
- # find a way to send some additional information about the
- # specific GNU MediaGoblin instance in the subject line. For
- # example "GNU MediaGoblin @ Wandborg - [...]".
- 'GNU MediaGoblin - Verify email',
- email_template.render(
- username=request.user['username'],
- verification_url='http://{host}{uri}?userid={userid}&token={verification_key}'.format(
- host=request.host,
- uri=request.urlgen('mediagoblin.auth.verify_email'),
- userid=unicode(request.user['_id']),
- verification_key=request.user['verification_key'])))
-
- return exc.HTTPFound(
- location=request.urlgen('mediagoblin.auth.resend_verification_success'))
-
-
-def resend_activation_success(request):
- template = request.template_env.get_template(
- 'mediagoblin/auth/resent_verification_email.html')
- return Response(
- template.render(
- {'request': request}))
+ send_verification_email(request.user, request)
+
+ return redirect(request, 'mediagoblin.auth.resend_verification_success')
if force_celery_always_eager:
celery_settings['CELERY_ALWAYS_EAGER'] = True
+ celery_settings['CELERY_EAGER_PROPAGATES_EXCEPTIONS'] = True
__import__(settings_module)
this_module = sys.modules[settings_module]
--- /dev/null
+# 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 os
+
+from mediagoblin.tests.tools import TEST_APP_CONFIG
+from mediagoblin import util
+from mediagoblin.celery_setup import setup_celery_from_config
+from mediagoblin.globals import setup_globals
+
+
+OUR_MODULENAME = 'mediagoblin.celery_setup.from_tests'
+
+
+def setup_self(setup_globals_func=setup_globals):
+ """
+ Set up celery for testing's sake, which just needs to set up
+ celery and celery only.
+ """
+ mgoblin_conf = util.read_config_file(TEST_APP_CONFIG)
+ mgoblin_section = mgoblin_conf['app:mediagoblin']
+
+ setup_celery_from_config(
+ mgoblin_section, mgoblin_conf,
+ settings_module=OUR_MODULENAME,
+ set_environ=False)
+
+
+if os.environ.get('CELERY_CONFIG_MODULE') == OUR_MODULENAME:
+ setup_self()
from bson.errors import InvalidId
from webob import exc
+from mediagoblin.util import redirect
from mediagoblin.db.util import ObjectId
def new_controller_func(request, *args, **kwargs):
if request.user and \
request.user.get('status') == u'needs_email_verification':
- return exc.HTTPFound(
- location = request.urlgen(
- 'mediagoblin.auth.verify_email_notice'))
+ return redirect(request,
+ 'mediagoblin.auth.verify_email_notice')
elif not request.user or request.user.get('status') != u'active':
return exc.HTTPFound(
location="%s?next=%s" % (
return controller(request, media=media, *args, **kwargs)
return _make_safe(wrapper, controller)
+
+def get_media_entry_by_id(controller):
+ """
+ Pass in a MediaEntry based off of a url component
+ """
+ def wrapper(request, *args, **kwargs):
+ try:
+ media = request.db.MediaEntry.find_one(
+ {'_id': ObjectId(request.matchdict['media']),
+ 'state': 'processed'})
+ except InvalidId:
+ return exc.HTTPNotFound()
+
+ # Still no media? Okay, 404.
+ if not media:
+ return exc.HTTPNotFound()
+
+ return controller(request, media=media, *args, **kwargs)
+
+ return _make_safe(wrapper, controller)
--- /dev/null
+# 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 wtforms
+
+
+class EditForm(wtforms.Form):
+ title = wtforms.TextField(
+ 'Title',
+ [wtforms.validators.Length(min=0, max=500)])
+ slug = wtforms.TextField(
+ 'Slug')
+ description = wtforms.TextAreaField('Description of this work')
--- /dev/null
+# 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/>.
+
+
+def may_edit_media(request, media):
+ """Check, if the request's user may edit the media details"""
+ if media['uploader'] == request.user['_id']:
+ return True
+ if request.user['is_admin']:
+ return True
+ return False
--- /dev/null
+# 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/>.
+
+
+from routes.route import Route
+
+edit_routes = [
+ # Media editing view handled in user_pages/routing.py
+]
--- /dev/null
+# 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/>.
+
+
+from webob import exc
+
+from mediagoblin.util import render_to_response, redirect
+from mediagoblin.edit import forms
+from mediagoblin.edit.lib import may_edit_media
+from mediagoblin.decorators import require_active_login, get_user_media_entry
+
+
+@get_user_media_entry
+@require_active_login
+def edit_media(request, media):
+ if not may_edit_media(request, media):
+ return exc.HTTPForbidden()
+
+ form = forms.EditForm(request.POST,
+ title = media['title'],
+ slug = media['slug'],
+ description = media['description'])
+
+ if request.method == 'POST' and form.validate():
+ # Make sure there isn't already a MediaEntry with such a slug
+ # and userid.
+ existing_user_slug_entries = request.db.MediaEntry.find(
+ {'slug': request.POST['slug'],
+ 'uploader': media['uploader'],
+ '_id': {'$ne': media['_id']}}).count()
+
+ if existing_user_slug_entries:
+ form.slug.errors.append(
+ u'An entry with that slug already exists for this user.')
+ else:
+ media['title'] = request.POST['title']
+ media['description'] = request.POST['description']
+ media['slug'] = request.POST['slug']
+ media.save()
+
+ return redirect(request, "mediagoblin.user_pages.media_home",
+ user=media.uploader()['username'], media=media['slug'])
+
+ return render_to_response(
+ request,
+ 'mediagoblin/edit/edit.html',
+ {'media': media,
+ 'form': form})
from mediagoblin.auth.routing import auth_routes
from mediagoblin.submit.routing import submit_routes
from mediagoblin.user_pages.routing import user_routes
+from mediagoblin.edit.routing import edit_routes
def get_mapper():
mapping = Mapper()
mapping.extend(auth_routes, '/auth')
mapping.extend(submit_routes, '/submit')
mapping.extend(user_routes, '/u')
+ mapping.extend(edit_routes, '/edit')
return mapping
Route('mediagoblin.submit.start', '/',
controller='mediagoblin.submit.views:submit_start'),
Route('mediagoblin.submit.success', '/success/',
- controller='mediagoblin.submit.views:submit_success'),
- ]
+ template='mediagoblin/submit/success.html',
+ controller='mediagoblin.views:simple_template_render')]
from os.path import splitext
from cgi import FieldStorage
-from webob import Response, exc
from werkzeug.utils import secure_filename
+from mediagoblin.util 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_initial
# queue it for processing
process_media_initial.delay(unicode(entry['_id']))
- # redirect
- return exc.HTTPFound(
- location=request.urlgen("mediagoblin.submit.success"))
+ return redirect(request, "mediagoblin.submit.success")
- # render
- template = request.template_env.get_template(
- 'mediagoblin/submit/start.html')
- return Response(
- template.render(
- {'request': request,
- 'submit_form': submit_form}))
+ return render_to_response(
+ request,
+ 'mediagoblin/submit/start.html',
+ {'submit_form': submit_form})
def submit_success(request):
- # render
- template = request.template_env.get_template(
- 'mediagoblin/submit/success.html')
- return Response(
- template.render(
- {'request': request}))
+ return render_to_response(
+ request, 'mediagoblin/submit/success.html', {})
<input type="submit" value="submit" class="button"/>
</div>
{% if next %}
- <input type="hidden" name="next" value="{{ next }}" class="button" />
+ <input type="hidden" name="next" value="{{ next }}" class="button"
+ style="display: none;"/>
{% endif %}
<p>Don't have an account yet? <a href="{{ request.urlgen('mediagoblin.auth.register') }}">Create one here!</a></p>
</div>
--- /dev/null
+{#
+# 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/>.
+#}
+{% extends "mediagoblin/base.html" %}
+
+{% import "/mediagoblin/utils/wtforms.html" as wtforms_util %}
+
+{% block mediagoblin_content %}
+ <h1>Edit details for {{ media.title }}</h1>
+
+ <form action="{{ request.urlgen('mediagoblin.edit.edit_media',
+ user= media.uploader().username,
+ media= media._id) }}"
+ method="POST" enctype="multipart/form-data">
+ <div class="submit_box form_box">
+ {{ wtforms_util.render_divs(form) }}
+ <div class="form_submit_buttons">
+ <input type="submit" value="submit" class="button" />
+ </div>
+ </div>
+ </form>
+ <img src="{{ request.app.public_store.file_url(
+ media['media_files']['thumb']) }}" />
+{% endblock %}
{# temporarily, an "image gallery" that isn't one really ;) #}
{% if media %}
- <img class="media_image" src="{{ request.app.public_store.file_url(
- media.media_files.main) }}" />
<h1>
{{media.title}}
</h1>
+ <img class="media_image" src="{{ request.app.public_store.file_url(
+ media.media_files.main) }}" />
<p>{{ media.description }}</p>
<p>Uploaded on
{{ "%4d-%02d-%02d"|format(media.created.year,
<a href="{{ request.urlgen('mediagoblin.user_pages.user_home',
user= media.uploader().username) }}">
{{- media.uploader().username }}</a></p>
+ <p><a href="{{ request.urlgen('mediagoblin.edit.edit_media',
+ user= media.uploader().username,
+ media= media._id) }}">Edit</a></p>
{% else %}
<p>Sorry, no such media found.<p/>
{% endif %}
<strong>{{ page }}</strong>
{% endif %}
{% else %}
- <span class=ellipsis>…</span>
+ <span class="ellipsis">…</span>
{% endif %}
{%- endfor %}
--- /dev/null
+[DEFAULT]
+debug = true
+
+[composite:main]
+use = egg:Paste#urlmap
+/ = mediagoblin
+/mgoblin_media/ = publicstore_serve
+/mgoblin_static/ = mediagoblin_static
+
+[app:mediagoblin]
+use = egg:mediagoblin#app
+filter-with = beaker
+queuestore_base_dir = %(here)s/test_user_dev/media/queue
+publicstore_base_dir = %(here)s/test_user_dev/media/public
+publicstore_base_url = /mgoblin_media/
+direct_remote_path = /mgoblin_static/
+email_sender_address = "notice@mediagoblin.example.org"
+email_debug_mode = true
+db_name = __mediagoblin_tests__
+# Celery shouldn't be set up by the paste app factory as it's set up
+# elsewhere
+celery_setup_elsewhere = true
+
+[app:publicstore_serve]
+use = egg:Paste#static
+document_root = %(here)s/user_dev/media/public
+
+[app:mediagoblin_static]
+use = egg:Paste#static
+document_root = %(here)s/mediagoblin/static/
+
+[filter:beaker]
+use = egg:Beaker#beaker_session
+cache_dir = %(here)s/test_user_dev/beaker
+beaker.session.key = mediagoblin
+# beaker.session.secret = somesupersecret
+beaker.session.data_dir = %(here)s/test_user_dev/beaker/sessions/data
+beaker.session.lock_dir = %(here)s/test_user_dev/beaker/sessions/lock
+
+[celery]
+celery_always_eager = true
+
+[server:main]
+use = egg:Paste#http
+host = 127.0.0.1
+port = 6543
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
+import urlparse
+
+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 globals as mgoblin_globals
+from mediagoblin import util
########################
pw, hashed_pw, '3><7R45417')
assert not auth_lib.bcrypt_check_password(
'notthepassword', hashed_pw, '3><7R45417')
+
+
+@setup_fresh_app
+def test_register_views(test_app):
+ """
+ Massive test function that all our registration-related views all work.
+ """
+ # Test doing a simple GET on the page
+ # -----------------------------------
+
+ test_app.get('/auth/register/')
+ # Make sure it rendered with the appropriate template
+ assert util.TEMPLATE_TEST_CONTEXT.has_key(
+ 'mediagoblin/auth/register.html')
+
+ # Try to register without providing anything, should error
+ # --------------------------------------------------------
+
+ util.clear_test_template_context()
+ test_app.post(
+ '/auth/register/', {})
+ context = util.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.']
+ assert form.confirm_password.errors == [u'This field is required.']
+ assert form.email.errors == [u'This field is required.']
+
+ # Try to register with fields that are known to be invalid
+ # --------------------------------------------------------
+
+ ## too short
+ util.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']
+ form = context['register_form']
+
+ assert form.username.errors == [
+ u'Field must be between 3 and 30 characters long.']
+ assert form.password.errors == [
+ u'Field must be between 6 and 30 characters long.']
+
+ ## bad form
+ util.clear_test_template_context()
+ test_app.post(
+ '/auth/register/', {
+ 'username': '@_@',
+ 'email': 'lollerskates'})
+ context = util.TEMPLATE_TEST_CONTEXT['mediagoblin/auth/register.html']
+ form = context['register_form']
+
+ assert form.username.errors == [
+ u'Invalid input.']
+ assert form.email.errors == [
+ u'Invalid email address.']
+
+ ## mismatching passwords
+ util.clear_test_template_context()
+ test_app.post(
+ '/auth/register/', {
+ 'password': 'herpderp',
+ 'confirm_password': 'derpherp'})
+ context = util.TEMPLATE_TEST_CONTEXT['mediagoblin/auth/register.html']
+ form = context['register_form']
+
+ assert form.password.errors == [
+ u'Passwords must match.']
+
+ ## At this point there should be no users in the database ;)
+ assert not mgoblin_globals.database.User.find().count()
+
+ # Successful register
+ # -------------------
+ util.clear_test_template_context()
+ response = test_app.post(
+ '/auth/register/', {
+ 'username': 'happygirl',
+ 'password': 'iamsohappy',
+ 'confirm_password': 'iamsohappy',
+ 'email': 'happygrrl@example.org'})
+ response.follow()
+
+ ## Did we redirect to the proper page? Use the right template?
+ assert_equal(
+ urlparse.urlsplit(response.location)[2],
+ '/auth/register/success/')
+ assert util.TEMPLATE_TEST_CONTEXT.has_key(
+ 'mediagoblin/auth/register_success.html')
+
+ ## Make sure user is in place
+ new_user = mgoblin_globals.database.User.find_one(
+ {'username': 'happygirl'})
+ assert new_user
+ assert new_user['status'] == u'needs_email_verification'
+ assert new_user['email_verified'] == False
+
+ ## Make sure we get email confirmation, and try verifying
+ assert len(util.EMAIL_TEST_INBOX) == 1
+ message = util.EMAIL_TEST_INBOX.pop()
+ assert message['To'] == 'happygrrl@example.org'
+ email_context = util.TEMPLATE_TEST_CONTEXT[
+ 'mediagoblin/auth/verification_email.txt']
+ assert email_context['verification_url'] in message.get_payload(decode=True)
+
+ path = urlparse.urlsplit(email_context['verification_url'])[2]
+ get_params = urlparse.urlsplit(email_context['verification_url'])[3]
+ assert path == u'/auth/verify_email/'
+ parsed_get_params = urlparse.parse_qs(get_params)
+
+ ### user should have these same parameters
+ assert parsed_get_params['userid'] == [
+ unicode(new_user['_id'])]
+ assert parsed_get_params['token'] == [
+ new_user['verification_key']]
+
+ ## Try verifying with bs verification key, shouldn't work
+ util.clear_test_template_context()
+ test_app.get(
+ "/auth/verify_email/?userid=%s&token=total_bs" % unicode(
+ new_user['_id']))
+ context = util.TEMPLATE_TEST_CONTEXT[
+ 'mediagoblin/auth/verify_email.html']
+ assert context['verification_successful'] == False
+ new_user = mgoblin_globals.database.User.find_one(
+ {'username': 'happygirl'})
+ assert new_user
+ assert new_user['status'] == u'needs_email_verification'
+ assert new_user['email_verified'] == False
+
+ ## Verify the email activation works
+ util.clear_test_template_context()
+ test_app.get("%s?%s" % (path, get_params))
+ context = util.TEMPLATE_TEST_CONTEXT[
+ 'mediagoblin/auth/verify_email.html']
+ assert context['verification_successful'] == True
+ new_user = mgoblin_globals.database.User.find_one(
+ {'username': 'happygirl'})
+ assert new_user
+ assert new_user['status'] == u'active'
+ assert new_user['email_verified'] == True
+
+ ## TODO: Try logging in
+
+ # Uniqueness checks
+ # -----------------
+ ## We shouldn't be able to register with that user twice
+ util.clear_test_template_context()
+ response = test_app.post(
+ '/auth/register/', {
+ 'username': 'happygirl',
+ 'password': 'iamsohappy2',
+ 'confirm_password': 'iamsohappy2',
+ 'email': 'happygrrl2@example.org'})
+
+ context = util.TEMPLATE_TEST_CONTEXT[
+ 'mediagoblin/auth/register.html']
+ form = context['register_form']
+ assert form.username.errors == [
+ u'Sorry, a user with that name already exists.']
+
+ ## TODO: Also check for double instances of an email address?
--- /dev/null
+# 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/>.
+
+from mediagoblin.tests.tools import get_test_app
+
+from mediagoblin import globals as mgoblin_globals
+
+
+def test_get_test_app_wipes_db():
+ """
+ Make sure we get a fresh database on every wipe :)
+ """
+ get_test_app()
+ assert mgoblin_globals.database.User.find().count() == 0
+
+ new_user = mgoblin_globals.database.User()
+ new_user['username'] = u'lolcat'
+ new_user['email'] = u'lol@cats.example.org'
+ new_user['pw_hash'] = u'pretend_this_is_a_hash'
+ new_user.save()
+ assert mgoblin_globals.database.User.find().count() == 1
+
+ get_test_app()
+
+ assert mgoblin_globals.database.User.find().count() == 0
--- /dev/null
+# 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 pkg_resources
+import os, shutil
+
+from paste.deploy import appconfig, loadapp
+from webtest import TestApp
+
+from mediagoblin import util
+from mediagoblin.decorators import _make_safe
+from mediagoblin.db.open import setup_connection_and_db_from_config
+
+
+MEDIAGOBLIN_TEST_DB_NAME = '__mediagoblinunittests__'
+TEST_APP_CONFIG = pkg_resources.resource_filename(
+ 'mediagoblin.tests', 'mgoblin_test_app.ini')
+TEST_USER_DEV = pkg_resources.resource_filename(
+ 'mediagoblin.tests', 'test_user_dev')
+MGOBLIN_APP = None
+
+USER_DEV_DIRECTORIES_TO_SETUP = [
+ 'media/public', 'media/queue',
+ 'beaker/sessions/data', 'beaker/sessions/lock']
+
+
+class BadCeleryEnviron(Exception): pass
+
+
+def get_test_app(dump_old_app=True):
+ if not os.environ.get('CELERY_CONFIG_MODULE') == \
+ 'mediagoblin.celery_setup.from_tests':
+ raise BadCeleryEnviron(
+ u"Sorry, you *absolutely* must run nosetests with the\n"
+ u"mediagoblin.celery_setup.from_tests module. Like so:\n"
+ u"$ CELERY_CONFIG_MODULE=mediagoblin.celery_setup.from_tests ./bin/nosetests")
+
+ # Just return the old app if that exists and it's okay to set up
+ # and return
+ if MGOBLIN_APP and not dump_old_app:
+ return MGOBLIN_APP
+
+ # Remove and reinstall user_dev directories
+ if os.path.exists(TEST_USER_DEV):
+ shutil.rmtree(TEST_USER_DEV)
+
+ for directory in USER_DEV_DIRECTORIES_TO_SETUP:
+ full_dir = os.path.join(TEST_USER_DEV, directory)
+ os.makedirs(full_dir)
+
+ # Get app config
+ config = appconfig(
+ 'config:' + os.path.basename(TEST_APP_CONFIG),
+ relative_to=os.path.dirname(TEST_APP_CONFIG),
+ name='mediagoblin')
+
+ # Wipe database
+ # @@: For now we're dropping collections, but we could also just
+ # collection.remove() ?
+ connection, db = setup_connection_and_db_from_config(
+ config.local_conf)
+
+ collections_to_wipe = [
+ collection
+ for collection in db.collection_names()
+ if not collection.startswith('system.')]
+
+ for collection in collections_to_wipe:
+ db.drop_collection(collection)
+
+ # Don't need these anymore...
+ del(connection)
+ del(db)
+
+ # TODO: Drop and recreate indexes
+
+ # setup app and return
+ test_app = loadapp(
+ 'config:' + TEST_APP_CONFIG)
+
+ return TestApp(test_app)
+
+
+def setup_fresh_app(func):
+ """
+ Decorator to setup a fresh test application for this function.
+
+ Cleans out test buckets and passes in a new, fresh test_app.
+ """
+ def wrapper(*args, **kwargs):
+ test_app = get_test_app()
+ util.clear_test_buckets()
+ return func(test_app, *args, **kwargs)
+
+ return _make_safe(wrapper, func)
Route('mediagoblin.user_pages.media_home', '/{user}/m/{media}/',
requirements=dict(m_id="[0-9a-fA-F]{24}"),
controller="mediagoblin.user_pages.views:media_home"),
+ Route('mediagoblin.edit.edit_media', "/{user}/m/{media}/edit/",
+ controller="mediagoblin.edit.views:edit_media"),
Route('mediagoblin.user_pages.atom_feed', '/{user}/atom/',
controller="mediagoblin.user_pages.views:atom_feed")]
# 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 webob import exc
from mediagoblin.db.util import DESCENDING
-from mediagoblin.util import Pagination
+from mediagoblin.util import Pagination, render_to_response
from mediagoblin.decorators import uses_pagination, get_user_media_entry
if media_entries == None:
return exc.HTTPNotFound()
- template = request.template_env.get_template(
- 'mediagoblin/user_pages/user.html')
-
- return Response(
- template.render(
- {'request': request,
- 'user': user,
- 'media_entries': media_entries,
- 'pagination': pagination}))
+ return render_to_response(
+ request,
+ 'mediagoblin/user_pages/user.html',
+ {'user': user,
+ 'media_entries': media_entries,
+ 'pagination': pagination})
@get_user_media_entry
def media_home(request, media):
"""'Homepage' of a MediaEntry()"""
- template = request.template_env.get_template(
- 'mediagoblin/user_pages/media.html')
- return Response(
- template.render(
- {'request': request,
- 'media': media}))
+ return render_to_response(
+ request,
+ 'mediagoblin/user_pages/media.html',
+ {'media': media})
+
ATOM_DEFAULT_NR_OF_UPDATED_ITEMS = 5
import gettext
import pkg_resources
import smtplib
+import os
import sys
import re
+import urllib
+from math import ceil
+import copy
+
+from babel.localedata import exists
import jinja2
-from mediagoblin.db.util import ObjectId
import translitcodec
+from paste.deploy.loadwsgi import NicerConfigParser
+from webob import Response, exc
from mediagoblin import globals as mgoblin_globals
+from mediagoblin.db.util import ObjectId
-import urllib
-from math import ceil
-import copy
TESTS_ENABLED = False
def _activate_testing():
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()
+
+
def get_jinja_loader(user_template_path=None):
"""
Set up the Jinja template loaders, possibly allowing for user
return jinja2.PackageLoader('mediagoblin', 'templates')
+SETUP_JINJA_ENVS = {}
+
+
def get_jinja_env(template_loader, locale):
"""
Set up the Jinja environment,
"""
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'])
mgoblin_globals.translations.gettext,
mgoblin_globals.translations.ngettext)
+ 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):
+ """Much like Django's shortcut.render()"""
+ return Response(render_template(request, template, context))
+
+
+def redirect(request, *args, **kwargs):
+ """Returns a HTTPFound(), takes a request and then urlgen params"""
+ return exc.HTTPFound(location=request.urlgen(*args, **kwargs))
+
+
def setup_user_in_request(request):
"""
Examine a request and tack on a request.user parameter if that's
return locale_to_lower_upper(target_lang)
+def read_config_file(conf_file):
+ """
+ Read a paste deploy style config file and process it.
+ """
+ if not os.path.exists(conf_file):
+ raise IOError(
+ "MEDIAGOBLIN_CONFIG not set or file does not exist")
+
+ parser = NicerConfigParser(conf_file)
+ parser.read(conf_file)
+ parser._defaults.setdefault(
+ 'here', os.path.dirname(os.path.abspath(conf_file)))
+ parser._defaults.setdefault(
+ '__file__', os.path.abspath(conf_file))
+
+ mgoblin_conf = dict(
+ [(section_name, dict(parser.items(section_name)))
+ for section_name in parser.sections()])
+
+ return mgoblin_conf
+
+
+SETUP_GETTEXTS = {}
+
def setup_gettext(locale):
"""
Setup the gettext instance based on this locale
# TODO: fallback nicely on translations from pt_PT to pt if not
# available, etc.
- this_gettext = gettext.translation(
- 'mediagoblin', TRANSLATIONS_PATH, [locale], fallback=True)
+ 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
mgoblin_globals.setup_globals(
translations=this_gettext)
# 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
+from mediagoblin.util import render_to_response
from mediagoblin.db.util import DESCENDING
def root_view(request):
media_entries = request.db.MediaEntry.find(
{u'state': u'processed'}).sort('created', DESCENDING)
- template = request.template_env.get_template(
- 'mediagoblin/root.html')
- return Response(
- template.render(
- {'request': request,
- 'media_entries': media_entries}))
+ return render_to_response(
+ request, 'mediagoblin/root.html',
+ {'media_entries': media_entries})
+
+
+def simple_template_render(request):
+ """
+ A view for absolutely simple template rendering.
+ Just make sure 'template' is in the matchdict!
+ """
+ template_name = request.matchdict['template']
+ return render_to_response(
+ request, template_name, {})
'Babel',
'translitcodec',
'argparse',
+ 'webtest',
],
test_suite='nose.collector',
projects / mediagoblin.git / commitdiff