def test_comments(self):
b = """
import sys # Foo
sys.exitfunc = f # Blah
"""
a = """
import sys
import atexit # Foo
atexit.register(f) # Blah
"""
self.check(b, a)
b = """
import apples, sys, crumbs, larry # Pleasant comments
sys.exitfunc = func
"""
a = """
import apples, sys, crumbs, larry, atexit # Pleasant comments
atexit.register(func)
"""
self.check(b, a)
python类exitfunc()的实例源码
def run_all(argv=None):
sys.exitfunc = lambda: sys.stderr.write('Shutting down....\n')
# always insert coverage when running tests through setup.py
if argv is None:
argv = [
'nosetests', '--with-xunit',
'--logging-format=%(levelname)s %(name)22s %(funcName)22s:%(lineno)-4d %(message)s',
'--with-xcoverage', '--cover-package=kibtool', '--cover-erase',
'--verbose',
]
nose.run_exit(
argv=argv,
defaultTest=abspath(dirname(__file__))
)
def test_comments(self):
b = """
import sys # Foo
sys.exitfunc = f # Blah
"""
a = """
import sys
import atexit # Foo
atexit.register(f) # Blah
"""
self.check(b, a)
b = """
import apples, sys, crumbs, larry # Pleasant comments
sys.exitfunc = func
"""
a = """
import apples, sys, crumbs, larry, atexit # Pleasant comments
atexit.register(func)
"""
self.check(b, a)
def test_comments(self):
b = """
import sys # Foo
sys.exitfunc = f # Blah
"""
a = """
import sys
import atexit # Foo
atexit.register(f) # Blah
"""
self.check(b, a)
b = """
import apples, sys, crumbs, larry # Pleasant comments
sys.exitfunc = func
"""
a = """
import apples, sys, crumbs, larry, atexit # Pleasant comments
atexit.register(func)
"""
self.check(b, a)
def test_comments(self):
b = """
import sys # Foo
sys.exitfunc = f # Blah
"""
a = """
import sys
import atexit # Foo
atexit.register(f) # Blah
"""
self.check(b, a)
b = """
import apples, sys, crumbs, larry # Pleasant comments
sys.exitfunc = func
"""
a = """
import apples, sys, crumbs, larry, atexit # Pleasant comments
atexit.register(func)
"""
self.check(b, a)
def test_scan_code__basic(monkeypatch, use_ast):
code = """
import os.path
from sys import maxint, exitfunc, platform
del exitfunc
def testfunc():
import shutil
"""
module = __scan_code(code, use_ast, monkeypatch)
assert len(module._deferred_imports) == 3
assert ([di[1][0] for di in module._deferred_imports]
== ['os.path', 'sys', 'shutil'])
assert module.is_global_attr('maxint')
assert module.is_global_attr('os')
assert module.is_global_attr('platform')
assert not module.is_global_attr('shutil') # not imported at module level
assert not module.is_global_attr('exitfunc')
def test_simple(self):
b = """
import sys
sys.exitfunc = my_atexit
"""
a = """
import sys
import atexit
atexit.register(my_atexit)
"""
self.check(b, a)
def test_names_import(self):
b = """
import sys, crumbs
sys.exitfunc = my_func
"""
a = """
import sys, crumbs, atexit
atexit.register(my_func)
"""
self.check(b, a)
def test_complex_expression(self):
b = """
import sys
sys.exitfunc = do(d)/a()+complex(f=23, g=23)*expression
"""
a = """
import sys
import atexit
atexit.register(do(d)/a()+complex(f=23, g=23)*expression)
"""
self.check(b, a)
def test_in_a_function(self):
b = """
import sys
def f():
sys.exitfunc = func
"""
a = """
import sys
import atexit
def f():
atexit.register(func)
"""
self.check(b, a)
def test_unchanged(self):
s = """f(sys.exitfunc)"""
self.unchanged(s)
def test_sys_override(self):
# be sure a preset sys.exitfunc is handled properly
exfunc = sys.exitfunc
sys.exitfunc = self.h1
reload(atexit)
try:
atexit.register(self.h2)
atexit._run_exitfuncs()
finally:
sys.exitfunc = exfunc
self.assertEqual(self.subst_io.getvalue(), "h2\nh1\n")
def exit():
"""Exit subprocess, possibly after first deleting sys.exitfunc
If config-main.cfg/.def 'General' 'delete-exitfunc' is True, then any
sys.exitfunc will be removed before exiting. (VPython support)
"""
if no_exitfunc:
try:
del sys.exitfunc
except AttributeError:
pass
capture_warnings(False)
sys.exit(0)
def test_simple(self):
b = """
import sys
sys.exitfunc = my_atexit
"""
a = """
import sys
import atexit
atexit.register(my_atexit)
"""
self.check(b, a)
def test_names_import(self):
b = """
import sys, crumbs
sys.exitfunc = my_func
"""
a = """
import sys, crumbs, atexit
atexit.register(my_func)
"""
self.check(b, a)
def test_complex_expression(self):
b = """
import sys
sys.exitfunc = do(d)/a()+complex(f=23, g=23)*expression
"""
a = """
import sys
import atexit
atexit.register(do(d)/a()+complex(f=23, g=23)*expression)
"""
self.check(b, a)
def test_in_a_function(self):
b = """
import sys
def f():
sys.exitfunc = func
"""
a = """
import sys
import atexit
def f():
atexit.register(func)
"""
self.check(b, a)
def test_no_sys_import(self):
b = """sys.exitfunc = f"""
a = """atexit.register(f)"""
msg = ("Can't find sys import; Please add an atexit import at the "
"top of your file.")
self.warns(b, a, msg)
def test_unchanged(self):
s = """f(sys.exitfunc)"""
self.unchanged(s)
def test_sys_override(self):
# be sure a preset sys.exitfunc is handled properly
exfunc = sys.exitfunc
sys.exitfunc = self.h1
reload(atexit)
try:
atexit.register(self.h2)
atexit._run_exitfuncs()
finally:
sys.exitfunc = exfunc
self.assertEqual(self.subst_io.getvalue(), "h2\nh1\n")
def exit():
"""Exit subprocess, possibly after first deleting sys.exitfunc
If config-main.cfg/.def 'General' 'delete-exitfunc' is True, then any
sys.exitfunc will be removed before exiting. (VPython support)
"""
if no_exitfunc:
try:
del sys.exitfunc
except AttributeError:
pass
capture_warnings(False)
sys.exit(0)
def test_simple(self):
b = """
import sys
sys.exitfunc = my_atexit
"""
a = """
import sys
import atexit
atexit.register(my_atexit)
"""
self.check(b, a)
def test_names_import(self):
b = """
import sys, crumbs
sys.exitfunc = my_func
"""
a = """
import sys, crumbs, atexit
atexit.register(my_func)
"""
self.check(b, a)
def test_complex_expression(self):
b = """
import sys
sys.exitfunc = do(d)/a()+complex(f=23, g=23)*expression
"""
a = """
import sys
import atexit
atexit.register(do(d)/a()+complex(f=23, g=23)*expression)
"""
self.check(b, a)
def test_in_a_function(self):
b = """
import sys
def f():
sys.exitfunc = func
"""
a = """
import sys
import atexit
def f():
atexit.register(func)
"""
self.check(b, a)
def test_no_sys_import(self):
b = """sys.exitfunc = f"""
a = """atexit.register(f)"""
msg = ("Can't find sys import; Please add an atexit import at the "
"top of your file.")
self.warns(b, a, msg)
def test_unchanged(self):
s = """f(sys.exitfunc)"""
self.unchanged(s)
def run(self):
# Close connection on exit (to test cleanup paths)
self.tid = get_tid()
log.info('starting thread: {} (TID {})'.format(self.name, self.tid))
old_exitfunc = getattr(sys, 'exitfunc', None)
def exit():
log.info('Closing hypervisors connexions')
for hyp_id, hostname in self.hyps.items():
self.hyps_conn[hyp_id].close()
if (old_exitfunc): old_exitfunc()
sys.exitfunc = exit
# self.r_status = RethinkHypEvent()
self.thread_event_loop = virEventLoopNativeStart()
for hyp_id, hostname in self.hyps.items():
self.add_hyp_to_receive_events(hyp_id)
while self.stop is not True:
time.sleep(0.1)
if self.stop is True:
for hyp_id in self.hyps:
self.del_hyp_to_receive_events(hyp_id)
def run(self):
# Close connection on exit (to test cleanup paths)
self.tid = get_tid()
log.info('starting thread: {} (TID {})'.format(self.name, self.tid))
old_exitfunc = getattr(sys, 'exitfunc', None)
def exit():
log.info('Closing hypervisors connexions')
for hyp_id, hostname in self.hyps.items():
self.hyps_conn[hyp_id].close()
if (old_exitfunc): old_exitfunc()
sys.exitfunc = exit
# self.r_status = RethinkHypEvent()
self.thread_event_loop = virEventLoopNativeStart()
for hyp_id, hostname in self.hyps.items():
self.add_hyp_to_receive_events(hyp_id)
while self.stop is not True:
time.sleep(0.1)
if self.stop is True:
for hyp_id in self.hyps:
self.del_hyp_to_receive_events(hyp_id)
def start(argv=None):
sys.exitfunc = lambda: sys.stderr.write('Shutting down...\n')
if argv is None:
argv = [
'nosetests',
'--verbose',
'--with-coverage',
'--cover-html', '--cover-html-dir=.htmlcov',
'--cover-erase',
'--cover-branches',
'--cover-package=chemtrails',
]
nose.run_exit(argv=argv, defaultTest=os.path.abspath(os.path.dirname(__file__)))
