¡@
OpenStack Study: fake_utils.py**** CubicPower OpenStack Study **** def fake_execute_get_log(): return _fake_execute_log
**** CubicPower OpenStack Study **** def fake_execute_clear_log(): global _fake_execute_log _fake_execute_log = []
**** CubicPower OpenStack Study **** def fake_execute_set_repliers(repliers): """Allows the client to configure replies to commands.""" global _fake_execute_repliers _fake_execute_repliers = repliers
**** CubicPower OpenStack Study **** def fake_execute_default_reply_handler(*ignore_args, **ignore_kwargs): """A reply handler for commands that haven't been added to the reply list. Returns empty strings for stdout and stderr. """ return '', ''
**** CubicPower OpenStack Study **** def fake_execute(*cmd_parts, **kwargs): """This function stubs out execute. It optionally executes a preconfigued function to return expected data. """ global _fake_execute_repliers process_input = kwargs.get('process_input', None) check_exit_code = kwargs.get('check_exit_code', 0) delay_on_retry = kwargs.get('delay_on_retry', True) attempts = kwargs.get('attempts', 1) run_as_root = kwargs.get('run_as_root', False) cmd_str = ' '.join(str(part) for part in cmd_parts) LOG.debug(_("Faking execution of cmd (subprocess): %s"), cmd_str) _fake_execute_log.append(cmd_str) reply_handler = fake_execute_default_reply_handler for fake_replier in _fake_execute_repliers: if re.match(fake_replier[0], cmd_str): reply_handler = fake_replier[1] LOG.debug(_('Faked command matched %s') % fake_replier[0]) break if isinstance(reply_handler, basestring): # If the reply handler is a string, return it as stdout reply = reply_handler, '' else: try: # Alternative is a function, so call it reply = reply_handler(cmd_parts, process_input=process_input, delay_on_retry=delay_on_retry, attempts=attempts, run_as_root=run_as_root, check_exit_code=check_exit_code) except processutils.ProcessExecutionError as e: LOG.debug(_('Faked command raised an exception %s'), e) raise LOG.debug(_("Reply to faked command is stdout='%(stdout)s' " "stderr='%(stderr)s'") % {'stdout': reply[0], 'stderr': reply[1]}) # Replicate the sleep call in the real function greenthread.sleep(0) return reply
**** CubicPower OpenStack Study **** def stub_out_utils_execute(stubs): fake_execute_set_repliers([]) fake_execute_clear_log() stubs.Set(utils, 'execute', fake_execute)
|