[01d9bf]: pyforge / pyforge / tests / test_commands.py Maximize Restore History

Download this file

test_commands.py    92 lines (81 with data), 2.9 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
from os import path, environ
import mock
from tg import config
from paste.deploy import loadapp
from paste.script.appinstall import SetupCommand
from pylons import c, g
from pyforge.command import reactor
from pyforge import model as M
from pyforge.tests import helpers
test_config = '%s#main_with_amqp' % (
environ.get('SANDBOX') and 'sandbox-test.ini' or 'test.ini')
def setUp(self):
"""Method called by nose before running each test"""
helpers.setup_basic_test(app_name='main_with_amqp')
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='hello',
user_id='badf00d')
callback(msg.data, msg)
msg.data = dict(project_id=bad_id,
mount_point='hello',
user_id='badf00d')
callback(msg.data, msg)
msg.data = dict(project_id=ok_id,
mount_point='hello',
user_id=M.User.anonymous()._id)
callback(msg.data, msg)
msg.data = dict(project_id=ok_id,
mount_point='hello')
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('hello')
a_callback = cmd.route_audit('hello_forge', c.app.__class__.auditor)
ac_callback = cmd.route_audit('hello_forge', c.app.__class__.class_auditor)
r_callback = cmd.route_react('hello_forge', c.app.__class__.reactor1)
rc_callback = cmd.route_react('hello_forge', c.app.__class__.reactor3)
msg = mock.Mock()
msg.ack = lambda:None
msg.delivery_info = dict(
routing_key='hello_forge.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 = '/projects/test/hello/'
cmd.command()
cmd.options.context = '/projects/test/'
cmd.command()
cmd.options.context = None
cmd.command()