def plot_compute_node(self):
tag = 'compute'
redirected = False
if sys.stdout == sys.__stdout__:
self.outfile = open(self.compute_dot_file, "w")
sys.stdout = self.outfile
redirected = True
self.__digraph_open(tag)
# Title
self.__cluster_open('ComputeNode', 'red')
self.__cluster_name('Compute Node', 1, 'yellow')
self.__cluster_close()
# Plot nodes
self.__cluster_open_plain('Nova')
self.__plot_vms()
self.__plot_linux_bridge()
self.__cluster_close_plain()
self.__cluster_open_plain('OVS')
self.__plot_br_int_compute()
self.__plot_br_tun(tag)
self.__cluster_close_plain()
# Plot edges
self.__plot_title_edges(tag)
self.__plot_vms_to_linuxbridge()
self.__plot_linuxbridge_to_br_int()
self.__plot_br_int_to_br_tun(tag)
if redirected:
self.__digraph_close()
self.outfile.close()
sys.stdout = sys.__stdout__
python类__stdout__()的实例源码
def plot_network_node(self):
tag = 'network'
redirected = False
if sys.stdout == sys.__stdout__:
self.outfile = open(self.network_dot_file, "w")
sys.stdout = self.outfile
redirected = True
self.__digraph_open(tag)
self.__cluster_open('NetworkNode', 'red')
self.__cluster_name('Network Node', 1, 'yellow')
self.__cluster_close()
# Plot nodes
self.__cluster_open_plain('OVS')
self.__plot_br_ex_network()
self.__plot_br_int_network()
self.__plot_br_tun(tag)
self.__cluster_close_plain()
# Plot edges
self.__plot_title_edges(tag)
self.__plot_br_int_to_br_tun(tag)
self.__plot_br_ex_to_br_int()
if redirected:
self.__digraph_close()
self.outfile.close()
sys.stdout = sys.__stdout__
def debug(msg):
if settings['debug']:
if sys.stdout != sys.__stdout__:
tmp = sys.stdout
sys.stdout = sys.__stdout__
print('DEBUG: ' + msg)
sys.stdout = tmp
else:
print('DEBUG: ' + msg)
def error(msg):
if sys.stdout != sys.__stdout__:
tmp = sys.stdout
sys.stdout = sys.__stdout__
print('ERROR: ' + msg)
sys.stdout = tmp
else:
print('ERROR: ' + msg)
def warning(msg):
if sys.stdout != sys.__stdout__:
tmp = sys.stdout
sys.stdout = sys.__stdout__
print('WARNING: ' + msg)
sys.stdout = tmp
else:
print('WARNING: ' + msg)
def set_trace():
"""Call pdb.set_trace in the calling frame, first restoring
sys.stdout to the real output stream. Note that sys.stdout is NOT
reset to whatever it was before the call once pdb is done!
"""
import pdb
import sys
stdout = sys.stdout
sys.stdout = sys.__stdout__
pdb.Pdb().set_trace(sys._getframe().f_back)
def parseArgs(self, argv):
"""Parse argv and env and configure running environment.
"""
self.config.configure(argv, doc=self.usage())
log.debug("configured %s", self.config)
# quick outs: version, plugins (optparse would have already
# caught and exited on help)
if self.config.options.version:
from nose import __version__
sys.stdout = sys.__stdout__
print "%s version %s" % (os.path.basename(sys.argv[0]), __version__)
sys.exit(0)
if self.config.options.showPlugins:
self.showPlugins()
sys.exit(0)
if self.testLoader is None:
self.testLoader = defaultTestLoader(config=self.config)
elif isclass(self.testLoader):
self.testLoader = self.testLoader(config=self.config)
plug_loader = self.config.plugins.prepareTestLoader(self.testLoader)
if plug_loader is not None:
self.testLoader = plug_loader
log.debug("test loader is %s", self.testLoader)
# FIXME if self.module is a string, add it to self.testNames? not sure
if self.config.testNames:
self.testNames = self.config.testNames
else:
self.testNames = tolist(self.defaultTest)
log.debug('defaultTest %s', self.defaultTest)
log.debug('Test names are %s', self.testNames)
if self.config.workingDir is not None:
os.chdir(self.config.workingDir)
self.createTests()
def __init__(self, string):
self.stdout = sys.__stdout__
self.appends = ''
self.prepends = ''
self.isatty = self.stdout.isatty()
def test_task(self):
class AddTask(Task):
addend_a = properties.Float('First add argument')
addend_b = properties.Float('Second add argument')
class Result(BaseResult):
value = properties.Float('Result of add operation')
def __call__(self):
self.report_status({'progress': 0., 'message': 'Starting'})
if self.addend_a == self.addend_b:
raise PermanentTaskFailure()
return self.Result(value=self.addend_a + self.addend_b)
add = AddTask(addend_a=0., addend_b=10.)
sys.stdout = temp_out = StringIO()
result = add()
sys.stdout = sys.__stdout__
assert temp_out.getvalue() == 'AddTask | 0% | Starting\n'
assert result.value == 10.
add = AddTask(addend_a=5., addend_b=5.)
with self.assertRaises(PermanentTaskFailure):
add()
with self.assertRaises(NotImplementedError):
Task()()
with self.assertRaises(ValueError):
Task().report_status(.5)
def __dir__(self):
return dir(sys.__stdout__)
def __getattribute__(self, name):
if name == '__members__':
return dir(sys.__stdout__)
try:
stream = _local.stream
except AttributeError:
stream = sys.__stdout__
return getattr(stream, name)
def __repr__(self):
return repr(sys.__stdout__)
# add the threaded stream as display hook
def __dir__(self):
return dir(sys.__stdout__)
def __getattribute__(self, name):
if name == '__members__':
return dir(sys.__stdout__)
try:
stream = _local.stream
except AttributeError:
stream = sys.__stdout__
return getattr(stream, name)
def write(self, data):
try:
self.inner_file.write(data)
except:
sys.stdout = sys.__stdout__
log.DefaultObserver.stderr = sys.stderr = sys.__stderr__
raise
def tearDown(self):
sys.stdout = sys.__stdout__
sys.stdin = sys.__stdin__
def __dir__(self):
return dir(sys.__stdout__)
def __getattribute__(self, name):
if name == '__members__':
return dir(sys.__stdout__)
try:
stream = _local.stream
except AttributeError:
stream = sys.__stdout__
return getattr(stream, name)
def __repr__(self):
return repr(sys.__stdout__)
# add the threaded stream as display hook
def __dir__(self):
return dir(sys.__stdout__)
