← Back to branch summary

~0x44/nova/bug838466

« back to all changes in this revision

Viewing changes to nova/tests/fake_utils.py

  • Committer: Chuck Short
  • Date: 2011-03-25 20:06:38 UTC
  • mfrom: (892 nova)
  • mto: This revision was merged to the branch mainline in revision 909.
  • Revision ID: zulcss@ubuntu.com-20110325200638-ymwssue0epam4pcj
merged trunk

Show diffs side-by-side

added added

removed removed

Lines of Context:
nova/tests/fake_utils.py
 
1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
 
2
 
 
3
# Copyright (c) 2011 Citrix Systems, Inc.
 
4
#
 
5
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
 
6
#    not use this file except in compliance with the License. You may obtain
 
7
#    a copy of the License at
 
8
#
 
9
#         http://www.apache.org/licenses/LICENSE-2.0
 
10
#
 
11
#    Unless required by applicable law or agreed to in writing, software
 
12
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 
13
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 
14
#    License for the specific language governing permissions and limitations
 
15
#    under the License.
 
16
 
 
17
"""This modules stubs out functions in nova.utils
 
18
"""
 
19
 
 
20
import re
 
21
import types
 
22
 
 
23
from eventlet import greenthread
 
24
 
 
25
from nova import exception
 
26
from nova import log as logging
 
27
from nova import utils
 
28
 
 
29
LOG = logging.getLogger('nova.tests.fake_utils')
 
30
 
 
31
_fake_execute_repliers = []
 
32
_fake_execute_log = []
 
33
 
 
34
 
 
35
def fake_execute_get_log():
 
36
    return _fake_execute_log
 
37
 
 
38
 
 
39
def fake_execute_clear_log():
 
40
    global _fake_execute_log
 
41
    _fake_execute_log = []
 
42
 
 
43
 
 
44
def fake_execute_set_repliers(repliers):
 
45
    """Allows the client to configure replies to commands"""
 
46
    global _fake_execute_repliers
 
47
    _fake_execute_repliers = repliers
 
48
 
 
49
 
 
50
def fake_execute_default_reply_handler(*ignore_args, **ignore_kwargs):
 
51
    """A reply handler for commands that haven't been added to the reply
 
52
    list.  Returns empty strings for stdout and stderr
 
53
    """
 
54
    return '', ''
 
55
 
 
56
 
 
57
def fake_execute(*cmd_parts, **kwargs):
 
58
    """This function stubs out execute, optionally executing
 
59
    a preconfigued function to return expected data
 
60
    """
 
61
    global _fake_execute_repliers
 
62
 
 
63
    process_input = kwargs.get('process_input', None)
 
64
    addl_env = kwargs.get('addl_env', None)
 
65
    check_exit_code = kwargs.get('check_exit_code', 0)
 
66
    cmd_str = ' '.join(str(part) for part in cmd_parts)
 
67
 
 
68
    LOG.debug(_("Faking execution of cmd (subprocess): %s"), cmd_str)
 
69
    _fake_execute_log.append(cmd_str)
 
70
 
 
71
    reply_handler = fake_execute_default_reply_handler
 
72
 
 
73
    for fake_replier in _fake_execute_repliers:
 
74
        if re.match(fake_replier[0], cmd_str):
 
75
            reply_handler = fake_replier[1]
 
76
            LOG.debug(_('Faked command matched %s') % fake_replier[0])
 
77
            break
 
78
 
 
79
    if isinstance(reply_handler, basestring):
 
80
        # If the reply handler is a string, return it as stdout
 
81
        reply = reply_handler, ''
 
82
    else:
 
83
        try:
 
84
            # Alternative is a function, so call it
 
85
            reply = reply_handler(cmd_parts,
 
86
                                  process_input=process_input,
 
87
                                  addl_env=addl_env,
 
88
                                  check_exit_code=check_exit_code)
 
89
        except exception.ProcessExecutionError as e:
 
90
            LOG.debug(_('Faked command raised an exception %s' % str(e)))
 
91
            raise
 
92
 
 
93
    stdout = reply[0]
 
94
    stderr = reply[1]
 
95
    LOG.debug(_("Reply to faked command is stdout='%(stdout)s' "
 
96
                "stderr='%(stderr)s'") % locals())
 
97
 
 
98
    # Replicate the sleep call in the real function
 
99
    greenthread.sleep(0)
 
100
    return reply
 
101
 
 
102
 
 
103
def stub_out_utils_execute(stubs):
 
104
    fake_execute_set_repliers([])
 
105
    fake_execute_clear_log()
 
106
    stubs.Set(utils, 'execute', fake_execute)