mediagoblin/tests/tools.py

   1 # GNU MediaGoblin -- federated, autonomous media hosting
   2 # Copyright (C) 2011, 2012 MediaGoblin contributors.  See AUTHORS.
   3 #
   4 # This program is free software: you can redistribute it and/or modify
   5 # it under the terms of the GNU Affero General Public License as published by
   6 # the Free Software Foundation, either version 3 of the License, or
   7 # (at your option) any later version.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU Affero General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU Affero General Public License
  15 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16
  17
  18 import os
  19 import pkg_resources
  20 import shutil
  21
  22 from functools import wraps
  23
  24 from paste.deploy import loadapp
  25 from webtest import TestApp
  26
  27 from mediagoblin import mg_globals
  28 from mediagoblin.db.models import User, MediaEntry, Collection
  29 from mediagoblin.tools import testing
  30 from mediagoblin.init.config import read_mediagoblin_config
  31 from mediagoblin.db.open import setup_connection_and_db_from_config
  32 from mediagoblin.db.base import Session
  33 from mediagoblin.meddleware import BaseMeddleware
  34 from mediagoblin.auth.lib import bcrypt_gen_password_hash
  35 from mediagoblin.gmg_commands.dbupdate import run_dbupdate
  36 from mediagoblin.init.celery import setup_celery_app
  37
  38
  39 MEDIAGOBLIN_TEST_DB_NAME = u'__mediagoblin_tests__'
  40 TEST_SERVER_CONFIG = pkg_resources.resource_filename(
  41     'mediagoblin.tests', 'test_paste.ini')
  42 TEST_APP_CONFIG = pkg_resources.resource_filename(
  43     'mediagoblin.tests', 'test_mgoblin_app.ini')
  44 TEST_USER_DEV = pkg_resources.resource_filename(
  45     'mediagoblin.tests', 'test_user_dev')
  46
  47
  48 USER_DEV_DIRECTORIES_TO_SETUP = [
  49     'media/public', 'media/queue',
  50     'beaker/sessions/data', 'beaker/sessions/lock']
  51
  52 BAD_CELERY_MESSAGE = """\
  53 Sorry, you *absolutely* must run tests with the
  54 mediagoblin.init.celery.from_tests module.  Like so:
  55 $ CELERY_CONFIG_MODULE=mediagoblin.init.celery.from_tests ./bin/py.test"""
  56
  57
  58 class BadCeleryEnviron(Exception): pass
  59
  60
  61 class TestingMeddleware(BaseMeddleware):
  62     """
  63     Meddleware for the Unit tests
  64
  65     It might make sense to perform some tests on all
  66     requests/responses. Or prepare them in a special
  67     manner. For example all html responses could be tested
  68     for being valid html *after* being rendered.
  69
  70     This module is getting inserted at the front of the
  71     meddleware list, which means: requests are handed here
  72     first, responses last. So this wraps up the "normal"
  73     app.
  74
  75     If you need to add a test, either add it directly to
  76     the appropiate process_request or process_response, or
  77     create a new method and call it from process_*.
  78     """
  79
  80     def process_response(self, request, response):
  81         # All following tests should be for html only!
  82         if getattr(response, 'content_type', None) != "text/html":
  83             # Get out early
  84             return
  85
  86         # If the template contains a reference to
  87         # /mgoblin_static/ instead of using
  88         # /request.staticdirect(), error out here.
  89         # This could probably be implemented as a grep on
  90         # the shipped templates easier...
  91         if response.text.find("/mgoblin_static/") >= 0:
  92             raise AssertionError(
  93                 "Response HTML contains reference to /mgoblin_static/ "
  94                 "instead of staticdirect. Request was for: "
  95                 + request.full_path)
  96
  97         return
  98
  99
 100 def suicide_if_bad_celery_environ():
 101     if not os.environ.get('CELERY_CONFIG_MODULE') == \
 102             'mediagoblin.init.celery.from_tests':
 103         raise BadCeleryEnviron(BAD_CELERY_MESSAGE)
 104
 105
 106 def get_app(request, paste_config=None, mgoblin_config=None):
 107     """Create a MediaGoblin app for testing.
 108
 109     Args:
 110      - request: Not an http request, but a pytest fixture request.  We
 111        use this to make temporary directories that pytest
 112        automatically cleans up as needed.
 113      - paste_config: particular paste config used by this application.
 114      - mgoblin_config: particular mediagoblin config used by this
 115        application.
 116     """
 117     paste_config = paste_config or TEST_SERVER_CONFIG
 118     mgoblin_config = mgoblin_config or TEST_APP_CONFIG
 119
 120     # This is the directory we're copying the paste/mgoblin config stuff into
 121     run_dir = request.config._tmpdirhandler.mktemp(
 122         'mgoblin_app', numbered=True)
 123     user_dev_dir = run_dir.mkdir('test_user_dev').strpath
 124
 125     new_paste_config = run_dir.join('paste.ini').strpath
 126     new_mgoblin_config = run_dir.join('mediagoblin.ini').strpath
 127     shutil.copyfile(paste_config, new_paste_config)
 128     shutil.copyfile(mgoblin_config, new_mgoblin_config)
 129
 130     suicide_if_bad_celery_environ()
 131
 132     # Make sure we've turned on testing
 133     testing._activate_testing()
 134
 135     # Leave this imported as it sets up celery.
 136     from mediagoblin.init.celery import from_tests
 137
 138     Session.rollback()
 139     Session.remove()
 140
 141     # install user_dev directories
 142     for directory in USER_DEV_DIRECTORIES_TO_SETUP:
 143         full_dir = os.path.join(user_dev_dir, directory)
 144         os.makedirs(full_dir)
 145
 146     # Get app config
 147     global_config, validation_result = read_mediagoblin_config(new_mgoblin_config)
 148     app_config = global_config['mediagoblin']
 149
 150     # Run database setup/migrations
 151     run_dbupdate(app_config, global_config)
 152
 153     # setup app and return
 154     test_app = loadapp(
 155         'config:' + new_paste_config)
 156
 157     # Re-setup celery
 158     setup_celery_app(app_config, global_config)
 159
 160     # Insert the TestingMeddleware, which can do some
 161     # sanity checks on every request/response.
 162     # Doing it this way is probably not the cleanest way.
 163     # We'll fix it, when we have plugins!
 164     mg_globals.app.meddleware.insert(0, TestingMeddleware(mg_globals.app))
 165
 166     app = TestApp(test_app)
 167
 168     return app
 169
 170
 171 def install_fixtures_simple(db, fixtures):
 172     """
 173     Very simply install fixtures in the database
 174     """
 175     for collection_name, collection_fixtures in fixtures.iteritems():
 176         collection = db[collection_name]
 177         for fixture in collection_fixtures:
 178             collection.insert(fixture)
 179
 180
 181 def assert_db_meets_expected(db, expected):
 182     """
 183     Assert a database contains the things we expect it to.
 184
 185     Objects are found via 'id', so you should make sure your document
 186     has an id.
 187
 188     Args:
 189      - db: pymongo or mongokit database connection
 190      - expected: the data we expect.  Formatted like:
 191          {'collection_name': [
 192              {'id': 'foo',
 193               'some_field': 'some_value'},]}
 194     """
 195     for collection_name, collection_data in expected.iteritems():
 196         collection = db[collection_name]
 197         for expected_document in collection_data:
 198             document = collection.find_one({'id': expected_document['id']})
 199             assert document is not None  # make sure it exists
 200             assert document == expected_document  # make sure it matches
 201
 202
 203 def fixture_add_user(username=u'chris', password=u'toast',
 204                      active_user=True):
 205     # Reuse existing user or create a new one
 206     test_user = User.query.filter_by(username=username).first()
 207     if test_user is None:
 208         test_user = User()
 209     test_user.username = username
 210     test_user.email = username + u'@example.com'
 211     if password is not None:
 212         test_user.pw_hash = bcrypt_gen_password_hash(password)
 213     if active_user:
 214         test_user.email_verified = True
 215         test_user.status = u'active'
 216
 217     test_user.save()
 218
 219     # Reload
 220     test_user = User.query.filter_by(username=username).first()
 221
 222     # ... and detach from session:
 223     Session.expunge(test_user)
 224
 225     return test_user
 226
 227
 228 def fixture_media_entry(title=u"Some title", slug=None,
 229                         uploader=None, save=True, gen_slug=True):
 230     entry = MediaEntry()
 231     entry.title = title
 232     entry.slug = slug
 233     entry.uploader = uploader or fixture_add_user().id
 234     entry.media_type = u'image'
 235
 236     if gen_slug:
 237         entry.generate_slug()
 238     if save:
 239         entry.save()
 240
 241     return entry
 242
 243
 244 def fixture_add_collection(name=u"My first Collection", user=None):
 245     if user is None:
 246         user = fixture_add_user()
 247     coll = Collection.query.filter_by(creator=user.id, title=name).first()
 248     if coll is not None:
 249         return coll
 250     coll = Collection()
 251     coll.creator = user.id
 252     coll.title = name
 253     coll.generate_slug()
 254     coll.save()
 255
 256     # Reload
 257     Session.refresh(coll)
 258
 259     # ... and detach from session:
 260     Session.expunge(coll)
 261
 262     return coll