本文整理汇总了Python中webhook2lambda2sqs.terraform_runner.TerraformRunner类的典型用法代码示例。如果您正苦于以下问题:Python TerraformRunner类的具体用法?Python TerraformRunner怎么用?Python TerraformRunner使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了TerraformRunner类的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: 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
示例2: get_base_url
def get_base_url(config, args):
"""
Get the API base url. Try Terraform state first, then
:py:class:`~.AWSInfo`.
:param config: configuration
:type config: :py:class:`~.Config`
:param args: command line arguments
:type args: :py:class:`argparse.Namespace`
:return: API base URL
:rtype: str
"""
try:
logger.debug('Trying to get Terraform base_url output')
runner = TerraformRunner(config, args.tf_path)
outputs = runner._get_outputs()
base_url = outputs['base_url']
logger.debug("Terraform base_url output: '%s'", base_url)
except Exception:
logger.info('Unable to find API base_url from Terraform state; '
'querying AWS.', exc_info=1)
aws = AWSInfo(config)
base_url = aws.get_api_base_url()
logger.debug("AWS api_base_url: '%s'", base_url)
if not base_url.endswith('/'):
base_url += '/'
return base_url
开发者ID:jantman,项目名称:webhook2lambda2sqs,代码行数:27,代码来源:runner.py
示例3: 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
示例4: get_base_url
def get_base_url(dir_path):
"""
get the base url to the API
"""
os.chdir(dir_path)
conf = Config(os.path.join(dir_path, 'config.json'))
tf_runner = TerraformRunner(conf, 'terraform')
outs = tf_runner._get_outputs()
return outs['base_url']
开发者ID:waffle-iron,项目名称:webhook2lambda2sqs,代码行数:9,代码来源:conftest.py
示例5: test_show_outputs
def test_show_outputs(self, capsys):
outs = {'a': 'b', 'foo': 'bar'}
with patch('%s._get_outputs' % pb, autospec=True) as mock_get:
mock_get.return_value = outs
with patch('%s._validate' % pb):
cls = TerraformRunner(self.mock_config(), 'terraform-bin')
cls._show_outputs()
assert mock_get.mock_calls == [call(cls)]
out, err = capsys.readouterr()
assert err == ''
assert out == "\n\n=> Terraform Outputs:\na = b\nfoo = bar\n"
开发者ID:AndrewFarley,项目名称:webhook2lambda2sqs,代码行数:11,代码来源:test_terraform_runner.py
示例6: test_setup_tf_pre090
def test_setup_tf_pre090(self):
with patch('%s._validate' % pb):
cls = TerraformRunner(self.mock_config(), 'terraform')
cls.tf_version = (0, 7, 2)
with patch('%s.logger' % pbm, autospec=True) as mock_logger:
with patch('%s._set_remote' % pb, autospec=True) as mock_set_rmt:
with patch('%s._run_tf' % pb, autospec=True) as mock_run:
mock_run.return_value = ('myoutput', 0)
cls._setup_tf()
assert mock_set_rmt.mock_calls == [call(cls, stream=False)]
assert mock_run.mock_calls == []
assert mock_logger.mock_calls == []
开发者ID:jantman,项目名称:webhook2lambda2sqs,代码行数:12,代码来源:test_terraform_runner.py
示例7: test_run_tf
def test_run_tf(self):
expected_args = 'terraform plan config foo bar'
with patch('%s._validate' % pb):
cls = TerraformRunner(self.mock_config(), 'terraform')
with patch('%s.logger' % pbm, autospec=True) as mock_logger:
with patch('%s.run_cmd' % pbm, autospec=True) as mock_run:
mock_run.return_value = ('myoutput', 0)
cls._run_tf('plan', cmd_args=['config', 'foo', 'bar'])
assert mock_run.mock_calls == [
call(expected_args, stream=False)
]
assert mock_logger.mock_calls == [
call.info('Running terraform command: %s', expected_args)
]
开发者ID:AndrewFarley,项目名称:webhook2lambda2sqs,代码行数:14,代码来源:test_terraform_runner.py
示例8: test_set_remote_none
def test_set_remote_none(self):
with patch('%s._validate' % pb):
cls = TerraformRunner(self.mock_config(), '/path/to/terraform')
with patch('%s.logger' % pbm, autospec=True) as mock_logger:
with patch('%s._args_for_remote' % pb, autospec=True) as mock_args:
with patch('%s.run_cmd' % pbm, autospec=True) as mock_run:
mock_args.return_value = None
cls._set_remote()
assert mock_args.mock_calls == [call(cls)]
assert mock_run.mock_calls == []
assert mock_logger.mock_calls == [
call.debug('_args_for_remote() returned None; not configuring '
'terraform remote')
]
开发者ID:AndrewFarley,项目名称:webhook2lambda2sqs,代码行数:14,代码来源:test_terraform_runner.py
示例9: test_args_for_remote
def test_args_for_remote(self):
conf = self.mock_config(conf={
'terraform_remote_state': {
'backend': 'consul',
'config': {
'path': 'consul/path',
'address': 'foo:1234'
}
}
})
with patch('%s._validate' % pb):
cls = TerraformRunner(conf, 'tfpath')
assert cls._args_for_remote() == [
'-backend=consul',
'-backend-config="address=foo:1234"',
'-backend-config="path=consul/path"'
]
开发者ID:AndrewFarley,项目名称:webhook2lambda2sqs,代码行数:17,代码来源:test_terraform_runner.py
示例10: main
def main(args=None):
"""
Main entry point
"""
# parse args
if args is None:
args = parse_args(sys.argv[1:])
# dump example config if that action
if args.action == 'example-config':
conf, doc = Config.example_config()
print(conf)
sys.stderr.write(doc + "\n")
return
# set logging level
if args.verbose > 1:
set_log_debug()
elif args.verbose == 1:
set_log_info()
# get our config
config = Config(args.config)
if args.action == 'logs':
aws = AWSInfo(config)
aws.show_cloudwatch_logs(count=args.log_count)
return
if args.action == 'queuepeek':
aws = AWSInfo(config)
aws.show_queue(name=args.queue_name, delete=args.queue_delete,
count=args.msg_count)
return
if args.action == 'test':
run_test(config, args)
return
# if generate or genapply, generate the configs
if args.action == 'generate' or args.action == 'genapply':
func_gen = LambdaFuncGenerator(config)
func_src = func_gen.generate()
# @TODO: also write func_source to disk
tf_gen = TerraformGenerator(config)
tf_gen.generate(func_src)
# if only generate, exit now
if args.action == 'generate':
return
# else setup a Terraform action
runner = TerraformRunner(config, args.tf_path)
# run the terraform action
if args.action == 'apply' or args.action == 'genapply':
runner.apply(args.stream_tf)
elif args.action == 'plan':
runner.plan(args.stream_tf)
else: # destroy
runner.destroy(args.stream_tf)
开发者ID:waffle-iron,项目名称:webhook2lambda2sqs,代码行数:60,代码来源:runner.py
示例11: 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
示例12: 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
示例13: test_set_remote
def test_set_remote(self):
expected_args = ['config', 'foo', 'bar']
with patch('%s._validate' % pb):
cls = TerraformRunner(self.mock_config(), 'terraform')
with patch('%s.logger' % pbm, autospec=True) as mock_logger:
with patch('%s._args_for_remote' % pb, autospec=True) as mock_args:
with patch('%s._run_tf' % pb, autospec=True) as mock_run:
mock_args.return_value = ['foo', 'bar']
mock_run.return_value = ('myoutput', 0)
cls._set_remote()
assert mock_args.mock_calls == [call(cls)]
assert mock_run.mock_calls == [
call(cls, 'remote', cmd_args=expected_args, stream=False)
]
assert mock_logger.mock_calls == [
call.warning('Setting terraform remote config: %s', 'foo bar'),
call.info('Terraform remote configured.')
]
开发者ID:AndrewFarley,项目名称:webhook2lambda2sqs,代码行数:18,代码来源:test_terraform_runner.py
示例14: 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
示例15: test_run_tf_fail
def test_run_tf_fail(self):
expected_args = 'terraform-bin plan config foo bar'
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.run_cmd' % pbm, autospec=True) as mock_run:
mock_run.return_value = ('myoutput', 5)
with pytest.raises(Exception) as excinfo:
cls._run_tf('plan', cmd_args=['config', 'foo', 'bar'],
stream=True)
assert exc_msg(excinfo.value) == 'terraform plan failed'
assert mock_run.mock_calls == [
call(expected_args, stream=True)
]
assert mock_logger.mock_calls == [
call.info('Running terraform command: %s', expected_args),
call.critical('Terraform command (%s) failed with exit code '
'%d:\n%s', expected_args, 5, 'myoutput')
]
开发者ID:AndrewFarley,项目名称:webhook2lambda2sqs,代码行数:19,代码来源:test_terraform_runner.py
示例16: test_get_outputs_0point7plus
def test_get_outputs_0point7plus(self):
expected = {
'base_url': 'https://ljgx260ix7.execute-api.us-east-1.ama/bar/',
'arn': 'arn:aws:iam::1234567890:role/foo',
'foobar': 'foo = bar = baz.w32'
}
resp = json.dumps(expected) + "\n"
with patch('%s._validate' % pb):
cls = TerraformRunner(self.mock_config(), 'terraform-bin')
cls.tf_version = (0, 7, 0)
with patch('%s.logger' % pbm, autospec=True) as mock_logger:
with patch('%s._run_tf' % pb, autospec=True) as mock_run:
mock_run.return_value = resp
res = cls._get_outputs()
assert res == expected
assert mock_run.mock_calls == [call(cls, 'output', cmd_args=['-json'])]
assert mock_logger.mock_calls == [
call.debug('Running: terraform output'),
call.debug('Terraform outputs: %s', expected)
]
开发者ID:AndrewFarley,项目名称:webhook2lambda2sqs,代码行数:20,代码来源:test_terraform_runner.py
示例17: test_get_outputs_pre0point7
def test_get_outputs_pre0point7(self):
resp = "base_url = https://ljgx260ix7.execute-api.us-east-1.ama/bar/\n"
resp += "arn = arn:aws:iam::1234567890:role/foo\n"
resp += "foobar = foo = bar = baz.w32\n\n"
expected = {
'base_url': 'https://ljgx260ix7.execute-api.us-east-1.ama/bar/',
'arn': 'arn:aws:iam::1234567890:role/foo',
'foobar': 'foo = bar = baz.w32'
}
with patch('%s._validate' % pb):
cls = TerraformRunner(self.mock_config(), 'terraform-bin')
cls.tf_version = (0, 6, 16)
with patch('%s.logger' % pbm, autospec=True) as mock_logger:
with patch('%s._run_tf' % pb, autospec=True) as mock_run:
mock_run.return_value = resp
res = cls._get_outputs()
assert res == expected
assert mock_run.mock_calls == [call(cls, 'output')]
assert mock_logger.mock_calls == [
call.debug('Running: terraform output'),
call.debug('Terraform outputs: %s', expected)
]
开发者ID:AndrewFarley,项目名称:webhook2lambda2sqs,代码行数:22,代码来源:test_terraform_runner.py
示例18: get_api_id
def get_api_id(config, args):
"""
Get the API ID from Terraform, or from AWS if that fails.
:param config: configuration
:type config: :py:class:`~.Config`
:param args: command line arguments
:type args: :py:class:`argparse.Namespace`
:return: API Gateway ID
:rtype: str
"""
try:
logger.debug('Trying to get Terraform rest_api_id output')
runner = TerraformRunner(config, args.tf_path)
outputs = runner._get_outputs()
depl_id = outputs['rest_api_id']
logger.debug("Terraform rest_api_id output: '%s'", depl_id)
except Exception:
logger.info('Unable to find API rest_api_id from Terraform state;'
' querying AWS.', exc_info=1)
aws = AWSInfo(config)
depl_id = aws.get_api_id()
logger.debug("AWS API ID: '%s'", depl_id)
return depl_id
开发者ID:jantman,项目名称:webhook2lambda2sqs,代码行数:24,代码来源:runner.py
示例19: main
def main(args=None):
"""
Main entry point
"""
# parse args
if args is None:
args = parse_args(sys.argv[1:])
# dump example config if that action
if args.action == 'example-config':
conf, doc = Config.example_config()
print(conf)
sys.stderr.write(doc + "\n")
return
# set logging level
if args.verbose > 1:
set_log_debug()
elif args.verbose == 1:
set_log_info()
# get our config
config = Config(args.config)
if args.action == 'logs':
aws = AWSInfo(config)
aws.show_cloudwatch_logs(count=args.log_count)
return
if args.action == 'apilogs':
api_id = get_api_id(config, args)
aws = AWSInfo(config)
aws.show_cloudwatch_logs(
count=args.log_count,
grp_name='API-Gateway-Execution-Logs_%s/%s' % (
api_id, config.stage_name
)
)
return
if args.action == 'queuepeek':
aws = AWSInfo(config)
aws.show_queue(name=args.queue_name, delete=args.queue_delete,
count=args.msg_count)
return
if args.action == 'test':
run_test(config, args)
return
if args.action in ['apply', 'genapply', 'plan', 'destroy']:
runner = TerraformRunner(config, args.tf_path)
tf_ver = runner.tf_version
else:
tf_ver = tuple(
[int(x) for x in args.tf_ver.split('.')]
)
# if generate or genapply, generate the configs
if args.action == 'generate' or args.action == 'genapply':
func_gen = LambdaFuncGenerator(config)
func_src = func_gen.generate()
# @TODO: also write func_source to disk
tf_gen = TerraformGenerator(config, tf_ver=tf_ver)
tf_gen.generate(func_src)
# if only generate, exit now
if args.action == 'generate':
return
# run the terraform action
if args.action == 'apply' or args.action == 'genapply':
runner.apply(args.stream_tf)
# conditionally set API Gateway Method settings
if config.get('api_gateway_method_settings') is not None:
aws = AWSInfo(config)
aws.set_method_settings()
elif args.action == 'plan':
runner.plan(args.stream_tf)
else: # destroy
runner.destroy(args.stream_tf)
开发者ID:jantman,项目名称:webhook2lambda2sqs,代码行数:81,代码来源:runner.py
示例20: test_args_for_remote_none
def test_args_for_remote_none(self):
conf = self.mock_config()
with patch('%s._validate' % pb):
cls = TerraformRunner(conf, 'tfpath')
assert cls._args_for_remote() is None
开发者ID:AndrewFarley,项目名称:webhook2lambda2sqs,代码行数:5,代码来源:test_terraform_runner.py
注:本文中的webhook2lambda2sqs.terraform_runner.TerraformRunner类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论