[a6d42a]: Allura / allura / tests / test_commands.py History

Parent: [7536f4] (diff)

Child: [e42fa4] (diff)

Download this file

test_commands.py    101 lines (89 with data), 3.2 kB 

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
from os import path, environ



import mock

from nose.tools import assert_raises



from tg import config

from paste.deploy import loadapp

from paste.script.appinstall import SetupCommand

from pylons import c, g



from alluratest.controller import setup_basic_test, setup_global_objects

from allura.command import reactor, script

from allura import model as M





test_config = 'test.ini#main'



def setUp(self):

    """Method called by nose before running each test"""

    #setup_basic_test(app_name='main_with_amqp')

    setup_basic_test()

    setup_global_objects()



def test_reactor_setup():

    cmd = reactor.ReactorSetupCommand('setup')

    cmd.args = [ test_config ]

    cmd.command()



def test_reactor():

    cmd = reactor.ReactorCommand('reactor')

    cmd.args = [ test_config ]

    cmd.options = mock.Mock()

    cmd.options.dry_run = True

    cmd.options.proc = 1

    configs = cmd.command()

    cmd.multi_worker_main(configs)

    cmd.periodic_main()



def test_reactor_callbacks():

    ok_id = M.Project.query.get(shortname='test')._id

    bad_id = None

    malformed_id = 'foo'

    def test_callback(callback, msg):

        msg.data = dict(project_id=malformed_id,

                mount_point='Wiki',

                user_id='badf00d')

        callback(msg.data, msg)

        msg.data = dict(project_id=bad_id,

                mount_point='Wiki',

                user_id='badf00d')

        callback(msg.data, msg)

        msg.data = dict(project_id=ok_id,

                mount_point='Wiki',

                user_id=M.User.anonymous()._id)

        callback(msg.data, msg)

        msg.data = dict(project_id=ok_id,

                mount_point='Wiki')

        callback(msg.data, msg)

        msg.data = dict(project_id=ok_id)

        callback(msg.data, msg)

        msg.data = dict()

        callback(msg.data, msg)

    cmd = reactor.ReactorCommand('reactor')

    cmd.args = [ test_config ]

    cmd.options = mock.Mock()

    cmd.options.dry_run = True

    cmd.options.proc = 1

    configs = cmd.command()

    g.set_project('test')

    g.set_app('wiki')

    # a_callback = cmd.route_audit('Wiki', c.app.__class__.auditor)

    # ac_callback = cmd.route_audit('Wiki', c.app.__class__.class_auditor)

    r_callback = cmd.route_react('Wiki', c.app.__class__.reactor)

    # rc_callback = cmd.route_react('Wiki', c.app.__class__.reactor3)

    msg = mock.Mock()

    msg.ack = lambda:None

    msg.delivery_info = dict(

        routing_key='Wiki.test')

    # test_callback(a_callback, msg)

    # test_callback(ac_callback, msg)

    test_callback(r_callback, msg)

    # test_callback(rc_callback, msg)



def test_send_message():

    cmd = reactor.SendMessageCommand('send_message')

    cmd.args = [ test_config, 'audit', 'nobody.listening', '{}' ]

    cmd.options = mock.Mock()

    cmd.options.context = '/p/test/Wiki/'

    cmd.command()

    cmd.options.context = '/p/test/'

    cmd.command()

    cmd.options.context = None

    cmd.command()



def test_script():

    cmd = script.ScriptCommand('script')

    cmd.args = [ test_config, 'allura/tests/tscript.py' ]

    cmd.command()

    cmd.args = [ test_config, 'allura/tests/tscript_error.py' ]

    assert_raises(ValueError, cmd.command)