本文整理汇总了Python中mzb_test_utils.run_successful_bench函数的典型用法代码示例。如果您正苦于以下问题：Python run_successful_bench函数的具体用法？Python run_successful_bench怎么用？Python run_successful_bench使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了run_successful_bench函数的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: worker_from_git_test
def worker_from_git_test():
# worker is located in the same repo as node
worker_commit = os.environ.get("NODE_COMMIT", "master")
mzbench_repo = os.environ.get("MZBENCH_REPO", "https://github.com/machinezone/mzbench")
run_successful_bench(
scripts_dir + "worker_from_git.erl", env={"worker_branch": worker_commit, "mzbench_repo": mzbench_repo}
)
开发者ID:moigagoo，项目名称:mzbench，代码行数:7，代码来源:mzb_basic_tests.py
示例2: lua_worker_from_git_test
def lua_worker_from_git_test():
worker_commit = os.environ.get('NODE_COMMIT', 'master')
mzbench_repo = os.environ.get('MZBENCH_REPO', 'https://github.com/satori-com/mzbench')
env = {'worker_branch': worker_commit,
'mzbench_repo': mzbench_repo}
run_successful_bench(scripts_dir + 'lua_worker_from_git.erl', env=env)
run_successful_bench(scripts_bdl_dir + 'lua_worker_from_git.bdl', env=env)
开发者ID:dskliarov，项目名称:mzbench，代码行数:7，代码来源:mzb_basic_tests.py
示例3: python_worker_from_git_test
def python_worker_from_git_test():
worker_commit = os.environ.get('NODE_COMMIT', 'master')
mzbench_repo = os.environ.get('MZBENCH_REPO', 'https://github.com/machinezone/mzbench')
run_successful_bench(
scripts_dir + 'python_worker_from_git.erl',
env={'worker_branch': worker_commit,
'mzbench_repo': mzbench_repo})
开发者ID:nkrode，项目名称:mzbench，代码行数:7，代码来源:mzb_basic_tests.py
示例4: env_test
def env_test():
run_successful_bench(scripts_dir + 'env.erl', env={
'jozin': 'bazin',
'wait_ms': '10',
'pool_size': '2',
'loop_time': '5',
'loop_rate': '2'})
开发者ID:nkrode，项目名称:mzbench，代码行数:7，代码来源:mzb_basic_tests.py
示例5: worker_from_git_test
def worker_from_git_test():
# worker is located in the same repo as node
worker_commit = os.environ.get('NODE_COMMIT', 'master')
mzbench_repo = os.environ.get('MZBENCH_REPO', 'https://github.com/machinezone/mzbench')
env = {'worker_branch': worker_commit,
'mzbench_repo': mzbench_repo}
run_successful_bench(scripts_dir + 'worker_from_git.erl', env=env)
run_successful_bench(scripts_bdl_dir + 'worker_from_git.bdl', env=env)
开发者ID:ttsvetanov，项目名称:mzbench，代码行数:8，代码来源:mzb_basic_tests.py
示例6: poisson_worker_start_test
def poisson_worker_start_test():
run_successful_bench(
mzbench_dir + "examples/worker_start_poisson.erl",
expected_log_message_regex="workers\.pool1\.started\.rps = 1\.",
)
run_successful_bench(
mzbench_dir + "examples.bdl/worker_start_poisson.bdl",
expected_log_message_regex="workers\.pool1\.started\.rps = 1\.",
)
开发者ID:timofey-barmin，项目名称:mzbench，代码行数:9，代码来源:mzb_basic_tests.py
示例7: check_migrations_test
def check_migrations_test():
mzbench_data_dir = tempfile.mkdtemp(prefix='mzbench_data_')
with start_mzbench_server(custom_data_location=mzbench_data_dir):
for i in range(5):
run_successful_bench(scripts_dir + 'correct_script.erl')
try:
cmd(mzbench_dir + '/bin/migrate.py ' + mzbench_data_dir)
finally:
shutil.rmtree(mzbench_data_dir)
开发者ID:Briends，项目名称:mzbench，代码行数:11，代码来源:mzb_migrations_tests.py
示例8: hooks_test
def hooks_test():
def check_log(log):
regex = re.compile(r"Run exec hook \"echo pre_hook_1\"(.*)Run exec hook \"echo pre_hook_2\"(.*)Run exec hook \"echo post_hook_1\"", re.DOTALL)
assert regex.search(log)
def check_user_log(log):
regex = re.compile(r"Dummy print: \"bar\"", re.DOTALL)
assert regex.search(log)
run_successful_bench(scripts_dir + 'hooks.erl',
check_log_function=check_log,
check_user_log_function=check_user_log)
开发者ID:nkrode，项目名称:mzbench，代码行数:12，代码来源:mzb_basic_tests.py
示例9: vars_defaults_test
def vars_defaults_test():
def check_log(log):
regexp1 = re.compile('the_var1_value_is_var1_default_value', re.DOTALL + re.UNICODE)
regexp2 = re.compile('the_var2_value_is_var2_new_value', re.DOTALL + re.UNICODE)
if regexp1.search(log) and regexp2.search(log):
return False
else:
return 'Unable to find correct values in log'
run_successful_bench(scripts_dir + 'vars_defaults.erl',
env={'var2': 'var2_new_value'},
check_user_log_function=check_log)
开发者ID:nkrode，项目名称:mzbench，代码行数:13，代码来源:mzb_basic_tests.py
示例10: vars_defaults_test
def vars_defaults_test():
def check_log(log):
regexp1 = re.compile("the_var1_value_is_var1_default_value", re.DOTALL + re.UNICODE)
regexp2 = re.compile("the_var2_value_is_var2_new_value", re.DOTALL + re.UNICODE)
if regexp1.search(log) and regexp2.search(log):
return False
else:
return "Unable to find correct values in log"
run_successful_bench(
scripts_dir + "vars_defaults.erl", env={"var2": "var2_new_value"}, check_log_function=check_log
)
开发者ID:moigagoo，项目名称:mzbench，代码行数:13，代码来源:mzb_basic_tests.py
示例11: data_endpoint_test
def data_endpoint_test():
bench_id = run_successful_bench(scripts_dir + 'data_script.erl')
csv_data_process = subprocess.Popen(
[mzbench_script,
'--host=localhost:4800',
'--format=csv',
'data',
str(bench_id)],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
json_data_process = subprocess.Popen(
[mzbench_script,
'--host=localhost:4800',
'--format=json',
'data',
str(bench_id)],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
log_process = subprocess.Popen(
[mzbench_script,
'--host=localhost:4800',
'log',
str(bench_id)],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
lout, lerr = log_process.communicate()
print 'Log collector stdout'
print lout
print 'Log collector stderr'
print lerr
csv_out, csv_err = csv_data_process.communicate()
print 'CSV data collector stdout'
print csv_out
print 'CSV data collector stderr'
print csv_err
json_out, json_err = json_data_process.communicate()
print 'JSON data collector stdout'
print json_out
print 'JSON data collector stderr'
print json_err
time.sleep(3)
csv_data_ret_code = csv_data_process.poll()
json_data_ret_code = json_data_process.poll()
assert csv_data_ret_code == 0
assert json_data_ret_code == 0
assert 'print,' in csv_out
assert 'print' in\
[metric['target'] for metric in json.loads(json_out)]
开发者ID:nkrode，项目名称:mzbench，代码行数:57，代码来源:mzb_basic_tests.py
示例12: poisson_worker_start_test
def poisson_worker_start_test():
bench_id = run_successful_bench(mzbench_dir + 'examples.bdl/worker_start_poisson.bdl')
output = subprocess.check_output(
[mzbench_script,
'--host=localhost:4800',
'--format=csv',
'data',
str(bench_id)])
assert re.compile('workers\.pool1\.started\.rps,.*,1\.', re.UNICODE).search(output)
开发者ID:dskliarov，项目名称:mzbench，代码行数:11，代码来源:mzb_basic_tests.py
示例13: run_command_test
def run_command_test():
def run_command(bid):
print "Running command for {0}".format(bid)
change_env_process = subprocess.Popen(
[mzbench_script,
'--host=localhost:4800',
'run_command',
str(bid),
'print("zzzzzz999")'],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
change_env_res, change_env_err = change_env_process.communicate()
print 'Running command for {0}\n{1}'.format(change_env_res, change_env_err)
run_successful_bench(
scripts_dir + 'loop_with_vars.erl',
post_start=run_command,
expected_log_message_regex=r'zzzzzz999',
env={'time': '60', 'rate': '1', 'message':'abc'})
开发者ID:dskliarov，项目名称:mzbench，代码行数:20，代码来源:mzb_basic_tests.py
示例14: env_change_test
def env_change_test():
def change_var(bid):
print "Changing env for {0}".format(bid)
change_env_process = subprocess.Popen(
[mzbench_script,
'--host=localhost:4800',
'change_env',
str(bid),
'--env=message=zzz'],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
change_env_res, change_env_err = change_env_process.communicate()
print 'Change env output: {0}\n{1}'.format(change_env_res, change_env_err)
assert('set' == json.loads(change_env_res)['status'])
time.sleep(20)
change_env_process = subprocess.Popen(
[mzbench_script,
'--host=localhost:4800',
'change_env',
str(bid),
'--env=rate=5'],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
change_env_res, change_env_err = change_env_process.communicate()
print 'Change env output: {0}\n{1}'.format(change_env_res, change_env_err)
assert('set' == json.loads(change_env_res)['status'])
bench_id = run_successful_bench(
scripts_dir + 'loop_with_vars.erl',
post_start=change_var,
expected_log_message_regex=r'zzz',
env={'time': '60', 'rate': '1', 'message':'abc'})
json_data_process = subprocess.Popen(
[mzbench_script,
'--host=localhost:4800',
'--format=json',
'data',
str(bench_id)],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
json_out, json_err = json_data_process.communicate()
time.sleep(3)
json_data_ret_code = json_data_process.poll()
assert json_data_ret_code == 0
datapoints = [metric['datapoints'] for metric in json.loads(json_out) if metric['target'] == 'print.rps'][0]
values = [d[0] for d in datapoints]
print "Datapoints: {0}".format(values)
assert(0.8 < values[1] < 1.2)
assert(4.8 < values[4] < 5.2)
开发者ID:nkrode，项目名称:mzbench，代码行数:54，代码来源:mzb_basic_tests.py
示例15: workers_per_node_test
def workers_per_node_test():
workers_per_node = 3
bench_id = run_successful_bench(scripts_dir + "workers_per_node.erl", workers_per_node=workers_per_node)
log_process = subprocess.Popen(
[mzbench_script, "--host=localhost:4800", "log", str(bench_id)], stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
out, err = log_process.communicate()
log = "{0} {1}".format(out, err)
if not re.findall("nodes_arg => 4", log):
print "Out: ", out
print "Err: ", err
raise RuntimeError("The bench should have allocated 4 worker nodes")
开发者ID:moigagoo，项目名称:mzbench，代码行数:13，代码来源:mzb_basic_tests.py
示例16: data_endpoint_test
def data_endpoint_test():
bench_id = run_successful_bench(scripts_dir + "data_script.erl")
csv_data_process = subprocess.Popen(
[mzbench_script, "--host=localhost:4800", "--format=csv", "data", str(bench_id)],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
json_data_process = subprocess.Popen(
[mzbench_script, "--host=localhost:4800", "--format=json", "data", str(bench_id)],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
log_process = subprocess.Popen(
[mzbench_script, "--host=localhost:4800", "log", str(bench_id)], stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
lout, lerr = log_process.communicate()
print "Log collector stdout"
print lout
print "Log collector stderr"
print lerr
csv_out, csv_err = csv_data_process.communicate()
print "CSV data collector stdout"
print csv_out
print "CSV data collector stderr"
print csv_err
json_out, json_err = json_data_process.communicate()
print "JSON data collector stdout"
print json_out
print "JSON data collector stderr"
print json_err
time.sleep(3)
csv_data_ret_code = csv_data_process.poll()
json_data_ret_code = json_data_process.poll()
assert csv_data_ret_code == 0
assert json_data_ret_code == 0
assert "print," in csv_out
assert "print" in [metric["target"] for metric in json.loads(json_out)]
开发者ID:moigagoo，项目名称:mzbench，代码行数:46，代码来源:mzb_basic_tests.py
示例17: env_change_test
def env_change_test():
def change_var(bid):
print "Changing env for {0}".format(bid)
change_env_process = subprocess.Popen(
[mzbench_script, "--host=localhost:4800", "change_env", str(bid), "--env=message=zzz"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
change_env_res, change_env_err = change_env_process.communicate()
print "Change env output: {0}\n{1}".format(change_env_res, change_env_err)
assert "set" == json.loads(change_env_res)["status"]
time.sleep(20)
change_env_process = subprocess.Popen(
[mzbench_script, "--host=localhost:4800", "change_env", str(bid), "--env=rate=5"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
change_env_res, change_env_err = change_env_process.communicate()
print "Change env output: {0}\n{1}".format(change_env_res, change_env_err)
assert "set" == json.loads(change_env_res)["status"]
bench_id = run_successful_bench(
scripts_dir + "loop_with_vars.erl",
post_start=change_var,
expected_log_message_regex=r"zzz",
env={"time": "60", "rate": "1", "message": "abc"},
)
json_data_process = subprocess.Popen(
[mzbench_script, "--host=localhost:4800", "--format=json", "data", str(bench_id)],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
json_out, json_err = json_data_process.communicate()
time.sleep(3)
json_data_ret_code = json_data_process.poll()
assert json_data_ret_code == 0
datapoints = [metric["datapoints"] for metric in json.loads(json_out) if metric["target"] == "print.rps"][0]
values = [d[0] for d in datapoints]
print "Datapoints: {0}".format(values)
assert 0.8 < values[2] < 1.2
assert 4.8 < values[5] < 5.2
开发者ID:moigagoo，项目名称:mzbench，代码行数:45，代码来源:mzb_basic_tests.py
示例18: bench_results_test
def bench_results_test():
bench_id = run_successful_bench(scripts_dir + 'data_script.erl')
json_results_process = subprocess.Popen(
[mzbench_script,
'--host=localhost:4800',
'results',
str(bench_id)],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
log_process = subprocess.Popen(
[mzbench_script,
'--host=localhost:4800',
'log',
str(bench_id)],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
lout, lerr = log_process.communicate()
print 'Log collector stdout'
print lout
print 'Log collector stderr'
print lerr
json_out, json_err = json_results_process.communicate()
print 'JSON results collector stdout'
print json_out
print 'JSON results collector stderr'
print json_err
time.sleep(3)
json_data_ret_code = json_results_process.poll()
assert json_data_ret_code == 0
results = json.loads(json_out)
assert results["dummy"]["type"] == "histogram"
assert results["dummy"]["percentiles"]["max"] > 0
assert results["print"]["type"] == "counter"
assert results["print"]["value"] == 1
开发者ID:dskliarov，项目名称:mzbench，代码行数:42，代码来源:mzb_basic_tests.py
示例19: poisson_worker_start_test
def poisson_worker_start_test():
run_successful_bench(mzbench_dir + 'examples/worker_start_poisson.erl',
expected_log_message_regex='workers\.pool1\.started\.rps = 1\.')
开发者ID:nkrode，项目名称:mzbench，代码行数:3，代码来源:mzb_basic_tests.py
示例20: loop_without_rate_test
def loop_without_rate_test():
run_successful_bench(scripts_dir + 'superloop.erl')
开发者ID:nkrode，项目名称:mzbench，代码行数:2，代码来源:mzb_basic_tests.py
注：本文中的mzb_test_utils.run_successful_bench函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论