本文整理汇总了Python中tests.checks.common.Fixtures类的典型用法代码示例。如果您正苦于以下问题：Python Fixtures类的具体用法？Python Fixtures怎么用？Python Fixtures使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了Fixtures类的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: requests_get_mock
def requests_get_mock(*args, **kwargs):
class MockResponse:
def __init__(self, json_data, status_code):
self.json_data = json_data
self.status_code = status_code
def json(self):
print self.json_data
return json.loads(self.json_data)
def raise_for_status(self):
return True
print 'DEBUG: {0}'.format(args[0])
print NAME_SYSTEM_STATE_URL
if args[0] == NAME_SYSTEM_STATE_URL:
print 'here'
with open(Fixtures.file('hdfs_namesystem_state', sdk_dir=FIXTURE_DIR), 'r') as f:
body = f.read()
return MockResponse(body, 200)
elif args[0] == NAME_SYSTEM_URL:
print 'here'
with open(Fixtures.file('hdfs_namesystem', sdk_dir=FIXTURE_DIR), 'r') as f:
body = f.read()
return MockResponse(body, 200)
开发者ID:dblackdblack，项目名称:integrations-core，代码行数:27，代码来源:test_hdfs_namenode.py
示例2: requests_get_mock
def requests_get_mock(*args, **kwargs):
class MockResponse:
def __init__(self, json_data, status_code):
self.json_data = json_data
self.status_code = status_code
def json(self):
return json.loads(self.json_data)
def raise_for_status(self):
return True
if args[0] == YARN_CLUSTER_METRICS_URL:
with open(Fixtures.file('cluster_metrics', sdk_dir=FIXTURE_DIR), 'r') as f:
body = f.read()
return MockResponse(body, 200)
elif args[0] == YARN_APPS_URL:
with open(Fixtures.file('apps_metrics', sdk_dir=FIXTURE_DIR), 'r') as f:
body = f.read()
return MockResponse(body, 200)
elif args[0] == YARN_NODES_URL:
with open(Fixtures.file('nodes_metrics', sdk_dir=FIXTURE_DIR), 'r') as f:
body = f.read()
return MockResponse(body, 200)
开发者ID:derekjn，项目名称:integrations-core，代码行数:27，代码来源:test_yarn.py
示例3: test_nginx_plus
def test_nginx_plus(self):
test_data = Fixtures.read_file('nginx_plus_in.json')
expected = eval(Fixtures.read_file('nginx_plus_out.python'))
nginx = load_check('nginx', self.config, self.agent_config)
parsed = nginx.parse_json(test_data)
parsed.sort()
self.assertEquals(parsed, expected)
开发者ID:Shopify，项目名称:dd-agent，代码行数:7，代码来源:test_nginx.py
示例4: test_checks
def test_checks(self):
config = {
'init_config': {},
'instances': [
{
'url': 'http://localhost:5051',
'tasks': ['hello']
}
]
}
mocks = {
'_get_stats': lambda x, y, z: json.loads(
Fixtures.read_file('stats.json', sdk_dir=self.FIXTURE_DIR)),
'_get_state': lambda x, y, z: json.loads(
Fixtures.read_file('state.json', sdk_dir=self.FIXTURE_DIR))
}
klass = get_check_class('mesos_slave')
check = klass('mesos_slave', {}, {})
self.run_check_twice(config, mocks=mocks)
metrics = {}
for d in (check.SLAVE_TASKS_METRICS, check.SYSTEM_METRICS, check.SLAVE_RESOURCE_METRICS,
check.SLAVE_EXECUTORS_METRICS, check.STATS_METRICS):
metrics.update(d)
[self.assertMetric(v[0]) for k, v in check.TASK_METRICS.iteritems()]
[self.assertMetric(v[0]) for k, v in metrics.iteritems()]
self.assertServiceCheck('hello.ok', count=1, status=AgentCheck.OK)
开发者ID:dblackdblack，项目名称:integrations-core，代码行数:28，代码来源:test_mesos_slave.py
示例5: side_effect
def side_effect(url, timeout, auth):
if "v2/apps" in url:
return Fixtures.read_json_file("apps.json")
elif "v2/deployments" in url:
return Fixtures.read_json_file("deployments.json")
else:
raise Exception("unknown url:" + url)
开发者ID:Everlane，项目名称:dd-agent，代码行数:7，代码来源:test_marathon.py
示例6: requests_get_mock
def requests_get_mock(*args, **kwargs):
class MockResponse:
def __init__(self, json_data, status_code):
self.json_data = json_data
self.status_code = status_code
def json(self):
return json.loads(self.json_data)
def raise_for_status(self):
return True
ci_dir = os.path.join(os.path.dirname(os.path.realpath(__file__)), "ci")
if args[0] == YARN_APPS_URL:
with open(Fixtures.file('apps_metrics', sdk_dir=ci_dir), 'r') as f:
body = f.read()
return MockResponse(body, 200)
elif args[0] == MR_JOBS_URL:
with open(Fixtures.file('job_metrics', sdk_dir=ci_dir), 'r') as f:
body = f.read()
return MockResponse(body, 200)
elif args[0] == MR_JOB_COUNTERS_URL:
with open(Fixtures.file('job_counter_metrics', sdk_dir=ci_dir), 'r') as f:
body = f.read()
return MockResponse(body, 200)
elif args[0] == MR_TASKS_URL:
with open(Fixtures.file('task_metrics', sdk_dir=ci_dir), 'r') as f:
body = f.read()
return MockResponse(body, 200)
开发者ID:dblackdblack，项目名称:integrations-core，代码行数:33，代码来源:test_mapreduce.py
示例7: test_checks
def test_checks(self):
config = {
'init_config': {},
'instances': [
{
'url': 'http://localhost:5050'
}
]
}
mocks = {
'_get_master_roles': lambda x, y, z: json.loads(Fixtures.read_file('roles.json')),
'_get_master_stats': lambda x, y, z: json.loads(Fixtures.read_file('stats.json')),
'_get_master_state': lambda x, y, z: json.loads(Fixtures.read_file('state.json')),
}
klass = get_check_class('mesos_master')
check = klass('mesos_master', {}, {})
self.run_check_twice(config, mocks=mocks)
metrics = {}
for d in (check.CLUSTER_TASKS_METRICS, check.CLUSTER_SLAVES_METRICS,
check.CLUSTER_RESOURCES_METRICS, check.CLUSTER_REGISTRAR_METRICS,
check.CLUSTER_FRAMEWORK_METRICS, check.SYSTEM_METRICS, check.STATS_METRICS):
metrics.update(d)
[self.assertMetric(v[0]) for k, v in check.FRAMEWORK_METRICS.iteritems()]
[self.assertMetric(v[0]) for k, v in metrics.iteritems()]
[self.assertMetric(v[0]) for k, v in check.ROLE_RESOURCES_METRICS.iteritems()]
self.assertMetric('mesos.cluster.total_frameworks')
self.assertMetric('mesos.framework.total_tasks')
self.assertMetric('mesos.role.frameworks.count')
self.assertMetric('mesos.role.weight')
开发者ID:alanbover，项目名称:dd-agent，代码行数:31，代码来源:test_mesos_master.py
示例8: side_effect
def side_effect(url, timeout, auth, acs_url, verify):
if "v2/apps" in url:
return Fixtures.read_json_file("apps.json", sdk_dir=ci_dir)
elif "v2/deployments" in url:
return Fixtures.read_json_file("deployments.json", sdk_dir=ci_dir)
elif "v2/queue" in url:
return Fixtures.read_json_file("queue.json", sdk_dir=ci_dir)
else:
raise Exception("unknown url:" + url)
开发者ID:dblackdblack，项目名称:integrations-core，代码行数:9，代码来源:test_marathon.py
示例9: test_nginx_plus
def test_nginx_plus(self):
test_data = Fixtures.read_file('nginx_plus_in.json', sdk_dir=FIXTURE_DIR)
expected = eval(Fixtures.read_file('nginx_plus_out.python', sdk_dir=FIXTURE_DIR))
nginx = load_check('nginx', self.config, self.agent_config)
parsed = nginx.parse_json(test_data)
parsed.sort()
# Check that the parsed test data is the same as the expected output
self.assertEquals(parsed, expected)
开发者ID:dblackdblack，项目名称:integrations-core，代码行数:9，代码来源:test_nginx.py
示例10: test_metrics
def test_metrics(self):
# To avoid the disparition of some gauges during the second check
mocks = {
'_retrieve_metrics': lambda x: json.loads(Fixtures.read_file("metrics.json")),
'_retrieve_kube_labels': lambda: json.loads(Fixtures.read_file("kube_labels.json")),
# parts of the json returned by the kubelet api is escaped, keep it untouched
'_retrieve_pods_list': lambda: json.loads(Fixtures.read_file("pods_list.json", string_escape=False)),
}
config = {
"instances": [
{
"host": "foo",
"enable_kubelet_checks": False
}
]
}
# Can't use run_check_twice due to specific metrics
self.run_check_twice(config, mocks=mocks, force_reload=True)
expected_tags = [
(['container_name:/kubelet', 'pod_name:no_pod'], [MEM, CPU, NET, DISK]),
(['kube_replication_controller:propjoe', 'kube_namespace:default', 'container_name:k8s_POD.e4cc795_propjoe-dhdzk_default_ba151259-36e0-11e5-84ce-42010af01c62_ef0ed5f9', 'pod_name:default/propjoe-dhdzk'], [MEM, CPU, FS, NET, NET_ERRORS]),
(['container_name:/kube-proxy', 'pod_name:no_pod'], [MEM, CPU, NET]),
(['kube_replication_controller:kube-dns-v8', 'kube_namespace:kube-system', 'container_name:k8s_POD.2688308a_kube-dns-v8-smhcb_kube-system_b80ffab3-3619-11e5-84ce-42010af01c62_295f14ff', 'pod_name:kube-system/kube-dns-v8-smhcb'], [MEM, CPU, FS, NET, NET_ERRORS]),
(['container_name:/docker-daemon', 'pod_name:no_pod'], [MEM, CPU, DISK, NET]),
(['kube_replication_controller:kube-dns-v8', 'kube_namespace:kube-system', 'container_name:k8s_etcd.2e44beff_kube-dns-v8-smhcb_kube-system_b80ffab3-3619-11e5-84ce-42010af01c62_e3e504ad', 'pod_name:kube-system/kube-dns-v8-smhcb'], [MEM, CPU, FS, NET, NET_ERRORS, DISK]),
(['kube_replication_controller:fluentd-cloud-logging-kubernetes-minion', 'kube_namespace:kube-system', 'container_name:k8s_POD.e4cc795_fluentd-cloud-logging-kubernetes-minion-mu4w_kube-system_d0feac1ad02da9e97c4bf67970ece7a1_49dd977d', 'pod_name:kube-system/fluentd-cloud-logging-kubernetes-minion-mu4w'], [MEM, CPU, FS, NET, NET_ERRORS, DISK]),
(['kube_replication_controller:kube-dns-v8', 'kube_namespace:kube-system', 'container_name:k8s_skydns.1e752dc0_kube-dns-v8-smhcb_kube-system_b80ffab3-3619-11e5-84ce-42010af01c62_7c1345a1', 'pod_name:kube-system/kube-dns-v8-smhcb'], [MEM, CPU, FS, NET, NET_ERRORS]),
(['container_name:/', 'pod_name:no_pod'], [MEM, CPU, FS, NET, NET_ERRORS, DISK]),
(['container_name:/system/docker', 'pod_name:no_pod'], [MEM, CPU, DISK, NET]),
(['kube_replication_controller:propjoe', 'kube_namespace:default', 'container_name:k8s_propjoe.21f63023_propjoe-dhdzk_default_ba151259-36e0-11e5-84ce-42010af01c62_19879457', 'pod_name:default/propjoe-dhdzk'], [MEM, CPU, FS, NET, NET_ERRORS]),
(['container_name:/system', 'pod_name:no_pod'], [MEM, CPU, NET, DISK]),
(['kube_replication_controller:kube-ui-v1', 'kube_namespace:kube-system', 'container_name:k8s_POD.3b46e8b9_kube-ui-v1-sv2sq_kube-system_b7e8f250-3619-11e5-84ce-42010af01c62_209ed1dc', 'pod_name:kube-system/kube-ui-v1-sv2sq'], [MEM, CPU, FS, NET, NET_ERRORS]),
(['kube_replication_controller:kube-dns-v8', 'kube_namespace:kube-system', 'container_name:k8s_kube2sky.1afa6a47_kube-dns-v8-smhcb_kube-system_b80ffab3-3619-11e5-84ce-42010af01c62_624bc34c', 'pod_name:kube-system/kube-dns-v8-smhcb'], [MEM, CPU, FS, NET, NET_ERRORS]),
(['kube_replication_controller:propjoe', 'kube_namespace:default', 'container_name:k8s_POD.e4cc795_propjoe-lkc3l_default_3a9b1759-4055-11e5-84ce-42010af01c62_45d1185b', 'pod_name:default/propjoe-lkc3l'], [MEM, CPU, FS, NET, NET_ERRORS]),
(['kube_replication_controller:haproxy-6db79c7bbcac01601ac35bcdb18868b3', 'kube_namespace:default', 'container_name:k8s_POD.e4cc795_haproxy-6db79c7bbcac01601ac35bcdb18868b3-rr7la_default_86527bf8-36cd-11e5-84ce-42010af01c62_5ad59bf3', 'pod_name:default/haproxy-6db79c7bbcac01601ac35bcdb18868b3-rr7la'], [MEM, CPU, FS, NET, NET_ERRORS]),
(['kube_replication_controller:haproxy-6db79c7bbcac01601ac35bcdb18868b3', 'kube_namespace:default', 'container_name:k8s_haproxy.69b6303b_haproxy-6db79c7bbcac01601ac35bcdb18868b3-rr7la_default_86527bf8-36cd-11e5-84ce-42010af01c62_a35b9731', 'pod_name:default/haproxy-6db79c7bbcac01601ac35bcdb18868b3-rr7la'], [MEM, CPU, FS, NET, NET_ERRORS]),
(['kube_replication_controller:kube-ui-v1','kube_namespace:kube-system', 'container_name:k8s_kube-ui.c17839c_kube-ui-v1-sv2sq_kube-system_b7e8f250-3619-11e5-84ce-42010af01c62_d2b9aa90', 'pod_name:kube-system/kube-ui-v1-sv2sq'], [MEM, CPU, FS, NET, NET_ERRORS]),
(['kube_replication_controller:propjoe','kube_namespace:default', 'container_name:k8s_propjoe.21f63023_propjoe-lkc3l_default_3a9b1759-4055-11e5-84ce-42010af01c62_9fe8b7b0', 'pod_name:default/propjoe-lkc3l'], [MEM, CPU, FS, NET, NET_ERRORS]),
(['kube_replication_controller:kube-dns-v8','kube_namespace:kube-system', 'container_name:k8s_healthz.4469a25d_kube-dns-v8-smhcb_kube-system_b80ffab3-3619-11e5-84ce-42010af01c62_241c34d1', 'pod_name:kube-system/kube-dns-v8-smhcb'], [MEM, CPU, FS, NET, NET_ERRORS, DISK]),
(['kube_replication_controller:fluentd-cloud-logging-kubernetes-minion','kube_namespace:kube-system', 'container_name:k8s_fluentd-cloud-logging.7721935b_fluentd-cloud-logging-kubernetes-minion-mu4w_kube-system_d0feac1ad02da9e97c4bf67970ece7a1_2c3c0879', 'pod_name:kube-system/fluentd-cloud-logging-kubernetes-minion-mu4w'], [MEM, CPU, FS, NET, NET_ERRORS, DISK]),
(['container_name:dd-agent', 'pod_name:no_pod'], [MEM, CPU, FS, NET, NET_ERRORS, DISK]),
(['kube_replication_controller:l7-lb-controller'], [PODS]),
(['kube_replication_controller:redis-slave'], [PODS]),
(['kube_replication_controller:frontend'], [PODS]),
(['kube_replication_controller:heapster-v11'], [PODS]),
]
for m, _type in METRICS:
for tags, types in expected_tags:
if _type in types:
self.assertMetric(m, count=1, tags=tags)
self.coverage_report()
开发者ID:DavidXArnold，项目名称:dd-agent，代码行数:54，代码来源:test_kubernetes.py
示例11: test_fail_1_1
def test_fail_1_1(self):
# To avoid the disparition of some gauges during the second check
mocks = {"_retrieve_metrics": lambda x: json.loads(Fixtures.read_file("metrics_1.1.json"))}
config = {"instances": [{"host": "foo"}]}
with mock.patch(
"utils.kubeutil.KubeUtil.retrieve_pods_list",
side_effect=lambda: json.loads(Fixtures.read_file("pods_list_1.1.json", string_escape=False)),
):
# Can't use run_check_twice due to specific metrics
self.run_check(config, mocks=mocks, force_reload=True)
self.assertServiceCheck("kubernetes.kubelet.check", status=AgentCheck.CRITICAL, tags=None, count=1)
开发者ID:WPMedia，项目名称:dd-agent，代码行数:12，代码来源:test_kubernetes.py
示例12: test_fail
def test_fail(self):
# To avoid the disparition of some gauges during the second check
mocks = {
'_retrieve_metrics': lambda x: json.loads(Fixtures.read_file("metrics.json")),
'_retrieve_kube_labels': lambda: json.loads(Fixtures.read_file("kube_labels.json")),
}
config = {
"instances": [{"host": "foo"}]
}
# Can't use run_check_twice due to specific metrics
self.run_check(config, mocks=mocks, force_reload=True)
self.assertServiceCheck("kubernetes.kubelet.check", status=AgentCheck.CRITICAL)
开发者ID:bobmarthin，项目名称:dd-agent，代码行数:13，代码来源:test_kubernetes.py
示例13: testSpeed
def testSpeed(self):
# Pretend to be gmetad and serve a large piece of content
original_file = Fixtures.file('ganglia.txt')
subprocess.Popen("nc -l 8651 < %s" % original_file, shell=True)
# Wait for 1 second
time.sleep(1)
g = Ganglia(logging.getLogger(__file__))
parsed = StringIO(g.check({'ganglia_host': 'localhost', 'ganglia_port': 8651}))
original = Fixtures.file('ganglia.txt')
x1 = tree.parse(parsed)
x2 = tree.parse(original)
# Cursory test
self.assertEquals([c.tag for c in x1.getroot()], [c.tag for c in x2.getroot()])
开发者ID:7040210，项目名称:dd-agent，代码行数:14，代码来源:test_ganglia.py
示例14: test_osd_status_metrics
def test_osd_status_metrics(self):
mocks = {
'_collect_raw': lambda x,y: json.loads(Fixtures.read_file('ceph_10.2.2.json')),
}
config = {
'instances': [{'host': 'foo'}]
}
self.run_check_twice(config, mocks=mocks, force_reload=True)
for osd, pct_used in [('osd1', 94), ('osd2', 95)]:
expected_tags = ['ceph_fsid:e0efcf84-e8ed-4916-8ce1-9c70242d390a','ceph_mon_state:leader',
'ceph_osd:%s' % osd]
for metric in ['ceph.osd.pct_used']:
self.assertMetric(metric, value=pct_used, count=1, tags=expected_tags)
self.assertMetric('ceph.num_full_osds', value=1, count=1,
tags=['ceph_fsid:e0efcf84-e8ed-4916-8ce1-9c70242d390a', 'ceph_mon_state:leader'])
self.assertMetric('ceph.num_near_full_osds', value=1, count=1,
tags=['ceph_fsid:e0efcf84-e8ed-4916-8ce1-9c70242d390a', 'ceph_mon_state:leader'])
for pool in ['rbd', 'scbench']:
expected_tags = ['ceph_fsid:e0efcf84-e8ed-4916-8ce1-9c70242d390a','ceph_mon_state:leader',
'ceph_pool:%s' % pool]
expected_metrics = ['ceph.read_op_per_sec', 'ceph.write_op_per_sec', 'ceph.op_per_sec']
for metric in expected_metrics:
self.assertMetric(metric, count=1, tags=expected_tags)
开发者ID:AltSchool，项目名称:dd-agent，代码行数:28，代码来源:test_ceph.py
示例15: test__fetch_host_data
def test__fetch_host_data(self):
"""
Test with both 1.1 and 1.2 version payloads
"""
with mock.patch('utils.kubernetes.KubeUtil.retrieve_pods_list') as mock_pods:
self.kubeutil.host_name = 'dd-agent-1rxlh'
mock_pods.return_value = json.loads(Fixtures.read_file("pods_list_1.2.json", string_escape=False))
self.kubeutil._fetch_host_data()
self.assertEqual(self.kubeutil._node_ip, '10.240.0.9')
self.assertEqual(self.kubeutil._node_name, 'kubernetes-massi-minion-k23m')
self.kubeutil.host_name = 'heapster-v11-l8sh1'
mock_pods.return_value = json.loads(Fixtures.read_file("pods_list_1.1.json", string_escape=False))
self.kubeutil._fetch_host_data()
self.assertEqual(self.kubeutil._node_ip, '10.240.0.9')
self.assertEqual(self.kubeutil._node_name, 'gke-cluster-1-8046fdfa-node-ld35')
开发者ID:Everlane，项目名称:dd-agent，代码行数:16，代码来源:test_kubernetes.py
示例16: test_historate
def test_historate(self):
# To avoid the disparition of some gauges during the second check
mocks = {
'_retrieve_metrics': lambda x: json.loads(Fixtures.read_file("metrics.json")),
'_retrieve_kube_labels': lambda: json.loads(Fixtures.read_file("kube_labels.json")),
# parts of the json returned by the kubelet api is escaped, keep it untouched
'_retrieve_pods_list': lambda: json.loads(Fixtures.read_file("pods_list.json", string_escape=False)),
}
config = {
"instances": [
{
"host": "foo",
"enable_kubelet_checks": False,
"use_histogram": True,
}
]
}
# Can't use run_check_twice due to specific metrics
self.run_check_twice(config, mocks=mocks, force_reload=True)
metric_suffix = ["count", "avg", "median", "max", "95percentile"]
expected_tags = [
(['pod_name:no_pod'], [MEM, CPU, NET, DISK, DISK_USAGE, NET_ERRORS]),
(['kube_replication_controller:propjoe', 'kube_namespace:default', 'pod_name:default/propjoe-dhdzk'], [MEM, CPU, FS, NET, NET_ERRORS]),
(['kube_replication_controller:kube-dns-v8', 'kube_namespace:kube-system', 'pod_name:kube-system/kube-dns-v8-smhcb'], [MEM, CPU, FS, NET, NET_ERRORS, DISK]),
(['kube_replication_controller:fluentd-cloud-logging-kubernetes-minion', 'kube_namespace:kube-system', 'pod_name:kube-system/fluentd-cloud-logging-kubernetes-minion-mu4w'], [MEM, CPU, FS, NET, NET_ERRORS, DISK]),
(['kube_replication_controller:kube-dns-v8', 'kube_namespace:kube-system', 'pod_name:kube-system/kube-dns-v8-smhcb'], [MEM, CPU, FS, NET, NET_ERRORS]),
(['kube_replication_controller:propjoe', 'kube_namespace:default', 'pod_name:default/propjoe-dhdzk'], [MEM, CPU, FS, NET, NET_ERRORS]),
(['kube_replication_controller:kube-ui-v1','kube_namespace:kube-system', 'pod_name:kube-system/kube-ui-v1-sv2sq'], [MEM, CPU, FS, NET, NET_ERRORS]),
(['kube_replication_controller:propjoe', 'kube_namespace:default', 'pod_name:default/propjoe-lkc3l'], [MEM, CPU, FS, NET, NET_ERRORS]),
(['kube_replication_controller:haproxy-6db79c7bbcac01601ac35bcdb18868b3', 'kube_namespace:default', 'pod_name:default/haproxy-6db79c7bbcac01601ac35bcdb18868b3-rr7la'], [MEM, CPU, FS, NET, NET_ERRORS]),
(['kube_replication_controller:l7-lb-controller'], [PODS]),
(['kube_replication_controller:redis-slave'], [PODS]),
(['kube_replication_controller:frontend'], [PODS]),
(['kube_replication_controller:heapster-v11'], [PODS]),
]
for m, _type in METRICS:
for m_suffix in metric_suffix:
for tags, types in expected_tags:
if _type in types:
self.assertMetric("{0}.{1}".format(m, m_suffix), count=1, tags=tags)
self.coverage_report()
开发者ID:DavidXArnold，项目名称:dd-agent，代码行数:46，代码来源:test_kubernetes.py
示例17: __init__
def __init__(self, *args, **kwargs):
unittest.TestCase.__init__(self, *args, **kwargs)
self.config = {"instances": [{
"access_id":"foo",
"access_secret": "bar"}]}
self.check = load_check(self.CHECK_NAME, self.config, {})
self.check._connect = Mock(return_value=(None, None, ["aggregation_key:localhost:8080"]))
self.check._get_stats = Mock(return_value=self.check.load_json(
Fixtures.read_file('riakcs_in.json')))
开发者ID:7040210，项目名称:dd-agent，代码行数:9，代码来源:test_riakcs.py
示例18: test_extract_meta
def test_extract_meta(self):
"""
Test with both 1.1 and 1.2 version payloads
"""
res = self.kubeutil.extract_meta({}, 'foo')
self.assertEqual(len(res), 0)
pods = json.loads(Fixtures.read_file("pods_list_1.1.json", string_escape=False))
res = self.kubeutil.extract_meta(pods, 'foo')
self.assertEqual(len(res), 0)
res = self.kubeutil.extract_meta(pods, 'uid')
self.assertEqual(len(res), 6)
pods = json.loads(Fixtures.read_file("pods_list_1.2.json", string_escape=False))
res = self.kubeutil.extract_meta(pods, 'foo')
self.assertEqual(len(res), 0)
res = self.kubeutil.extract_meta(pods, 'uid')
self.assertEqual(len(res), 4)
开发者ID:Everlane，项目名称:dd-agent，代码行数:18，代码来源:test_kubernetes.py
示例19: test_21_metrics
def test_21_metrics(self):
self.run_check(self.config)
expected = eval(Fixtures.read_file('riakcs21_metrics.python', sdk_dir=FIXTURE_DIR))
for m in expected:
self.assertMetric(m[0], m[2], m[3].get('tags', []),
metric_type=m[3]["type"])
# verify non-default (and not in config) metric is not sent
with self.assertRaises(AssertionError):
self.assertMetric("riakcs.bucket_policy_get_in_one")
开发者ID:dblackdblack，项目名称:integrations-core，代码行数:9，代码来源:test_riakcs.py
示例20: testSpeed
def testSpeed(self):
# Pretend to be gmetad and serve a large piece of content
original_file = Fixtures.file('ganglia.txt')
server = subprocess.Popen("nc -l 8651 < %s" % original_file, shell=True)
# Wait for 1 second
time.sleep(1)
pfile = tempfile.NamedTemporaryFile()
g = Ganglia(logging.getLogger(__file__))
# Running the profiler
# profile.runctx("g.check({'ganglia_host': 'localhost', 'ganglia_port': 8651})", {}, {"g": g}, pfile.name)
# p = pstats.Stats(pfile.name)
# p.sort_stats('time').print_stats()
parsed = StringIO(g.check({'ganglia_host': 'localhost', 'ganglia_port': 8651}))
original = Fixtures.file('ganglia.txt')
x1 = tree.parse(parsed)
x2 = tree.parse(original)
# Cursory test
self.assertEquals([c.tag for c in x1.getroot()], [c.tag for c in x2.getroot()])
开发者ID:KnownSubset，项目名称:dd-agent，代码行数:19，代码来源:test_ganglia.py
注：本文中的tests.checks.common.Fixtures类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论