本文整理汇总了Python中mock.call.warning函数的典型用法代码示例。如果您正苦于以下问题：Python warning函数的具体用法？Python warning怎么用？Python warning使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了warning函数的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_apply
def test_apply(self):
with patch('%s._validate' % pb):
cls = TerraformRunner(self.mock_config(), 'terraform-bin')
with patch('%s.logger' % pbm, autospec=True) as mock_logger:
with patch('%s._set_remote' % pb, autospec=True) as mock_set:
with patch('%s._setup_tf' % pb, autospec=True) as mock_setup:
with patch('%s._run_tf' % pb, autospec=True) as mock_run:
with patch('%s._taint_deployment' % pb,
autospec=True) as mock_taint:
mock_run.return_value = 'output'
with patch('%s._show_outputs' % pb,
autospec=True) as mock_show:
cls.apply()
assert mock_setup.mock_calls == [call(cls, stream=False)]
assert mock_set.mock_calls == []
assert mock_run.mock_calls == [
call(cls, 'apply', cmd_args=['-input=false', '-refresh=true', '.'],
stream=False)
]
assert mock_logger.mock_calls == [
call.warning('Running terraform apply: %s',
'-input=false -refresh=true .'),
call.warning("Terraform apply finished successfully:\n%s", 'output')
]
assert mock_show.mock_calls == [call(cls)]
assert mock_taint.mock_calls == [call(cls, stream=False)]
开发者ID:jantman，项目名称:webhook2lambda2sqs，代码行数:26，代码来源:test_terraform_runner.py
示例2: test_generate
def test_generate(self):
with patch('%s.logger' % pbm, autospec=True) as mock_logger:
with patch('%s._get_config' % pb, autospec=True) as mock_get:
with patch('%s.open' % pbm, mock_open(), create=True) as m_open:
with patch('%s._write_zip' % pb, autospec=True) as mock_zip:
mock_get.return_value = 'myjson'
self.cls.generate('myfunc')
assert mock_get.mock_calls == [call(self.cls, 'myfunc')]
assert m_open.mock_calls == [
call('./webhook2lambda2sqs_func.py', 'w'),
call().__enter__(),
call().write('myfunc'),
call().__exit__(None, None, None),
call('./webhook2lambda2sqs.tf.json', 'w'),
call().__enter__(),
call().write('myjson'),
call().__exit__(None, None, None)
]
assert mock_zip.mock_calls == [
call(self.cls, 'myfunc', './webhook2lambda2sqs_func.zip')
]
assert mock_logger.mock_calls == [
call.warning('Writing lambda function source to: '
'./webhook2lambda2sqs_func.py'),
call.debug('lambda function written'),
call.warning('Writing lambda function source zip file to: '
'./webhook2lambda2sqs_func.zip'),
call.debug('lambda zip written'),
call.warning('Writing terraform configuration JSON to: '
'./webhook2lambda2sqs.tf.json'),
call.debug('terraform configuration written'),
call.warning('Completed writing lambda function and TF config.')
]
开发者ID:waffle-iron，项目名称:webhook2lambda2sqs，代码行数:33，代码来源:test_tf_generator.py
示例3: test_apply_stream
def test_apply_stream(self):
def se_exc(*args, **kwargs):
raise Exception('foo')
with patch('%s._validate' % pb):
cls = TerraformRunner(self.mock_config(), 'terraform-bin')
with patch('%s.logger' % pbm, autospec=True) as mock_logger:
with patch('%s._set_remote' % pb, autospec=True) as mock_set:
with patch('%s._run_tf' % pb, autospec=True) as mock_run:
with patch('%s._taint_deployment' % pb,
autospec=True) as mock_taint:
mock_run.return_value = 'output'
mock_taint.side_effect = se_exc
with patch('%s._show_outputs' % pb,
autospec=True) as mock_show:
cls.apply(stream=True)
assert mock_set.mock_calls == [call(cls, stream=True)]
assert mock_run.mock_calls == [
call(cls, 'apply', cmd_args=['-input=false', '-refresh=true', '.'],
stream=True)
]
assert mock_logger.mock_calls == [
call.warning('Running terraform apply: %s',
'-input=false -refresh=true .'),
call.warning("Terraform apply finished successfully.")
]
assert mock_show.mock_calls == [call(cls)]
assert mock_taint.mock_calls == [call(cls, stream=True)]
开发者ID:AndrewFarley，项目名称:webhook2lambda2sqs，代码行数:29，代码来源:test_terraform_runner.py
示例4: test_update_limits_from_api_invalid_region_503
def test_update_limits_from_api_invalid_region_503(self):
resp = {
'ResponseMetadata': {
'HTTPStatusCode': 503,
'RequestId': '7d74c6f0-c789-11e5-82fe-a96cdaa6d564'
},
'Error': {
'Message': 'Service Unavailable',
'Code': '503'
}
}
ce = ClientError(resp, 'GetSendQuota')
def se_get():
raise ce
mock_conn = Mock()
mock_conn.get_send_quota.side_effect = se_get
with patch('%s.connect' % pb) as mock_connect:
with patch('%s.logger' % pbm) as mock_logger:
cls = _SesService(21, 43)
cls.conn = mock_conn
cls._update_limits_from_api()
assert mock_connect.mock_calls == [call()]
assert mock_conn.mock_calls == [call.get_send_quota()]
assert mock_logger.mock_calls == [
call.warning('Skipping SES: %s', ce)
]
assert cls.limits['Daily sending quota'].api_limit is None
开发者ID:jantman，项目名称:awslimitchecker，代码行数:30，代码来源:test_ses.py
示例5: test_get_limit_check_id_subscription_required
def test_get_limit_check_id_subscription_required(self):
def se_api(language=None):
response = {
'ResponseMetadata': {
'HTTPStatusCode': 400,
'RequestId': '3cc9b2a8-c6e5-11e5-bc1d-b13dcea36176'
},
'Error': {
'Message': 'AWS Premium Support Subscription is required '
'to use this service.',
'Code': 'SubscriptionRequiredException'
}
}
raise ClientError(response, 'operation')
assert self.cls.have_ta is True
self.mock_conn.describe_trusted_advisor_checks.side_effect = se_api
with patch('awslimitchecker.trustedadvisor'
'.logger', autospec=True) as mock_logger:
res = self.cls._get_limit_check_id()
assert self.cls.have_ta is False
assert res == (None, None)
assert self.mock_conn.mock_calls == [
call.describe_trusted_advisor_checks(language='en')
]
assert mock_logger.mock_calls == [
call.debug("Querying Trusted Advisor checks"),
call.warning("Cannot check TrustedAdvisor: %s",
'AWS Premium Support Subscription is required to '
'use this service.')
]
开发者ID:EslamElHusseiny，项目名称:awslimitchecker，代码行数:32，代码来源:test_trustedadvisor.py
示例6: test_init_no_sensors
def test_init_no_sensors(self):
with patch('%s.logger' % pbm, autospec=True) as mock_logger:
with patch.multiple(
pb,
autospec=True,
find_host_id=DEFAULT,
discover_engine=DEFAULT,
discover_sensors=DEFAULT,
) as mocks:
with patch('%s._list_classes' % pbm,
autospec=True) as mock_list:
mocks['find_host_id'].return_value = 'myhostid'
mocks['discover_engine'].return_value = (
'foo.bar.baz', 1234
)
with pytest.raises(SystemExit) as excinfo:
SensorDaemon()
assert mock_logger.mock_calls == [
call.warning('This machine running with host_id %s', 'myhostid'),
call.critical('ERROR - no sensors discovered.')
]
assert mocks['find_host_id'].call_count == 1
assert mocks['discover_engine'].call_count == 1
assert mocks['discover_sensors'].call_count == 1
assert excinfo.value.code == 1
assert mock_list.mock_calls == []
开发者ID:jantman，项目名称:RPyMostat-sensor，代码行数:26，代码来源:test_sensor_daemon.py
示例7: test_get_limit_check_id_subscription_required
def test_get_limit_check_id_subscription_required(self):
def se_api(language):
status = 400
reason = 'Bad Request'
body = {
'message': 'AWS Premium Support Subscription is required to '
'use this service.',
'__type': 'SubscriptionRequiredException'
}
raise JSONResponseError(status, reason, body)
self.mock_conn.describe_trusted_advisor_checks.side_effect = se_api
assert self.cls.have_ta is True
with patch('awslimitchecker.trustedadvisor'
'.logger', autospec=True) as mock_logger:
res = self.cls._get_limit_check_id()
assert self.cls.have_ta is False
assert res == (None, None)
assert self.mock_conn.mock_calls == [
call.describe_trusted_advisor_checks('en')
]
assert mock_logger.mock_calls == [
call.debug("Querying Trusted Advisor checks"),
call.warning("Cannot check TrustedAdvisor: %s",
"AWS Premium Support "
"Subscription is required to use this service.")
]
开发者ID:EmilVarona，项目名称:awslimitchecker，代码行数:28，代码来源:test_trustedadvisor.py
示例8: test_init_nondefault
def test_init_nondefault(self):
with patch('%s.setup_signal_handlers' % pb) as mock_setup_signals:
with patch('%s.get_tls_factory' % pb) as mock_get_tls:
with patch('%s.logger' % pbm) as mock_logger:
with patch('%s.getpid' % pbm) as mock_getpid:
mock_getpid.return_value = 12345
cls = VaultRedirector(
'consul:123',
redir_to_https=True,
redir_to_ip=True,
log_disable=True,
poll_interval=1.234,
bind_port=1234,
check_id='foo:bar'
)
assert mock_setup_signals.mock_calls == [call()]
assert mock_logger.mock_calls == [
call.warning(
'Starting VaultRedirector with ALL LOGGING DISABLED; send '
'SIGUSR1 to PID %d enable logging.',
12345
)
]
assert mock_get_tls.mock_calls == []
assert cls.active_node_ip_port is None
assert cls.last_poll_time is None
assert cls.consul_host_port == 'consul:123'
assert cls.redir_https is True
assert cls.redir_ip is True
assert cls.log_enabled is False
assert cls.poll_interval == 1.234
assert cls.bind_port == 1234
assert cls.check_id == 'foo:bar'
assert cls.consul_scheme == 'https'
开发者ID:jantman，项目名称:vault-redirector-twisted，代码行数:34，代码来源:test_redirector.py
示例9: test_find_usage_invalid_region_503
def test_find_usage_invalid_region_503(self):
resp = {
'ResponseMetadata': {
'HTTPStatusCode': 503,
'RequestId': '7d74c6f0-c789-11e5-82fe-a96cdaa6d564'
},
'Error': {
'Message': 'Service Unavailable',
'Code': '503'
}
}
ce = ClientError(resp, 'GetSendQuota')
def se_get():
raise ce
mock_conn = Mock()
mock_conn.get_send_quota.side_effect = se_get
with patch('%s.connect' % pb) as mock_connect:
with patch('%s.logger' % pbm) as mock_logger:
cls = _SesService(21, 43)
cls.conn = mock_conn
assert cls._have_usage is False
cls.find_usage()
assert mock_connect.mock_calls == [call()]
assert cls._have_usage is False
assert mock_logger.mock_calls == [
call.debug('Checking usage for service %s', 'SES'),
call.warning(
'Skipping SES: %s', ce
)
]
assert mock_conn.mock_calls == [call.get_send_quota()]
assert len(cls.limits['Daily sending quota'].get_current_usage()) == 0
开发者ID:jantman，项目名称:awslimitchecker，代码行数:35，代码来源:test_ses.py
示例10: test_show_one_queue_empty
def test_show_one_queue_empty(self, capsys):
conn = Mock()
conn.receive_message.return_value = {}
with patch('%s.logger' % pbm, autospec=True) as mock_logger:
with patch('%s._url_for_queue' % pb, autospec=True) as mock_url:
mock_url.return_value = 'myurl'
with patch('%s._delete_msg' % pb, autospec=True) as mock_del:
self.cls._show_one_queue(conn, 'foo', 1, delete=True)
out, err = capsys.readouterr()
assert err == ''
expected_out = "=> Queue 'foo' appears empty.\n"
assert out == expected_out
assert mock_del.mock_calls == []
assert conn.mock_calls == [
call.receive_message(
QueueUrl='myurl',
AttributeNames=['All'],
MessageAttributeNames=['All'],
MaxNumberOfMessages=1,
WaitTimeSeconds=20
)
]
assert mock_url.mock_calls == [
call(self.cls, conn, 'foo')
]
assert mock_logger.mock_calls == [
call.debug("Queue '%s' url: %s", 'foo', 'myurl'),
call.warning("Receiving %d messages from queue'%s'; this may "
"take up to 20 seconds.", 1, 'foo'),
call.debug('received no messages')
]
开发者ID:waffle-iron，项目名称:webhook2lambda2sqs，代码行数:31，代码来源:test_aws.py
示例11: test_find_usage_spot_instances
def test_find_usage_spot_instances(self):
data = fixtures.test_find_usage_spot_instances
mock_conn = Mock()
mock_client_conn = Mock()
mock_client_conn.describe_spot_instance_requests.return_value = data
cls = _Ec2Service(21, 43)
cls.resource_conn = mock_conn
cls.conn = mock_client_conn
with patch('awslimitchecker.services.ec2.logger') as mock_logger:
cls._find_usage_spot_instances()
assert mock_conn.mock_calls == []
assert mock_client_conn.mock_calls == [
call.describe_spot_instance_requests()
]
lim = cls.limits['Max spot instance requests per region']
usage = lim.get_current_usage()
assert len(usage) == 1
assert usage[0].get_value() == 2
assert mock_logger.mock_calls == [
call.debug('Getting spot instance request usage'),
call.warning('EC2 spot instance support is experimental and '
'results may not me accurate in all cases. Please '
'see the notes at: <http://awslimitchecker'
'.readthedocs.io/en/latest/limits.html#ec2>'),
call.debug('NOT counting spot instance request %s state=%s',
'reqID1', 'closed'),
call.debug('Counting spot instance request %s state=%s',
'reqID2', 'active'),
call.debug('Counting spot instance request %s state=%s',
'reqID3', 'open'),
call.debug('NOT counting spot instance request %s state=%s',
'reqID4', 'failed'),
call.debug('Setting "Max spot instance requests per region" '
'limit (%s) current usage to: %d', lim, 2)
]
开发者ID:bflad，项目名称:awslimitchecker，代码行数:35，代码来源:test_ec2.py
示例12: test_init_default
def test_init_default(self):
sensors = [Mock(), Mock()]
with patch('%s.logger' % pbm, autospec=True) as mock_logger:
with patch.multiple(
pb,
autospec=True,
find_host_id=DEFAULT,
discover_engine=DEFAULT,
discover_sensors=DEFAULT,
) as mocks:
with patch('%s._list_classes' % pbm,
autospec=True) as mock_list:
mocks['find_host_id'].return_value = 'myhostid'
mocks['discover_engine'].return_value = (
'foo.bar.baz', 1234
)
mocks['discover_sensors'].return_value = sensors
cls = SensorDaemon()
assert cls.dry_run is False
assert cls.dummy_data is False
assert cls.engine_port == 1234
assert cls.engine_addr == 'foo.bar.baz'
assert cls.interval == 60.0
assert cls.host_id == 'myhostid'
assert cls.sensors == sensors
assert mock_logger.mock_calls == [
call.warning('This machine running with host_id %s', 'myhostid')
]
assert mocks['find_host_id'].mock_calls == [call(cls)]
assert mocks['discover_engine'].mock_calls == [call(cls)]
assert mocks['discover_sensors'].mock_calls == [call(cls, {})]
assert mock_list.mock_calls == []
开发者ID:jantman，项目名称:RPyMostat-sensor，代码行数:32，代码来源:test_sensor_daemon.py
示例13: test_destroy_stream
def test_destroy_stream(self):
with patch('%s._validate' % pb):
cls = TerraformRunner(self.mock_config(), 'terraform-bin')
with patch('%s.logger' % pbm, autospec=True) as mock_logger:
with patch('%s._set_remote' % pb, autospec=True) as mock_set:
with patch('%s._run_tf' % pb, autospec=True) as mock_run:
mock_run.return_value = 'output'
cls.destroy(stream=True)
assert mock_set.mock_calls == [call(cls, stream=True)]
assert mock_run.mock_calls == [
call(cls, 'destroy', cmd_args=['-refresh=true', '-force', '.'],
stream=True)
]
assert mock_logger.mock_calls == [
call.warning('Running terraform destroy: %s',
'-refresh=true -force .'),
call.warning("Terraform destroy finished successfully.")
]
开发者ID:AndrewFarley，项目名称:webhook2lambda2sqs，代码行数:18，代码来源:test_terraform_runner.py
示例14: test_init_dry_run
def test_init_dry_run(self, mock_post, mock_get, mock_requests, mock_logger):
g = Groups(self.dummy_ice_url, dry_run=True)
self.assertEquals(g.dry_run, True)
self.assertEquals(mock_logger.mock_calls,
[call.warning('DRY RUN only - will not make any changes')]
)
self.assertEquals(mock_requests.mock_calls, [])
self.assertEquals(mock_get.mock_calls, [])
self.assertEquals(mock_post.mock_calls, [])
开发者ID:andrewmcgilvray，项目名称:ice_pick，代码行数:9，代码来源:tests.py
示例15: test_delete_application_group_dry_run
def test_delete_application_group_dry_run(self, mock_post, mock_get, mock_requests, mock_logger):
groups = Groups(self.dummy_ice_url, dry_run=True)
mock_logger.reset_mock()
groups.delete_application_group('foo')
self.assertEquals(mock_logger.mock_calls,
[call.warning('Would GET deleteApplicationGroup?name=foo')]
)
self.assertEquals(mock_get.mock_calls, [])
self.assertEquals(mock_post.mock_calls, [])
开发者ID:andrewmcgilvray，项目名称:ice_pick，代码行数:9，代码来源:tests.py
示例16: test_plan
def test_plan(self):
with patch('%s._validate' % pb):
cls = TerraformRunner(self.mock_config(), 'terraform-bin')
with patch('%s.logger' % pbm, autospec=True) as mock_logger:
with patch('%s._set_remote' % pb, autospec=True) as mock_set:
with patch('%s._run_tf' % pb, autospec=True) as mock_run:
mock_run.return_value = 'output'
cls.plan()
assert mock_set.mock_calls == [call(cls, stream=False)]
assert mock_run.mock_calls == [
call(cls, 'plan', cmd_args=['-input=false', '-refresh=true', '.'],
stream=False)
]
assert mock_logger.mock_calls == [
call.warning('Running terraform plan: %s',
'-input=false -refresh=true .'),
call.warning("Terraform plan finished successfully:\n%s", 'output')
]
开发者ID:AndrewFarley，项目名称:webhook2lambda2sqs，代码行数:18，代码来源:test_terraform_runner.py
示例17: test_taint_deployment_stream
def test_taint_deployment_stream(self):
with patch('%s._validate' % pb):
cls = TerraformRunner(self.mock_config(), 'terraform-bin')
with patch('%s.logger' % pbm, autospec=True) as mock_logger:
with patch('%s._set_remote' % pb, autospec=True) as mock_set:
with patch('%s._run_tf' % pb, autospec=True) as mock_run:
mock_run.return_value = 'output'
cls._taint_deployment(stream=True)
assert mock_set.mock_calls == []
assert mock_run.mock_calls == [
call(cls, 'taint', cmd_args=['aws_api_gateway_deployment.depl'],
stream=True)
]
assert mock_logger.mock_calls == [
call.warning('Running terraform taint: %s as workaround for '
'<https://github.com/hashicorp/terraform/issues/6613>',
'aws_api_gateway_deployment.depl'),
call.warning("Terraform taint finished successfully.")
]
开发者ID:AndrewFarley，项目名称:webhook2lambda2sqs，代码行数:19，代码来源:test_terraform_runner.py
示例18: test_listentcp
def test_listentcp(self):
self.cls.reactor = Mock(spec_set=reactor)
mock_site = Mock()
with patch('%s.logger' % pbm) as mock_logger:
self.cls.listentcp(mock_site)
assert mock_logger.mock_calls == [
call.warning('Setting TCP listener on port %d for HTTP requests',
8080)
]
assert self.cls.reactor.mock_calls == [call.listenTCP(8080, mock_site)]
开发者ID:jantman，项目名称:vault-redirector-twisted，代码行数:10，代码来源:test_redirector.py
示例19: test_ice_post_dry_run
def test_ice_post_dry_run(self, mock_requests, mock_logger):
url = 'http://foo.com/dashboard/foobar'
g = Groups('http://foo.com/', dry_run=True)
mock_logger.reset_mock()
res = g._ice_post('foobar', {'baz': 'blam'})
self.assertEquals(mock_logger.mock_calls,
[call.warning("DRY RUN: Would POST to http://foo.com/dashboard/foobar: {'baz': 'blam'}")]
)
self.assertEquals(mock_requests.mock_calls, [])
开发者ID:andrewmcgilvray，项目名称:ice_pick，代码行数:10，代码来源:tests.py
示例20: test_init_nondefault
def test_init_nondefault(self):
dummy = Mock()
with patch('%s.logger' % pbm, autospec=True) as mock_logger:
with patch.multiple(
pb,
autospec=True,
find_host_id=DEFAULT,
discover_engine=DEFAULT,
discover_sensors=DEFAULT,
) as mocks:
with patch('%s._list_classes' % pbm,
autospec=True) as mock_list:
mocks['find_host_id'].return_value = 'myhostid'
mocks['discover_engine'].return_value = (
'foo.bar.baz', 1234
)
mocks['discover_sensors'].return_value = [dummy]
cls = SensorDaemon(
dry_run=True,
dummy_data=True,
engine_port=1234,
engine_addr='foo.bar.baz',
interval=12.34,
class_args={'foo': 'bar'}
)
assert cls.dry_run is True
assert cls.dummy_data is True
assert cls.engine_port == 1234
assert cls.engine_addr == 'foo.bar.baz'
assert cls.interval == 12.34
assert cls.host_id == 'myhostid'
assert cls.sensors == [dummy]
assert mock_logger.mock_calls == [
call.warning('This machine running with host_id %s', 'myhostid'),
call.warning("DRY RUN MODE - will not POST data to Engine.")
]
assert mocks['find_host_id'].mock_calls == [call(cls)]
assert mocks['discover_engine'].mock_calls == []
assert mocks['discover_sensors'].mock_calls == [
call(cls, {'foo': 'bar'})
]
assert mock_list.mock_calls == []
开发者ID:jantman，项目名称:RPyMostat-sensor，代码行数:42，代码来源:test_sensor_daemon.py
注：本文中的mock.call.warning函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论