def test_importlib_import_relative_class_raises(self):
assert __package__
with self.assertRaises(ImportError):
importlib.__import__('msg.TestMsg', globals=globals(), level=1)
# TODO
# def test_double_import_uses_cache(self): #
# print_importers()
# # Verify that files exists and are importable
# import std_msgs.msg as std_msgs
#
# self.assertTrue(std_msgs.Bool is not None)
# self.assertTrue(callable(std_msgs.Bool))
# self.assertTrue(std_msgs.Bool._type == 'std_msgs/Bool')
#
# import std_msgs.msg as std_msgs2
#
# self.assertTrue(std_msgs == std_msgs2)
python类__import__()的实例源码
def test_importlib_import_relative_srv(self):
# Verify that files exists and are importable
test_srvs = importlib.__import__('srv', globals=globals(), level=1)
test_msgs = importlib.__import__('msg', globals=globals(), level=1)
self.assert_test_service_classes(test_srvs.TestSrv, test_srvs.TestSrvRequest, test_srvs.TestSrvResponse,
test_srvs.TestSrvDeps, test_srvs.TestSrvDepsRequest, test_srvs.TestSrvDepsResponse,
test_msgs.TestRosMsgDeps, test_msgs.TestRosMsg)
# TODO
# def test_double_import_uses_cache(self): #
# print_importers()
# # Verify that files exists and are importable
# import std_msgs.msg as std_msgs
#
# self.assertTrue(std_msgs.Bool is not None)
# self.assertTrue(callable(std_msgs.Bool))
# self.assertTrue(std_msgs.Bool._type == 'std_msgs/Bool')
#
# import std_msgs.msg as std_msgs2
#
# self.assertTrue(std_msgs == std_msgs2)
def bench(name, cleanup=lambda: None, *, seconds=1, repeat=3):
"""Bench the given statement as many times as necessary until total
executions take one second."""
stmt = "__import__({!r})".format(name)
timer = timeit.Timer(stmt)
for x in range(repeat):
total_time = 0
count = 0
while total_time < seconds:
try:
total_time += timer.timeit(1)
finally:
cleanup()
count += 1
else:
# One execution too far
if total_time > seconds:
count -= 1
yield count // seconds
def main(import_):
__builtins__.__import__ = import_
benchmarks = (from_cache, builtin_mod,
source_using_bytecode, source_wo_bytecode,
source_writing_bytecode,
decimal_using_bytecode, decimal_writing_bytecode,
decimal_wo_bytecode,)
seconds = 1
seconds_plural = 's' if seconds > 1 else ''
repeat = 3
header = "Measuring imports/second over {} second{}, best out of {}\n"
print(header.format(seconds, seconds_plural, repeat))
for benchmark in benchmarks:
print(benchmark.__doc__, "[", end=' ')
sys.stdout.flush()
results = []
for result in benchmark(seconds=seconds, repeat=repeat):
results.append(result)
print(result, end=' ')
sys.stdout.flush()
assert not sys.dont_write_bytecode
print("]", "best is", format(max(results), ',d'))
def importModule(self, baseName, module):
"""Load a python module that is a loadable Qudi module.
@param string baseName: the module base package (hardware, logic, or gui)
@param string module: the python module name inside the base package
@return object: the loaded python module
"""
logger.info('Loading module ".{0}.{1}"'.format(baseName, module))
if not isBase(baseName):
raise Exception('You are trying to cheat the '
'system with some category {0}'.format(baseName))
# load the python module
mod = importlib.__import__('{0}.{1}'.format(
baseName, module), fromlist=['*'])
# print('refcnt:', sys.getrefcount(mod))
return mod
def load(name):
try:
obj = __import__(name)
reload(obj)
return obj
except:
pass
try:
import importlib
obj = importlib.__import__(name)
importlib.reload(obj)
return obj
except:
pass
def test_importlib_import_absolute_msg(self):
# Verify that files exists and are importable
std_msgs = importlib.__import__('std_msgs.msg')
std_msgs = std_msgs.msg
self.assert_std_message_classes(std_msgs.Bool, std_msgs.Header)
def test_importlib_import_absolute_class_raises(self):
with self.assertRaises(ImportError):
importlib.__import__('std_msgs.msg.Bool')
# BROKEN 3.4 ?
def test_importlib_import_relative_msg(self):
# Verify that files exists and are importable
test_msgs = importlib.__import__('msg', globals=globals(), level=1)
self.assert_test_message_classes(test_msgs.TestMsg, test_msgs.TestMsgDeps, test_msgs.TestRosMsgDeps, test_msgs.TestRosMsg)
# BROKEN 3.4 ?
def test_importlib_import_absolute_srv(self):
# Verify that files exists and are importable
std_srvs = importlib.__import__('std_srvs.srv')
std_srvs = std_srvs.srv
self.assert_std_service_classes(std_srvs.SetBool, std_srvs.SetBoolRequest, std_srvs.SetBoolResponse)
def test_importlib_import_absolute_msg(self):
# Verify that files exists and are importable
std_msgs = importlib.__import__('std_msgs.msg')
std_msgs = std_msgs.msg
self.assert_std_message_classes(std_msgs.Bool, std_msgs.Header)
def test_importlib_import_absolute_class_raises(self):
with self.assertRaises(ImportError):
importlib.__import__('std_msgs.msg.Bool')
# BROKEN 3.4 ?
def test_importlib_import_relative_msg(self):
# Verify that files exists and are importable
subtest_msgs = importlib.__import__('msg', globals=globals(), level=1)
test_msgs = importlib.__import__('test_rosimport.msg')
test_msgs = test_msgs.msg
self.assert_test_message_classes(subtest_msgs.SubTestMsg, subtest_msgs.SubTestMsgDeps, test_msgs.TestRosMsgDeps, test_msgs.TestRosMsg)
# BROKEN 3.4 ?
def test_importlib_import_absolute_srv(self):
# Verify that files exists and are importable
# __import__ checks sys.modules by itself
# but the test is not reflecting anything if we use the already loaded module.
if sys.modules.get('std_srvs.srv'):
#TODO : EVERYWHERE !
raise unittest.SkipTest("module previously loaded".format('std_srvs.srv'))
else:
std_srvs = importlib.__import__('std_srvs.srv')
std_srvs = std_srvs.srv
self.assert_std_service_classes(std_srvs.SetBool, std_srvs.SetBoolRequest, std_srvs.SetBoolResponse)
def load(name):
try:
obj = __import__(name)
reload(obj)
return obj
except:
pass
try:
import importlib
obj = importlib.__import__(name)
importlib.reload(obj)
return obj
except:
pass
def import_(*args, **kwargs):
"""Delegate to allow for injecting different implementations of import."""
if using___import__:
return __import__(*args, **kwargs)
else:
return importlib.__import__(*args, **kwargs)
def importlib_only(fxn):
"""Decorator to skip a test if using __builtins__.__import__."""
return unittest.skipIf(using___import__, "importlib-specific test")(fxn)
def import_(*args, **kwargs):
"""Delegate to allow for injecting different implementations of import."""
if using___import__:
return __import__(*args, **kwargs)
else:
return importlib.__import__(*args, **kwargs)
def importlib_only(fxn):
"""Decorator to skip a test if using __builtins__.__import__."""
return unittest.skipIf(using___import__, "importlib-specific test")(fxn)
def generate_cf_dynamo_schema(target):
dynamo_connection = DynamoDBConnection()
class FakeClient(object):
def create_table(self, *args, **kwargs):
write_schema_to_yaml(target, **kwargs)
return {}
client = FakeClient()
dynamo_connection.client = client
class FakeDynamo(object):
def list_tables(self):
return []
def create_table(self, *args):
result = dynamo_connection.create_table(*args)
def describe_table(self, *args):
StatusStruct = namedtuple('Status', 'status')
return StatusStruct(status='ACTIVE')
dynamo = FakeDynamo()
engine = Engine()
engine.dynamo = dynamo
sys.path = ['{}/app/models'.format(target)] + sys.path
modelModules = glob.glob('{}/app/models'.format(target) + '/*.py')
models = [basename(f)[:-3] for f in modelModules if isfile(f)]
for modelName in models:
if modelName != '__init__':
engine.register(getattr(importlib.__import__(modelName), modelName))
engine.create_schema()
importloader.py 文件源码
项目：shadowsocksr-python
作者: nanqinlang-shadowsocksr
项目源码
文件源码
阅读 20
收藏 0
点赞 0
评论 0
def load(name):
try:
obj = __import__(name)
reload(obj)
return obj
except:
pass
try:
import importlib
obj = importlib.__import__(name)
importlib.reload(obj)
return obj
except:
pass
def load(name):
try:
obj = __import__(name)
reload(obj)
return obj
except:
pass
try:
import importlib
obj = importlib.__import__(name)
importlib.reload(obj)
return obj
except:
pass
def load(name):
try:
obj = __import__(name)
reload(obj)
return obj
except:
pass
try:
import importlib
obj = importlib.__import__(name)
importlib.reload(obj)
return obj
except:
pass
def load(name):
try:
obj = __import__(name)
reload(obj)
return obj
except:
pass
try:
import importlib
obj = importlib.__import__(name)
importlib.reload(obj)
return obj
except:
pass
def load(name):
try:
obj = __import__(name)
reload(obj)
return obj
except:
pass
try:
import importlib
obj = importlib.__import__(name)
importlib.reload(obj)
return obj
except:
pass
def rundb(target):
"""
Start running a local DynamoDB instance.
:param target:
:return:
"""
load_env(target)
os.environ['AWS_REGION'] = 'us-west-2'
shared_db = './dynamo_db/shared-local-instance.db'
if os.path.exists(shared_db):
os.remove(shared_db)
dynamo_command = ['java', '-Djava.library.path={}/dynamo_db/DynamoDBLocal_lib'.format(CWD), '-jar', '{}/dynamo_db/DynamoDBLocal.jar'.format(CWD), '-sharedDb', '-dbPath', './dynamo_db']
try:
dynamo_process = subprocess.Popen(dynamo_command, stdin=subprocess.PIPE, stderr=subprocess.STDOUT)
except Exception as e:
pass
try:
'''
Connect to DynamoDB and register and create tables for application models.
'''
engine = Engine()
engine.connect(os.environ['AWS_REGION'], host='localhost',
port=8000,
access_key='anything',
secret_key='anything',
is_secure=False)
# load models
sys.path = ['./app/models'] + sys.path
modelModules = glob.glob('./app/models' + "/*.py")
models = [basename(f)[:-3] for f in modelModules if isfile(f)]
for modelName in models:
if modelName != '__init__':
engine.register(getattr(__import__(modelName), modelName))
engine.create_schema()
tables = [table for table in engine.dynamo.list_tables()]
print("This engine has the following tables " + str(tables))
for table in tables:
engine.dynamo.describe_table(table)
except Exception as e:
# IF anything goes wrong, then we self-destruct.
dynamo_process.kill()
raise e
# Wait for process to finish.
dynamo_process.wait()
