mediagoblin/tests/test_pluginapi.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 import os
  18 import json
  19 import sys
  20
  21 from configobj import ConfigObj
  22 import pytest
  23 import pkg_resources
  24 from validate import VdtTypeError
  25
  26 from mediagoblin import mg_globals
  27 from mediagoblin.init.plugins import setup_plugins
  28 from mediagoblin.init.config import read_mediagoblin_config
  29 from mediagoblin.gmg_commands.assetlink import link_plugin_assets
  30 from mediagoblin.tools import pluginapi
  31 from mediagoblin.tests.tools import get_app
  32 from mediagoblin.tools.common import CollectingPrinter
  33
  34
  35 def with_cleanup(*modules_to_delete):
  36     def _with_cleanup(fun):
  37         """Wrapper that saves and restores mg_globals"""
  38         def _with_cleanup_inner(*args, **kwargs):
  39             old_app_config = mg_globals.app_config
  40             old_global_config = mg_globals.global_config
  41             # Need to delete icky modules before and after so as to make
  42             # sure things work correctly.
  43             for module in modules_to_delete:
  44                 try:
  45                     del sys.modules[module]
  46                 except KeyError:
  47                     pass
  48             # The plugin cache gets populated as a side-effect of
  49             # importing, so it's best to clear it before and after a test.
  50             pman = pluginapi.PluginManager()
  51             pman.clear()
  52             try:
  53                 return fun(*args, **kwargs)
  54             finally:
  55                 mg_globals.app_config = old_app_config
  56                 mg_globals.global_config = old_global_config
  57                 # Need to delete icky modules before and after so as to make
  58                 # sure things work correctly.
  59                 for module in modules_to_delete:
  60                     try:
  61                         del sys.modules[module]
  62                     except KeyError:
  63                         pass
  64                 pman.clear()
  65
  66         _with_cleanup_inner.__name__ = fun.__name__
  67         return _with_cleanup_inner
  68     return _with_cleanup
  69
  70
  71 def build_config(sections):
  72     """Builds a ConfigObj object with specified data
  73
  74     :arg sections: list of ``(section_name, section_data,
  75         subsection_list)`` tuples where section_data is a dict and
  76         subsection_list is a list of ``(section_name, section_data,
  77         subsection_list)``, ...
  78
  79     For example:
  80
  81     >>> build_config([
  82     ...    ('mediagoblin', {'key1': 'val1'}, []),
  83     ...    ('section2', {}, [
  84     ...        ('subsection1', {}, [])
  85     ...        ])
  86     ...    ])
  87     """
  88     cfg = ConfigObj()
  89     cfg.filename = 'foo'
  90     def _iter_section(cfg, section_list):
  91         for section_name, data, subsection_list in section_list:
  92             cfg[section_name] = data
  93             _iter_section(cfg[section_name], subsection_list)
  94
  95     _iter_section(cfg, sections)
  96     return cfg
  97
  98
  99 @with_cleanup()
 100 def test_no_plugins():
 101     """Run setup_plugins with no plugins in config"""
 102     cfg = build_config([('mediagoblin', {}, [])])
 103     mg_globals.app_config = cfg['mediagoblin']
 104     mg_globals.global_config = cfg
 105
 106     pman = pluginapi.PluginManager()
 107     setup_plugins()
 108
 109     # Make sure we didn't load anything.
 110     assert len(pman.plugins) == 0
 111
 112
 113 @with_cleanup('mediagoblin.plugins.sampleplugin')
 114 def test_one_plugin():
 115     """Run setup_plugins with a single working plugin"""
 116     cfg = build_config([
 117             ('mediagoblin', {}, []),
 118             ('plugins', {}, [
 119                     ('mediagoblin.plugins.sampleplugin', {}, [])
 120                     ])
 121             ])
 122
 123     mg_globals.app_config = cfg['mediagoblin']
 124     mg_globals.global_config = cfg
 125
 126     pman = pluginapi.PluginManager()
 127     setup_plugins()
 128
 129     # Make sure we only found one plugin
 130     assert len(pman.plugins) == 1
 131     # Make sure the plugin is the one we think it is.
 132     assert pman.plugins[0] == 'mediagoblin.plugins.sampleplugin'
 133     # Make sure there was one hook registered
 134     assert len(pman.hooks) == 1
 135     # Make sure _setup_plugin_called was called once
 136     import mediagoblin.plugins.sampleplugin
 137     assert mediagoblin.plugins.sampleplugin._setup_plugin_called == 1
 138
 139
 140 @with_cleanup('mediagoblin.plugins.sampleplugin')
 141 def test_same_plugin_twice():
 142     """Run setup_plugins with a single working plugin twice"""
 143     cfg = build_config([
 144             ('mediagoblin', {}, []),
 145             ('plugins', {}, [
 146                     ('mediagoblin.plugins.sampleplugin', {}, []),
 147                     ('mediagoblin.plugins.sampleplugin', {}, []),
 148                     ])
 149             ])
 150
 151     mg_globals.app_config = cfg['mediagoblin']
 152     mg_globals.global_config = cfg
 153
 154     pman = pluginapi.PluginManager()
 155     setup_plugins()
 156
 157     # Make sure we only found one plugin
 158     assert len(pman.plugins) == 1
 159     # Make sure the plugin is the one we think it is.
 160     assert pman.plugins[0] == 'mediagoblin.plugins.sampleplugin'
 161     # Make sure there was one hook registered
 162     assert len(pman.hooks) == 1
 163     # Make sure _setup_plugin_called was called once
 164     import mediagoblin.plugins.sampleplugin
 165     assert mediagoblin.plugins.sampleplugin._setup_plugin_called == 1
 166
 167
 168 @with_cleanup()
 169 def test_disabled_plugin():
 170     """Run setup_plugins with a single working plugin twice"""
 171     cfg = build_config([
 172             ('mediagoblin', {}, []),
 173             ('plugins', {}, [
 174                     ('-mediagoblin.plugins.sampleplugin', {}, []),
 175                     ])
 176             ])
 177
 178     mg_globals.app_config = cfg['mediagoblin']
 179     mg_globals.global_config = cfg
 180
 181     pman = pluginapi.PluginManager()
 182     setup_plugins()
 183
 184     # Make sure we didn't load the plugin
 185     assert len(pman.plugins) == 0
 186
 187
 188 CONFIG_ALL_CALLABLES = [
 189         ('mediagoblin', {}, []),
 190         ('plugins', {}, [
 191                 ('mediagoblin.tests.testplugins.callables1', {}, []),
 192                 ('mediagoblin.tests.testplugins.callables2', {}, []),
 193                 ('mediagoblin.tests.testplugins.callables3', {}, []),
 194             ])
 195     ]
 196
 197
 198 @with_cleanup()
 199 def test_hook_handle():
 200     """
 201     Test the hook_handle method
 202     """
 203     cfg = build_config(CONFIG_ALL_CALLABLES)
 204
 205     mg_globals.app_config = cfg['mediagoblin']
 206     mg_globals.global_config = cfg
 207
 208     setup_plugins()
 209
 210     # Just one hook provided
 211     call_log = []
 212     assert pluginapi.hook_handle(
 213         "just_one", call_log) == "Called just once"
 214     assert call_log == ["expect this one call"]
 215
 216     # Nothing provided and unhandled not okay
 217     call_log = []
 218     pluginapi.hook_handle(
 219         "nothing_handling", call_log) == None
 220     assert call_log == []
 221
 222     # Nothing provided and unhandled okay
 223     call_log = []
 224     assert pluginapi.hook_handle(
 225         "nothing_handling", call_log, unhandled_okay=True) is None
 226     assert call_log == []
 227
 228     # Multiple provided, go with the first!
 229     call_log = []
 230     assert pluginapi.hook_handle(
 231         "multi_handle", call_log) == "the first returns"
 232     assert call_log == ["Hi, I'm the first"]
 233
 234     # Multiple provided, one has CantHandleIt
 235     call_log = []
 236     assert pluginapi.hook_handle(
 237         "multi_handle_with_canthandle",
 238         call_log) == "the second returns"
 239     assert call_log == ["Hi, I'm the second"]
 240
 241
 242 @with_cleanup()
 243 def test_hook_runall():
 244     """
 245     Test the hook_runall method
 246     """
 247     cfg = build_config(CONFIG_ALL_CALLABLES)
 248
 249     mg_globals.app_config = cfg['mediagoblin']
 250     mg_globals.global_config = cfg
 251
 252     setup_plugins()
 253
 254     # Just one hook, check results
 255     call_log = []
 256     assert pluginapi.hook_runall(
 257         "just_one", call_log) == ["Called just once"]
 258     assert call_log == ["expect this one call"]
 259
 260     # None provided, check results
 261     call_log = []
 262     assert pluginapi.hook_runall(
 263         "nothing_handling", call_log) == []
 264     assert call_log == []
 265
 266     # Multiple provided, check results
 267     call_log = []
 268     assert pluginapi.hook_runall(
 269         "multi_handle", call_log) == [
 270             "the first returns",
 271             "the second returns",
 272             "the third returns",
 273         ]
 274     assert call_log == [
 275         "Hi, I'm the first",
 276         "Hi, I'm the second",
 277         "Hi, I'm the third"]
 278
 279     # Multiple provided, one has CantHandleIt, check results
 280     call_log = []
 281     assert pluginapi.hook_runall(
 282         "multi_handle_with_canthandle", call_log) == [
 283             "the second returns",
 284             "the third returns",
 285         ]
 286     assert call_log == [
 287         "Hi, I'm the second",
 288         "Hi, I'm the third"]
 289
 290
 291 @with_cleanup()
 292 def test_hook_transform():
 293     """
 294     Test the hook_transform method
 295     """
 296     cfg = build_config(CONFIG_ALL_CALLABLES)
 297
 298     mg_globals.app_config = cfg['mediagoblin']
 299     mg_globals.global_config = cfg
 300
 301     setup_plugins()
 302
 303     assert pluginapi.hook_transform(
 304         "expand_tuple", (-1, 0)) == (-1, 0, 1, 2, 3)
 305
 306
 307 def test_plugin_config():
 308     """
 309     Make sure plugins can set up their own config
 310     """
 311     config, validation_result = read_mediagoblin_config(
 312         pkg_resources.resource_filename(
 313             'mediagoblin.tests', 'appconfig_plugin_specs.ini'))
 314
 315     pluginspec_section = config['plugins'][
 316         'mediagoblin.tests.testplugins.pluginspec']
 317     assert pluginspec_section['some_string'] == 'not blork'
 318     assert pluginspec_section['dont_change_me'] == 'still the default'
 319
 320     # Make sure validation works... this should be an error
 321     assert isinstance(
 322         validation_result[
 323             'plugins'][
 324                 'mediagoblin.tests.testplugins.pluginspec'][
 325                     'some_int'],
 326         VdtTypeError)
 327
 328     # the callables thing shouldn't really have anything though.
 329     assert len(config['plugins'][
 330         'mediagoblin.tests.testplugins.callables1']) == 0
 331
 332
 333 @pytest.fixture()
 334 def context_modified_app(request):
 335     """
 336     Get a MediaGoblin app fixture using appconfig_context_modified.ini
 337     """
 338     return get_app(
 339         request,
 340         mgoblin_config=pkg_resources.resource_filename(
 341             'mediagoblin.tests', 'appconfig_context_modified.ini'))
 342
 343
 344 def test_modify_context(context_modified_app):
 345     """
 346     Test that we can modify both the view/template specific and
 347     global contexts for templates.
 348     """
 349     # Specific thing passed into a page
 350     result = context_modified_app.get("/modify_context/specific/")
 351     assert result.body.strip() == b"""Specific page!
 352
 353 specific thing: in yer specificpage
 354 global thing: globally appended!
 355 something: orother
 356 doubleme: happyhappy"""
 357
 358     # General test, should have global context variable only
 359     result = context_modified_app.get("/modify_context/")
 360     assert result.body.strip() == b"""General page!
 361
 362 global thing: globally appended!
 363 lol: cats
 364 doubleme: joyjoy"""
 365
 366
 367 @pytest.fixture()
 368 def static_plugin_app(request):
 369     """
 370     Get a MediaGoblin app fixture using appconfig_static_plugin.ini
 371     """
 372     return get_app(
 373         request,
 374         mgoblin_config=pkg_resources.resource_filename(
 375             'mediagoblin.tests', 'appconfig_static_plugin.ini'))
 376
 377
 378 def test_plugin_assetlink(static_plugin_app):
 379     """
 380     Test that the assetlink command works correctly
 381     """
 382     linked_assets_dir = mg_globals.app_config['plugin_linked_assets_dir']
 383     plugin_link_dir = os.path.join(
 384         linked_assets_dir.rstrip(os.path.sep),
 385         'staticstuff')
 386
 387     plugin_statics = pluginapi.hook_runall("static_setup")
 388     assert len(plugin_statics) == 1
 389     plugin_static = plugin_statics[0]
 390
 391     def run_assetlink():
 392         printer = CollectingPrinter()
 393
 394         link_plugin_assets(
 395             plugin_static, linked_assets_dir, printer)
 396
 397         return printer
 398
 399     # it shouldn't exist yet
 400     assert not os.path.lexists(plugin_link_dir)
 401
 402     # link dir doesn't exist, link it
 403     result = run_assetlink().collection[0]
 404     assert result == \
 405         'Linked asset directory for plugin "staticstuff":\n  %s\nto:\n  %s\n' % (
 406             plugin_static.file_path.rstrip(os.path.sep),
 407             plugin_link_dir)
 408     assert os.path.lexists(plugin_link_dir)
 409     assert os.path.islink(plugin_link_dir)
 410     assert os.path.realpath(plugin_link_dir) == plugin_static.file_path
 411
 412     # link dir exists, leave it alone
 413     # (and it should exist still since we just ran it..)
 414     result = run_assetlink().collection[0]
 415     assert result == 'Skipping "staticstuff"; already set up.\n'
 416     assert os.path.lexists(plugin_link_dir)
 417     assert os.path.islink(plugin_link_dir)
 418     assert os.path.realpath(plugin_link_dir) == plugin_static.file_path
 419
 420     # link dir exists, is a symlink to somewhere else (re-link)
 421     junk_file_path = os.path.join(
 422         linked_assets_dir.rstrip(os.path.sep),
 423         'junk.txt')
 424     with open(junk_file_path, 'w') as junk_file:
 425         junk_file.write('barf')
 426
 427     os.unlink(plugin_link_dir)
 428     os.symlink(junk_file_path, plugin_link_dir)
 429
 430     result = run_assetlink().combined_string
 431     assert result == """Old link found for "staticstuff"; removing.
 432 Linked asset directory for plugin "staticstuff":
 433   %s
 434 to:
 435   %s
 436 """ % (plugin_static.file_path.rstrip(os.path.sep), plugin_link_dir)
 437     assert os.path.lexists(plugin_link_dir)
 438     assert os.path.islink(plugin_link_dir)
 439     assert os.path.realpath(plugin_link_dir) == plugin_static.file_path
 440
 441     # link dir exists, but is a non-symlink
 442     os.unlink(plugin_link_dir)
 443     with open(plugin_link_dir, 'w') as clobber_file:
 444         clobber_file.write('clobbered!')
 445
 446     result = run_assetlink().collection[0]
 447     assert result == 'Could not link "staticstuff": %s exists and is not a symlink\n' % (
 448         plugin_link_dir)
 449
 450     with open(plugin_link_dir, 'r') as clobber_file:
 451         assert clobber_file.read() == 'clobbered!'
 452
 453
 454 def test_plugin_staticdirect(static_plugin_app):
 455     """
 456     Test that the staticdirect utilities pull up the right things
 457     """
 458     result = json.loads(
 459         static_plugin_app.get('/staticstuff/').body.decode())
 460
 461     assert len(result) == 2
 462
 463     assert result['mgoblin_bunny_pic'] == '/test_static/images/bunny_pic.png'
 464     assert result['plugin_bunny_css'] == \
 465         '/plugin_static/staticstuff/css/bunnify.css'