meissa/dda-python-terraform
6
3
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Merge pull request #97 from aubustou/develop

Browse source 
Support for Terraform 0.14 and Python 3.6+
This commit is contained in:
Spikeophant 2021-03-01 07:14:42 -08:00 committed by GitHub
commit e1ae747ebf
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
19 changed files with 893 additions and 884 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
.bumpversion.cfg
View file

@ -1,7 +0,0 @@
[bumpversion]
current_version = 0.10.1
commit = True
tag = False
[bumpversion:file:setup.py]

2
.gitignore vendored
View file

@ -14,6 +14,7 @@ Icon
# virtualenv
.virtualenv/
venv/
# Intellij
@ -21,3 +22,4 @@ Icon
# VSCode
.vscode/
pyrightconfig.json

27
.pre-commit-config.yaml Normal file
View file

@ -0,0 +1,27 @@
default_language_version:
python: python3.6
repos:
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v3.1.0 # v2.1.0
hooks:
- id: trailing-whitespace
- id: end-of-file-fixer
- id: check-docstring-first
- id: check-json
- id: check-merge-conflict
- id: check-toml
- id: check-yaml
- id: debug-statements
- id: requirements-txt-fixer
- repo: https://github.com/pycqa/isort
rev: 5.5.2
hooks:
- id: isort
- repo: https://github.com/lovesegfault/beautysh
rev: 6.0.1
hooks:
- id: beautysh
- repo: https://github.com/psf/black
rev: 19.10b0
hooks:
- id: black

34
.travis.yml
View file

@ -1,11 +1,12 @@
language: python
python:
- '2.7'
- '3.5'
- '3.6'
- '3.7'
- '3.8'
- '3.9'
before_install: sudo apt-get install unzip
before_script:
- export TFVER=0.10.0
- export TFVER=0.13.4
- export TFURL=https://releases.hashicorp.com/terraform/
- TFURL+=$TFVER
- TFURL+="/terraform_"
@ -16,37 +17,12 @@ before_script:
- unzip terraform_bin.zip -d tf_bin
install:
- curl https://bootstrap.pypa.io/ez_setup.py -o - | python
- pip install tox-travis
- pip install .
script:
- export PATH=$PATH:$PWD/tf_bin
- tox
- pytest -v
branches:
only:
- master
- develop
- release/**
deploy:
- provider: pypi
distributions: sdist
server: https://testpypi.python.org/pypi
user: beelit94
password:
secure: sWxc+p/gdq3k2WbUGNG2F4TukFNkTkvq6OPaiOvyfgWThYNk6/juRkMd8flmTbh0VGhcjFbpDLeSApb2kFhfiokYJSH1hOOcmXf8xzYH8/+R4DDEiGa5Y/pR9TBvYu4S8eJEfFUFfb1BBpapykj7o43hcaqMExBJIdVJU7aeoEAC1jQeTJh8wWwdJKHy2dNSM+6RVhk3e5+b0LfK7Bk5sU5P+YdEMj79MJU450J4OmZXWzJgvBN5/2QfVa5LrUD00nYuGuiBniz2lVevIHWjUYawUzpPsTa7F0s2WemG9YcV7U8u06xNjY9Ce3CTbxNhc7OIKq+TCkOgR3qZFXVJ8A87G+AT2iQ01VslQ4DJCxnJNTnpqojWnwf6MFL9O8ONioWYO32bhQFKOQ806ASHP4lNMRDKqx8hXtP5In7/r0SARKscv6Bas83rp+FESkKD5vWgkZJG+yx96LlwRLUhSVnVyb/nOJ++zt5RR3BvY2O4p9YAZY3Qt8TQihOdBQKnY3UXsMyNaE25+yvyNWpmyJiePRbTUd+cpLnycnqG9Ll8v6TpFXb6ahFMjlAFfJNQYlREfseClTHSRjZNxfsXGQCsJh6TZAq7jOB5hCk3q41eOUFWARxbyj8j59NBV8fSQrrGJJ9/VZKQeYiQlBB9KpK4PrnH84oeQ8i+VSbVr5w=
on:
branch: release/**
tags: false
condition: $TRAVIS_PYTHON_VERSION = "3.5"
- provider: pypi
distributions: sdist
user: beelit94
password:
secure: QhCiTLrBvw/Uzt3eiLEmvMP3uHnayVCETqEDA+2+Q9vFavqj0CHA76zqYonBFqnh0a3HFCRIVVt+6ynpZ10kpQ3tAObIw+pY39ZPnpAhOjSpFzzMdpIF9Bhv9A93ng2iSESAZPAOwktHzUwjFx0Zvl0lSYD9rutHgttGgdU2CajiUtwTUhCTjOAVdR2Gm+15H808vzKWnMaKflXxZt+fkt279mQTYAtz6eBWtZwIKry/uAJCSrPSWtbi50O0HsWRMXLXWH5Jn/BVjWSDSM92DssUDq0D+tQyp4M5nQXJ9EyAvEdsKNLx3cvNruznh2ohI2jmcoIjwFiS6+wrEmUiXkP86iyzCSqL/EbcOG0xUh3vbfYtMBp7jENgD405+3SEhPY4PlqUmc+HDtB7FUcHz4y7wGWJRGyQzNnjJ6Tv0Ajdz5mfJubWVIvHjcRqkxTVtUKt50o00xZ62M0ZzQkDTIHQEsZly0XeHAgSvNzWkmjt9BiBrZ9OkoWVkRpSrCBy/EcpDNPCTSfSzOQ0Nq1ePFjkkW1n8QWDW9Pdb+/7/P2y9E2S8CT+nXBkRQeQiO86Qf1Ireg7k9TA5VYisVZ6bEXEc9UV0mAojpSsC7zWhVlbAoltN6ZbjKmqy/wqn2QIcJemcSie0JigzKpdw7l8FPT2lCRyTKlYLpRyKXzSkNI=
on:
branch: master
tags: false
condition: $TRAVIS_PYTHON_VERSION = "3.5"
notifications:
email:
recipients:
- beelit94@gmail.com

13
README.md
View file

@ -3,8 +3,11 @@
python-terraform is a python module provide a wrapper of `terraform` command line tool.
`terraform` is a tool made by Hashicorp, please refer to https://terraform.io/
[![Code style: black](https://img.shields.io/badge/code%20style-black-000000.svg)](https://github.com/psf/black)
[![pre-commit](https://img.shields.io/badge/pre--commit-enabled-brightgreen?logo=pre-commit&logoColor=white)](https://github.com/pre-commit/pre-commit)
### Status
[![Build Status](https://travis-ci.org/beelit94/python-terraform.svg?branch=develop)](https://travis-ci.org/beelit94/python-terraform)
[![Build Status](https://travis-ci.org/aubustou/python-terraform.svg?branch=develop)](https://travis-ci.org/aubustou/python-terraform)
## Installation
pip install python-terraform
@ -137,11 +140,3 @@ This make api caller don't have a general rule to follow but to do
a exhaustive method implementation which I don't prefer to.
Therefore I end-up with using `IsFlagged` or `IsNotFlagged` as value of option
like `-no-color` and `True/False` value reserved for option like `refresh=true`

470
python_terraform/__init__.py
View file

@ -1,462 +1,8 @@
# -*- coding: utf-8 -*-
# above is for compatibility of python2.7.11
import subprocess
import os
import sys
import json
import logging
import tempfile
from python_terraform.tfstate import Tfstate
try: # Python 2.7+
from logging import NullHandler
except ImportError:
class NullHandler(logging.Handler):
def emit(self, record):
pass
log = logging.getLogger(__name__)
log.addHandler(NullHandler())
COMMAND_WITH_SUBCOMMANDS = ['workspace']
class IsFlagged:
pass
class IsNotFlagged:
pass
class TerraformCommandError(subprocess.CalledProcessError):
def __init__(self, ret_code, cmd, out, err):
super(TerraformCommandError, self).__init__(ret_code, cmd)
self.out = out
self.err = err
class Terraform(object):
"""
Wrapper of terraform command line tool
https://www.terraform.io/
"""
def __init__(self, working_dir=None,
targets=None,
state=None,
variables=None,
parallelism=None,
var_file=None,
terraform_bin_path=None,
is_env_vars_included=True,
):
"""
:param working_dir: the folder of the working folder, if not given,
will be current working folder
:param targets: list of target
as default value of apply/destroy/plan command
:param state: path of state file relative to working folder,
as a default value of apply/destroy/plan command
:param variables: default variables for apply/destroy/plan command,
will be override by variable passing by apply/destroy/plan method
:param parallelism: default parallelism value for apply/destroy command
:param var_file: passed as value of -var-file option,
could be string or list, list stands for multiple -var-file option
:param terraform_bin_path: binary path of terraform
:type is_env_vars_included: bool
:param is_env_vars_included: included env variables when calling terraform cmd
"""
self.is_env_vars_included = is_env_vars_included
self.working_dir = working_dir
self.state = state
self.targets = [] if targets is None else targets
self.variables = dict() if variables is None else variables
self.parallelism = parallelism
self.terraform_bin_path = terraform_bin_path \
if terraform_bin_path else 'terraform'
self.var_file = var_file
self.temp_var_files = VariableFiles()
# store the tfstate data
self.tfstate = None
self.read_state_file(self.state)
def __getattr__(self, item):
def wrapper(*args, **kwargs):
cmd_name = str(item)
if cmd_name.endswith('_cmd'):
cmd_name = cmd_name[:-4]
log.debug('called with %r and %r' % (args, kwargs))
return self.cmd(cmd_name, *args, **kwargs)
return wrapper
def apply(self, dir_or_plan=None, input=False, skip_plan=False, no_color=IsFlagged,
**kwargs):
"""
refer to https://terraform.io/docs/commands/apply.html
no-color is flagged by default
:param no_color: disable color of stdout
:param input: disable prompt for a missing variable
:param dir_or_plan: folder relative to working folder
:param skip_plan: force apply without plan (default: false)
:param kwargs: same as kwags in method 'cmd'
:returns return_code, stdout, stderr
"""
default = kwargs
default['input'] = input
default['no_color'] = no_color
default['auto-approve'] = (skip_plan == True)
option_dict = self._generate_default_options(default)
args = self._generate_default_args(dir_or_plan)
return self.cmd('apply', *args, **option_dict)
def _generate_default_args(self, dir_or_plan):
return [dir_or_plan] if dir_or_plan else []
def _generate_default_options(self, input_options):
option_dict = dict()
option_dict['state'] = self.state
option_dict['target'] = self.targets
option_dict['var'] = self.variables
option_dict['var_file'] = self.var_file
option_dict['parallelism'] = self.parallelism
option_dict['no_color'] = IsFlagged
option_dict['input'] = False
option_dict.update(input_options)
return option_dict
def destroy(self, dir_or_plan=None, force=IsFlagged, **kwargs):
"""
refer to https://www.terraform.io/docs/commands/destroy.html
force/no-color option is flagged by default
:return: ret_code, stdout, stderr
"""
default = kwargs
default['force'] = force
options = self._generate_default_options(default)
args = self._generate_default_args(dir_or_plan)
return self.cmd('destroy', *args, **options)
def plan(self, dir_or_plan=None, detailed_exitcode=IsFlagged, **kwargs):
"""
refer to https://www.terraform.io/docs/commands/plan.html
:param detailed_exitcode: Return a detailed exit code when the command exits.
:param dir_or_plan: relative path to plan/folder
:param kwargs: options
:return: ret_code, stdout, stderr
"""
options = kwargs
options['detailed_exitcode'] = detailed_exitcode
options = self._generate_default_options(options)
args = self._generate_default_args(dir_or_plan)
return self.cmd('plan', *args, **options)
def init(self, dir_or_plan=None, backend_config=None,
reconfigure=IsFlagged, backend=True, **kwargs):
"""
refer to https://www.terraform.io/docs/commands/init.html
By default, this assumes you want to use backend config, and tries to
init fresh. The flags -reconfigure and -backend=true are default.
:param dir_or_plan: relative path to the folder want to init
:param backend_config: a dictionary of backend config options. eg.
t = Terraform()
t.init(backend_config={'access_key': 'myaccesskey',
'secret_key': 'mysecretkey', 'bucket': 'mybucketname'})
:param reconfigure: whether or not to force reconfiguration of backend
:param backend: whether or not to use backend settings for init
:param kwargs: options
:return: ret_code, stdout, stderr
"""
options = kwargs
options['backend_config'] = backend_config
options['reconfigure'] = reconfigure
options['backend'] = backend
options = self._generate_default_options(options)
args = self._generate_default_args(dir_or_plan)
return self.cmd('init', *args, **options)
def generate_cmd_string(self, cmd, *args, **kwargs):
"""
for any generate_cmd_string doesn't written as public method of terraform
examples:
1. call import command,
ref to https://www.terraform.io/docs/commands/import.html
--> generate_cmd_string call:
terraform import -input=true aws_instance.foo i-abcd1234
--> python call:
tf.generate_cmd_string('import', 'aws_instance.foo', 'i-abcd1234', input=True)
2. call apply command,
--> generate_cmd_string call:
terraform apply -var='a=b' -var='c=d' -no-color the_folder
--> python call:
tf.generate_cmd_string('apply', the_folder, no_color=IsFlagged, var={'a':'b', 'c':'d'})
:param cmd: command and sub-command of terraform, seperated with space
refer to https://www.terraform.io/docs/commands/index.html
:param args: arguments of a command
:param kwargs: same as kwags in method 'cmd'
:return: string of valid terraform command
"""
cmds = cmd.split()
cmds = [self.terraform_bin_path] + cmds
if cmd in COMMAND_WITH_SUBCOMMANDS:
args = list(args)
subcommand = args.pop(0)
cmds.append(subcommand)
for option, value in kwargs.items():
if '_' in option:
option = option.replace('_', '-')
if type(value) is list:
for sub_v in value:
cmds += ['-{k}={v}'.format(k=option, v=sub_v)]
continue
if type(value) is dict:
if 'backend-config' in option:
for bk, bv in value.items():
cmds += ['-backend-config={k}={v}'.format(k=bk, v=bv)]
continue
# since map type sent in string won't work, create temp var file for
# variables, and clean it up later
elif option == 'var':
# We do not create empty var-files if there is no var passed.
# An empty var-file would result in an error: An argument or block definition is required here
if value:
filename = self.temp_var_files.create(value)
cmds += ['-var-file={0}'.format(filename)]
continue
# simple flag,
if value is IsFlagged:
cmds += ['-{k}'.format(k=option)]
continue
if value is None or value is IsNotFlagged:
continue
if type(value) is bool:
value = 'true' if value else 'false'
cmds += ['-{k}={v}'.format(k=option, v=value)]
cmds += args
return cmds
def cmd(self, cmd, *args, **kwargs):
"""
run a terraform command, if success, will try to read state file
:param cmd: command and sub-command of terraform, seperated with space
refer to https://www.terraform.io/docs/commands/index.html
:param args: arguments of a command
:param kwargs: any option flag with key value without prefixed dash character
if there's a dash in the option name, use under line instead of dash,
ex. -no-color --> no_color
if it's a simple flag with no value, value should be IsFlagged
ex. cmd('taint', allow＿missing=IsFlagged)
if it's a boolean value flag, assign True or false
if it's a flag could be used multiple times, assign list to it's value
if it's a "var" variable flag, assign dictionary to it
if a value is None, will skip this option
if the option 'capture_output' is passed (with any value other than
True), terraform output will be printed to stdout/stderr and
"None" will be returned as out and err.
if the option 'raise_on_error' is passed (with any value that evaluates to True),
and the terraform command returns a nonzerop return code, then
a TerraformCommandError exception will be raised. The exception object will
have the following properties:
returncode: The command's return code
out: The captured stdout, or None if not captured
err: The captured stderr, or None if not captured
:return: ret_code, out, err
"""
capture_output = kwargs.pop('capture_output', True)
raise_on_error = kwargs.pop('raise_on_error', False)
synchronous = kwargs.pop('synchronous', True)
if capture_output is True:
stderr = subprocess.PIPE
stdout = subprocess.PIPE
elif capture_output == "framework":
stderr = None
stdout = None
else:
stderr = sys.stderr
stdout = sys.stdout
cmds = self.generate_cmd_string(cmd, *args, **kwargs)
log.debug('command: {c}'.format(c=' '.join(cmds)))
working_folder = self.working_dir if self.working_dir else None
environ_vars = {}
if self.is_env_vars_included:
environ_vars = os.environ.copy()
p = subprocess.Popen(cmds, stdout=stdout, stderr=stderr,
cwd=working_folder, env=environ_vars)
if not synchronous:
return p, None, None
out, err = p.communicate()
ret_code = p.returncode
log.debug('output: {o}'.format(o=out))
if ret_code == 0:
self.read_state_file()
else:
log.warning('error: {e}'.format(e=err))
self.temp_var_files.clean_up()
if capture_output is True:
out = out.decode('utf-8')
err = err.decode('utf-8')
else:
out = None
err = None
if ret_code != 0 and raise_on_error:
raise TerraformCommandError(
ret_code, ' '.join(cmds), out=out, err=err)
return ret_code, out, err
def output(self, *args, **kwargs):
"""
https://www.terraform.io/docs/commands/output.html
Note that this method does not conform to the (ret_code, out, err) return convention. To use
the "output" command with the standard convention, call "output_cmd" instead of
"output".
:param args: Positional arguments. There is one optional positional
argument NAME; if supplied, the returned output text
will be the json for a single named output value.
:param kwargs: Named options, passed to the command. In addition,
'full_value': If True, and NAME is provided, then
the return value will be a dict with
"value', 'type', and 'sensitive'
properties.
:return: None, if an error occured
Output value as a string, if NAME is provided and full_value
is False or not provided
Output value as a dict with 'value', 'sensitive', and 'type' if
NAME is provided and full_value is True.
dict of named dicts each with 'value', 'sensitive', and 'type',
if NAME is not provided
"""
full_value = kwargs.pop('full_value', False)
name_provided = (len(args) > 0)
kwargs['json'] = IsFlagged
if not kwargs.get('capture_output', True) is True:
raise ValueError('capture_output is required for this method')
ret, out, err = self.output_cmd(*args, **kwargs)
if ret != 0:
return None
out = out.lstrip()
value = json.loads(out)
if name_provided and not full_value:
value = value['value']
return value
def read_state_file(self, file_path=None):
"""
read .tfstate file
:param file_path: relative path to working dir
:return: states file in dict type
"""
working_dir = self.working_dir or ''
file_path = file_path or self.state or ''
if not file_path:
backend_path = os.path.join(file_path, '.terraform',
'terraform.tfstate')
if os.path.exists(os.path.join(working_dir, backend_path)):
file_path = backend_path
else:
file_path = os.path.join(file_path, 'terraform.tfstate')
file_path = os.path.join(working_dir, file_path)
self.tfstate = Tfstate.load_file(file_path)
def set_workspace(self, workspace, *args, **kwargs):
"""
set workspace
:param workspace: the desired workspace.
:return: status
"""
return self.cmd('workspace', 'select', workspace, *args, **kwargs)
def create_workspace(self, workspace, *args, **kwargs):
"""
create workspace
:param workspace: the desired workspace.
:return: status
"""
return self.cmd('workspace', 'new', workspace, *args, **kwargs)
def delete_workspace(self, workspace, *args, **kwargs):
"""
delete workspace
:param workspace: the desired workspace.
:return: status
"""
return self.cmd('workspace', 'delete', workspace, *args, **kwargs)
def show_workspace(self, **kwargs):
"""
show workspace, this command does not need the [DIR] part
:return: workspace
"""
return self.cmd('workspace', 'show', **kwargs)
def __exit__(self, exc_type, exc_value, traceback):
self.temp_var_files.clean_up()
class VariableFiles(object):
def __init__(self):
self.files = []
def create(self, variables):
with tempfile.NamedTemporaryFile('w+t', suffix='.tfvars.json', delete=False) as temp:
log.debug('{0} is created'.format(temp.name))
self.files.append(temp)
log.debug(
'variables wrote to tempfile: {0}'.format(str(variables)))
temp.write(json.dumps(variables))
file_name = temp.name
return file_name
def clean_up(self):
for f in self.files:
os.unlink(f.name)
self.files = []
from .terraform import (
IsFlagged,
IsNotFlagged,
Terraform,
TerraformCommandError,
VariableFiles,
)
from .tfstate import Tfstate

483
python_terraform/terraform.py Normal file
View file

@ -0,0 +1,483 @@
import json
import logging
import os
import subprocess
import sys
import tempfile
from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Type, Union
from python_terraform.tfstate import Tfstate
logger = logging.getLogger(__name__)
COMMAND_WITH_SUBCOMMANDS = {"workspace"}
class TerraformFlag:
pass
class IsFlagged(TerraformFlag):
pass
class IsNotFlagged(TerraformFlag):
pass
CommandOutput = Tuple[Optional[int], Optional[str], Optional[str]]
class TerraformCommandError(subprocess.CalledProcessError):
def __init__(self, ret_code: int, cmd: str, out: Optional[str], err: Optional[str]):
super(TerraformCommandError, self).__init__(ret_code, cmd)
self.out = out
self.err = err
logger.error("Error with command %s. Reason: %s", self.cmd, self.err)
class Terraform:
"""Wrapper of terraform command line tool.
https://www.terraform.io/
"""
def __init__(
self,
working_dir: Optional[str] = None,
targets: Optional[Sequence[str]] = None,
state: Optional[str] = None,
variables: Optional[Dict[str, str]] = None,
parallelism: Optional[str] = None,
var_file: Optional[str] = None,
terraform_bin_path: Optional[str] = None,
is_env_vars_included: bool = True,
):
"""
:param working_dir: the folder of the working folder, if not given,
will be current working folder
:param targets: list of target
as default value of apply/destroy/plan command
:param state: path of state file relative to working folder,
as a default value of apply/destroy/plan command
:param variables: default variables for apply/destroy/plan command,
will be override by variable passing by apply/destroy/plan method
:param parallelism: default parallelism value for apply/destroy command
:param var_file: passed as value of -var-file option,
could be string or list, list stands for multiple -var-file option
:param terraform_bin_path: binary path of terraform
:type is_env_vars_included: bool
:param is_env_vars_included: included env variables when calling terraform cmd
"""
self.is_env_vars_included = is_env_vars_included
self.working_dir = working_dir
self.state = state
self.targets = [] if targets is None else targets
self.variables = dict() if variables is None else variables
self.parallelism = parallelism
self.terraform_bin_path = (
terraform_bin_path if terraform_bin_path else "terraform"
)
self.var_file = var_file
self.temp_var_files = VariableFiles()
# store the tfstate data
self.tfstate = None
self.read_state_file(self.state)
def __getattr__(self, item: str) -> Callable:
def wrapper(*args, **kwargs):
cmd_name = str(item)
if cmd_name.endswith("_cmd"):
cmd_name = cmd_name[:-4]
logger.debug("called with %r and %r", args, kwargs)
return self.cmd(cmd_name, *args, **kwargs)
return wrapper
def apply(
self,
dir_or_plan: Optional[str] = None,
input: bool = False,
skip_plan: bool = True,
no_color: Type[TerraformFlag] = IsFlagged,
**kwargs,
) -> CommandOutput:
"""Refer to https://terraform.io/docs/commands/apply.html
no-color is flagged by default
:param no_color: disable color of stdout
:param input: disable prompt for a missing variable
:param dir_or_plan: folder relative to working folder
:param skip_plan: force apply without plan (default: false)
:param kwargs: same as kwags in method 'cmd'
:returns return_code, stdout, stderr
"""
if not skip_plan:
return self.plan(dir_or_plan=dir_or_plan, **kwargs)
default = kwargs.copy()
default["input"] = input
default["no_color"] = no_color
default["auto-approve"] = True # a False value will require an input
option_dict = self._generate_default_options(default)
args = self._generate_default_args(dir_or_plan)
return self.cmd("apply", *args, **option_dict)
def _generate_default_args(self, dir_or_plan: Optional[str]) -> Sequence[str]:
return [dir_or_plan] if dir_or_plan else []
def _generate_default_options(
self, input_options: Dict[str, Any]
) -> Dict[str, Any]:
return {
"state": self.state,
"target": self.targets,
"var": self.variables,
"var_file": self.var_file,
"parallelism": self.parallelism,
"no_color": IsFlagged,
"input": False,
**input_options,
}
def destroy(
self,
dir_or_plan: Optional[str] = None,
force: Type[TerraformFlag] = IsFlagged,
**kwargs,
) -> CommandOutput:
"""Refer to https://www.terraform.io/docs/commands/destroy.html
force/no-color option is flagged by default
:return: ret_code, stdout, stderr
"""
default = kwargs.copy()
default["force"] = force
options = self._generate_default_options(default)
args = self._generate_default_args(dir_or_plan)
return self.cmd("destroy", *args, **options)
def plan(
self,
dir_or_plan: Optional[str] = None,
detailed_exitcode: Type[TerraformFlag] = IsFlagged,
**kwargs,
) -> CommandOutput:
"""Refer to https://www.terraform.io/docs/commands/plan.html
:param detailed_exitcode: Return a detailed exit code when the command exits.
:param dir_or_plan: relative path to plan/folder
:param kwargs: options
:return: ret_code, stdout, stderr
"""
options = kwargs.copy()
options["detailed_exitcode"] = detailed_exitcode
options = self._generate_default_options(options)
args = self._generate_default_args(dir_or_plan)
return self.cmd("plan", *args, **options)
def init(
self,
dir_or_plan: Optional[str] = None,
backend_config: Optional[Dict[str, str]] = None,
reconfigure: Type[TerraformFlag] = IsFlagged,
backend: bool = True,
**kwargs,
) -> CommandOutput:
"""Refer to https://www.terraform.io/docs/commands/init.html
By default, this assumes you want to use backend config, and tries to
init fresh. The flags -reconfigure and -backend=true are default.
:param dir_or_plan: relative path to the folder want to init
:param backend_config: a dictionary of backend config options. eg.
t = Terraform()
t.init(backend_config={'access_key': 'myaccesskey',
'secret_key': 'mysecretkey', 'bucket': 'mybucketname'})
:param reconfigure: whether or not to force reconfiguration of backend
:param backend: whether or not to use backend settings for init
:param kwargs: options
:return: ret_code, stdout, stderr
"""
options = kwargs.copy()
options.update(
{
"backend_config": backend_config,
"reconfigure": reconfigure,
"backend": backend,
}
)
options = self._generate_default_options(options)
args = self._generate_default_args(dir_or_plan)
return self.cmd("init", *args, **options)
def generate_cmd_string(self, cmd: str, *args, **kwargs) -> List[str]:
"""For any generate_cmd_string doesn't written as public method of Terraform
examples:
1. call import command,
ref to https://www.terraform.io/docs/commands/import.html
--> generate_cmd_string call:
terraform import -input=true aws_instance.foo i-abcd1234
--> python call:
tf.generate_cmd_string('import', 'aws_instance.foo', 'i-abcd1234', input=True)
2. call apply command,
--> generate_cmd_string call:
terraform apply -var='a=b' -var='c=d' -no-color the_folder
--> python call:
tf.generate_cmd_string('apply', the_folder, no_color=IsFlagged, var={'a':'b', 'c':'d'})
:param cmd: command and sub-command of terraform, seperated with space
refer to https://www.terraform.io/docs/commands/index.html
:param args: arguments of a command
:param kwargs: same as kwags in method 'cmd'
:return: string of valid terraform command
"""
cmds = cmd.split()
cmds = [self.terraform_bin_path] + cmds
if cmd in COMMAND_WITH_SUBCOMMANDS:
args = list(args)
subcommand = args.pop(0)
cmds.append(subcommand)
for option, value in kwargs.items():
if "_" in option:
option = option.replace("_", "-")
if isinstance(value, list):
for sub_v in value:
cmds += [f"-{option}={sub_v}"]
continue
if isinstance(value, dict):
if "backend-config" in option:
for bk, bv in value.items():
cmds += [f"-backend-config={bk}={bv}"]
continue
# since map type sent in string won't work, create temp var file for
# variables, and clean it up later
elif option == "var":
# We do not create empty var-files if there is no var passed.
# An empty var-file would result in an error: An argument or block definition is required here
if value:
filename = self.temp_var_files.create(value)
cmds += [f"-var-file={filename}"]
continue
# simple flag,
if value is IsFlagged:
cmds += [f"-{option}"]
continue
if value is None or value is IsNotFlagged:
continue
if isinstance(value, bool):
value = "true" if value else "false"
cmds += [f"-{option}={value}"]
cmds += args
return cmds
def cmd(
self,
cmd: str,
*args,
capture_output: Union[bool, str] = True,
raise_on_error: bool = True,
synchronous: bool = True,
**kwargs,
) -> CommandOutput:
"""Run a terraform command, if success, will try to read state file
:param cmd: command and sub-command of terraform, seperated with space
refer to https://www.terraform.io/docs/commands/index.html
:param args: arguments of a command
:param kwargs: any option flag with key value without prefixed dash character
if there's a dash in the option name, use under line instead of dash,
ex. -no-color --> no_color
if it's a simple flag with no value, value should be IsFlagged
ex. cmd('taint', allow＿missing=IsFlagged)
if it's a boolean value flag, assign True or false
if it's a flag could be used multiple times, assign list to it's value
if it's a "var" variable flag, assign dictionary to it
if a value is None, will skip this option
if the option 'capture_output' is passed (with any value other than
True), terraform output will be printed to stdout/stderr and
"None" will be returned as out and err.
if the option 'raise_on_error' is passed (with any value that evaluates to True),
and the terraform command returns a nonzerop return code, then
a TerraformCommandError exception will be raised. The exception object will
have the following properties:
returncode: The command's return code
out: The captured stdout, or None if not captured
err: The captured stderr, or None if not captured
:return: ret_code, out, err
"""
if capture_output is True:
stderr = subprocess.PIPE
stdout = subprocess.PIPE
elif capture_output == "framework":
stderr = None
stdout = None
else:
stderr = sys.stderr
stdout = sys.stdout
cmds = self.generate_cmd_string(cmd, *args, **kwargs)
logger.info("Command: %s", " ".join(cmds))
working_folder = self.working_dir if self.working_dir else None
environ_vars = {}
if self.is_env_vars_included:
environ_vars = os.environ.copy()
p = subprocess.Popen(
cmds, stdout=stdout, stderr=stderr, cwd=working_folder, env=environ_vars
)
if not synchronous:
return None, None, None
out, err = p.communicate()
ret_code = p.returncode
logger.info("output: %s", out)
if ret_code == 0:
self.read_state_file()
else:
logger.warning("error: %s", err)
self.temp_var_files.clean_up()
if capture_output is True:
out = out.decode()
err = err.decode()
else:
out = None
err = None
if ret_code and raise_on_error:
raise TerraformCommandError(ret_code, " ".join(cmds), out=out, err=err)
return ret_code, out, err
def output(
self, *args, capture_output: bool = True, **kwargs
) -> Union[None, str, Dict[str, str], Dict[str, Dict[str, str]]]:
"""Refer https://www.terraform.io/docs/commands/output.html
Note that this method does not conform to the (ret_code, out, err) return
convention. To use the "output" command with the standard convention,
call "output_cmd" instead of "output".
:param args: Positional arguments. There is one optional positional
argument NAME; if supplied, the returned output text
will be the json for a single named output value.
:param kwargs: Named options, passed to the command. In addition,
'full_value': If True, and NAME is provided, then
the return value will be a dict with
"value', 'type', and 'sensitive'
properties.
:return: None, if an error occured
Output value as a string, if NAME is provided and full_value
is False or not provided
Output value as a dict with 'value', 'sensitive', and 'type' if
NAME is provided and full_value is True.
dict of named dicts each with 'value', 'sensitive', and 'type',
if NAME is not provided
"""
kwargs["json"] = IsFlagged
if capture_output is False:
raise ValueError("capture_output is required for this method")
ret, out, _ = self.output_cmd(*args, **kwargs)
if ret:
return None
return json.loads(out.lstrip())
def read_state_file(self, file_path=None) -> None:
"""Read .tfstate file
:param file_path: relative path to working dir
:return: states file in dict type
"""
working_dir = self.working_dir or ""
file_path = file_path or self.state or ""
if not file_path:
backend_path = os.path.join(file_path, ".terraform", "terraform.tfstate")
if os.path.exists(os.path.join(working_dir, backend_path)):
file_path = backend_path
else:
file_path = os.path.join(file_path, "terraform.tfstate")
file_path = os.path.join(working_dir, file_path)
self.tfstate = Tfstate.load_file(file_path)
def set_workspace(self, workspace, *args, **kwargs) -> CommandOutput:
"""Set workspace
:param workspace: the desired workspace.
:return: status
"""
return self.cmd("workspace", "select", workspace, *args, **kwargs)
def create_workspace(self, workspace, *args, **kwargs) -> CommandOutput:
"""Create workspace
:param workspace: the desired workspace.
:return: status
"""
return self.cmd("workspace", "new", workspace, *args, **kwargs)
def delete_workspace(self, workspace, *args, **kwargs) -> CommandOutput:
"""Delete workspace
:param workspace: the desired workspace.
:return: status
"""
return self.cmd("workspace", "delete", workspace, *args, **kwargs)
def show_workspace(self, **kwargs) -> CommandOutput:
"""Show workspace, this command does not need the [DIR] part
:return: workspace
"""
return self.cmd("workspace", "show", **kwargs)
def __exit__(self, exc_type, exc_value, traceback) -> None:
self.temp_var_files.clean_up()
class VariableFiles:
def __init__(self):
self.files = []
def create(self, variables: Dict[str, str]) -> str:
with tempfile.NamedTemporaryFile(
"w+t", suffix=".tfvars.json", delete=False
) as temp:
logger.debug("%s is created", temp.name)
self.files.append(temp)
logger.debug("variables wrote to tempfile: %s", variables)
temp.write(json.dumps(variables))
file_name = temp.name
return file_name
def clean_up(self):
for f in self.files:
os.unlink(f.name)
self.files = []

25
python_terraform/tfstate.py
View file

@ -1,26 +1,25 @@
# -*- coding: utf-8 -*-
# above is for compatibility of python2.7.11
import json
import os
import logging
import os
from typing import Dict, Optional
log = logging.getLogger(__name__)
logger = logging.getLogger(__name__)
class Tfstate(object):
def __init__(self, data=None):
self.tfstate_file = None
class Tfstate:
def __init__(self, data: Optional[Dict[str, str]] = None):
self.tfstate_file: Optional[str] = None
self.native_data = data
if data:
self.__dict__ = data
@staticmethod
def load_file(file_path):
def load_file(file_path: str) -> "Tfstate":
"""Read the tfstate file and load its contents.
Parses then as JSON and put the result into the object.
"""
Read the tfstate file and load its contents, parses then as JSON and put the result into the object
"""
log.debug('read data from {0}'.format(file_path))
logger.debug("read data from %s", file_path)
if os.path.exists(file_path):
with open(file_path) as f:
json_data = json.load(f)
@ -29,6 +28,6 @@ class Tfstate(object):
tf_state.tfstate_file = file_path
return tf_state
log.debug('{0} is not exist'.format(file_path))
logger.debug("%s does not exist", file_path)
return Tfstate()

4
requirements.txt
View file

@ -1,3 +1 @@
tox-pyenv
pytest
tox
/bin/bash: q : commande introuvable

8
setup.cfg
View file

@ -1,2 +1,10 @@
[wheel]
universal = 1
[isort]
line_length=88
known_third_party=
indent=' '
multi_line_output=3
sections=FUTURE,STDLIB,THIRDPARTY,FIRSTPARTY,LOCALFOLDER
include_trailing_comma=true

48
setup.py
View file

@ -7,12 +7,13 @@ except ImportError:
from distutils.core import setup
dependencies = []
module_name = 'python-terraform'
short_description = 'This is a python module provide a wrapper ' \
'of terraform command line tool'
module_name = "python-terraform"
short_description = (
"This is a python module provide a wrapper " "of terraform command line tool"
)
try:
with open('DESCRIPTION.rst') as f:
with open("DESCRIPTION.rst") as f:
long_description = f.read()
except IOError:
long_description = short_description
@ -20,36 +21,37 @@ except IOError:
setup(
name=module_name,
version='0.10.2',
url='https://github.com/beelit94/python-terraform',
license='MIT',
author='Freddy Tan',
author_email='beelit94@gmail.com',
version="0.14.0",
url="https://github.com/beelit94/python-terraform",
license="MIT",
author="Freddy Tan",
author_email="beelit94@gmail.com",
description=short_description,
long_description=long_description,
packages=['python_terraform'],
packages=["python_terraform"],
package_data={},
platforms='any',
platforms="any",
install_requires=dependencies,
tests_require=["pytest"],
python_requires=">=3.6",
classifiers=[
# As from http://pypi.python.org/pypi?%3Aaction=list_classifiers
# 'Development Status :: 1 - Planning',
# 'Development Status :: 2 - Pre-Alpha',
# 'Development Status :: 3 - Alpha',
'Development Status :: 4 - Beta',
"Development Status :: 4 - Beta",
# 'Development Status :: 5 - Production/Stable',
# 'Development Status :: 6 - Mature',
# 'Development Status :: 7 - Inactive',
'Environment :: Console',
'Intended Audience :: Developers',
'License :: OSI Approved :: MIT License',
'Operating System :: POSIX',
'Operating System :: MacOS',
'Operating System :: Unix',
"Environment :: Console",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Operating System :: POSIX",
"Operating System :: MacOS",
"Operating System :: Unix",
# 'Operating System :: Windows',
'Programming Language :: Python',
'Programming Language :: Python :: 2',
'Programming Language :: Python :: 3',
'Topic :: Software Development :: Libraries :: Python Modules',
]
"Programming Language :: Python",
"Programming Language :: Python :: 3",
"Topic :: Software Development :: Libraries :: Python Modules",
],
)

10
test/bad_fmt/test.tf
View file

@ -5,12 +5,12 @@ variable "test_var" {
provider "archive" {}
variable "test_list_var" {
type = "list"
type = list(string)
default = ["a", "b"]
}
variable "test_map_var" {
type = "map"
type = map
default = {
"a" = "a"
@ -19,13 +19,13 @@ variable "test_map_var" {
}
output "test_output" {
value = "${var.test_var}"
value = var.test_var
}
output "test_list_output" {
value = "${var.test_list_var}"
value = var.test_list_var
}
output "test_map_output" {
value = "${var.test_map_var}"
value = var.test_map_var
}

516
test/test_terraform.py
View file

@ -1,15 +1,16 @@
try:
from cStringIO import StringIO # Python 2
except ImportError:
from io import StringIO
from python_terraform import *
from contextlib import contextmanager
import pytest
import os
import fnmatch
import logging
import os
import re
import shutil
import fnmatch
from contextlib import contextmanager
from io import StringIO
from typing import Callable
import pytest
from _pytest.logging import LogCaptureFixture, caplog
from python_terraform import IsFlagged, IsNotFlagged, Terraform, TerraformCommandError
logging.basicConfig(level=logging.DEBUG)
root_logger = logging.getLogger()
@ -19,77 +20,91 @@ current_path = os.path.dirname(os.path.realpath(__file__))
FILE_PATH_WITH_SPACE_AND_SPACIAL_CHARS = "test 'test.out!"
STRING_CASES = [
[
lambda x: x.generate_cmd_string('apply', 'the_folder',
no_color=IsFlagged),
"terraform apply -no-color the_folder"
lambda x: x.generate_cmd_string("apply", "the_folder", no_color=IsFlagged),
"terraform apply -no-color the_folder",
],
[
lambda x: x.generate_cmd_string('push', 'path', vcs=True,
token='token',
atlas_address='url'),
"terraform push -vcs=true -token=token -atlas-address=url path"
lambda x: x.generate_cmd_string(
"push", "path", vcs=True, token="token", atlas_address="url"
),
"terraform push -vcs=true -token=token -atlas-address=url path",
],
]
CMD_CASES = [
['method', 'expected_output', 'expected_ret_code', 'expected_exception', 'expected_logs', 'folder'],
[
"method",
"expected_output",
"expected_ret_code",
"expected_exception",
"expected_logs",
"folder",
],
[
[
lambda x: x.cmd('plan', 'var_to_output', no_color=IsFlagged, var={'test_var': 'test'}),
lambda x: x.cmd(
"plan",
"var_to_output",
no_color=IsFlagged,
var={"test_var": "test"},
raise_on_error=False,
),
# Expected output varies by terraform version
["doesn't need to do anything", # Terraform < 0.10.7 (used in travis env)
"no\nactions need to be performed"], # Terraform >= 0.10.7
"Plan: 0 to add, 0 to change, 0 to destroy.",
0,
False,
'',
'var_to_output'
"",
"var_to_output",
],
# try import aws instance
[
lambda x: x.cmd('import', 'aws_instance.foo', 'i-abcd1234', no_color=IsFlagged),
'',
lambda x: x.cmd(
"import",
"aws_instance.foo",
"i-abcd1234",
no_color=IsFlagged,
raise_on_error=False,
),
"",
1,
False,
'command: terraform import -no-color aws_instance.foo i-abcd1234',
''
],
# try import aws instance with raise_on_error
[
lambda x: x.cmd('import', 'aws_instance.foo', 'i-abcd1234', no_color=IsFlagged, raise_on_error=True),
'',
1,
True,
'command: terraform import -no-color aws_instance.foo i-abcd1234',
''
"Error: No Terraform configuration files",
"",
],
# test with space and special character in file path
[
lambda x: x.cmd('plan', 'var_to_output', out=FILE_PATH_WITH_SPACE_AND_SPACIAL_CHARS),
'',
lambda x: x.cmd(
"plan",
"var_to_output",
out=FILE_PATH_WITH_SPACE_AND_SPACIAL_CHARS,
raise_on_error=False,
),
"",
0,
False,
'',
'var_to_output'
"",
"var_to_output",
],
# test workspace command (commands with subcommand)
[
lambda x: x.cmd('workspace', 'show', no_color=IsFlagged),
'',
lambda x: x.cmd(
"workspace", "show", no_color=IsFlagged, raise_on_error=False
),
"",
0,
False,
'command: terraform workspace show -no-color',
''
"Command: terraform workspace show -no-color",
"",
],
],
]
]
@pytest.fixture(scope='function')
@pytest.fixture(scope="function")
def fmt_test_file(request):
target = os.path.join(current_path, 'bad_fmt', 'test.backup')
orgin = os.path.join(current_path, 'bad_fmt', 'test.tf')
shutil.copy(orgin,
target)
target = os.path.join(current_path, "bad_fmt", "test.backup")
orgin = os.path.join(current_path, "bad_fmt", "test.tf")
shutil.copy(orgin, target)
def td():
shutil.move(target, orgin)
@ -98,28 +113,28 @@ def fmt_test_file(request):
return
@pytest.fixture()
def string_logger(request):
log_stream = StringIO()
handler = logging.StreamHandler(log_stream)
root_logger.addHandler(handler)
# @pytest.fixture()
# def string_logger(request) -> Callable[..., str]:
# log_stream = StringIO()
# handler = logging.StreamHandler(log_stream)
# root_logger.addHandler(handler)
def td():
root_logger.removeHandler(handler)
log_stream.close()
# def td():
# root_logger.removeHandler(handler)
# log_stream.close()
request.addfinalizer(td)
return lambda: str(log_stream.getvalue())
# request.addfinalizer(td)
# return lambda: str(log_stream.getvalue())
@pytest.fixture()
def workspace_setup_teardown():
"""
Fixture used in workspace related tests
"""Fixture used in workspace related tests.
Create and tear down a workspace
*Use as a contextmanager*
"""
@contextmanager
def wrapper(workspace_name, create=True, delete=True, *args, **kwargs):
tf = Terraform(working_dir=current_path)
@ -128,22 +143,18 @@ def workspace_setup_teardown():
tf.create_workspace(workspace_name, *args, **kwargs)
yield tf
if delete:
tf.set_workspace('default')
tf.set_workspace("default")
tf.delete_workspace(workspace_name)
yield wrapper
class TestTerraform(object):
def teardown_method(self, method):
""" teardown any state that was previously setup with a setup_method
call.
"""
exclude = ['test_tfstate_file',
'test_tfstate_file2',
'test_tfstate_file3']
class TestTerraform:
def teardown_method(self, _) -> None:
"""Teardown any state that was previously setup with a setup_method call."""
exclude = ["test_tfstate_file", "test_tfstate_file2", "test_tfstate_file3"]
def purge(dir, pattern):
def purge(dir: str, pattern: str) -> None:
for root, dirnames, filenames in os.walk(dir):
dirnames[:] = [d for d in dirnames if d not in exclude]
for filename in fnmatch.filter(filenames, pattern):
@ -153,15 +164,13 @@ class TestTerraform(object):
d = os.path.join(root, dirname)
shutil.rmtree(d)
purge('.', '*.tfstate')
purge('.', '*.tfstate.backup')
purge('.', '*.terraform')
purge('.', FILE_PATH_WITH_SPACE_AND_SPACIAL_CHARS)
purge(".", "*.tfstate")
purge(".", "*.tfstate.backup")
purge(".", "*.terraform")
purge(".", FILE_PATH_WITH_SPACE_AND_SPACIAL_CHARS)
@pytest.mark.parametrize([
"method", "expected"
], STRING_CASES)
def test_generate_cmd_string(self, method, expected):
@pytest.mark.parametrize(["method", "expected"], STRING_CASES)
def test_generate_cmd_string(self, method: Callable[..., str], expected: str):
tf = Terraform(working_dir=current_path)
result = method(tf)
@ -170,288 +179,271 @@ class TestTerraform(object):
assert s in result
@pytest.mark.parametrize(*CMD_CASES)
def test_cmd(self, method, expected_output, expected_ret_code, expected_exception, expected_logs, string_logger, folder):
def test_cmd(
self,
method: Callable[..., str],
expected_output: str,
expected_ret_code: int,
expected_exception: bool,
expected_logs: str,
caplog: LogCaptureFixture,
folder: str,
):
with caplog.at_level(logging.INFO):
tf = Terraform(working_dir=current_path)
tf.init(folder)
try:
ret, out, err = method(tf)
ret, out, _ = method(tf)
assert not expected_exception
except TerraformCommandError as e:
assert expected_exception
ret = e.returncode
out = e.out
err = e.err
logs = string_logger()
logs = logs.replace('\n', '')
if isinstance(expected_output, list):
ok = False
for xo in expected_output:
if xo in out:
ok = True
break
if not ok:
assert expected_output[0] in out
else:
assert expected_output in out
assert expected_ret_code == ret
assert expected_logs in logs
assert expected_logs in caplog.text
@pytest.mark.parametrize(
("folder", "variables", "var_files", "expected_output", "options"),
[
("var_to_output",
{'test_var': 'test'}, None, "test_output=test", {}),
("var_to_output", {'test_list_var': ['c', 'd']}, None, "test_list_output=[c,d]", {}),
("var_to_output", {'test_map_var': {"c": "c", "d": "d"}}, None, "test_map_output={a=ab=bc=cd=d}", {}),
("var_to_output", {'test_map_var': {"c": "c", "d": "d"}}, 'var_to_output/test_map_var.json', "test_map_output={a=ab=bc=cd=de=ef=f}", {}),
("var_to_output", {}, None, "\x1b[0m\x1b[1m\x1b[32mApplycomplete!", {"no_color": IsNotFlagged})
])
("var_to_output", {"test_var": "test"}, None, "test_output=test", {}),
(
"var_to_output",
{"test_list_var": ["c", "d"]},
None,
'test_list_output=["c","d",]',
{},
),
(
"var_to_output",
{"test_map_var": {"c": "c", "d": "d"}},
None,
'test_map_output={"c"="c""d"="d"}',
{},
),
(
"var_to_output",
{"test_map_var": {"c": "c", "d": "d"}},
"var_to_output/test_map_var.json",
# Values are overriden
'test_map_output={"e"="e""f"="f"}',
{},
),
(
"var_to_output",
{},
None,
"\x1b[0m\x1b[1m\x1b[32mApplycomplete!",
{"no_color": IsNotFlagged},
),
],
)
def test_apply(self, folder, variables, var_files, expected_output, options):
tf = Terraform(working_dir=current_path, variables=variables, var_file=var_files)
# after 0.10.0 we always need to init
tf = Terraform(
working_dir=current_path, variables=variables, var_file=var_files
)
tf.init(folder)
ret, out, err = tf.apply(folder, **options)
assert ret == 0
assert expected_output in out.replace('\n', '').replace(' ', '')
assert err == ''
assert expected_output in out.replace("\n", "").replace(" ", "")
assert err == ""
def test_apply_with_var_file(self, string_logger):
def test_apply_with_var_file(self, caplog: LogCaptureFixture):
with caplog.at_level(logging.INFO):
tf = Terraform(working_dir=current_path)
tf.init()
tf.apply(var_file=os.path.join(current_path, 'tfvar_file', 'test.tfvars'))
logs = string_logger()
logs = logs.split('\n')
for log in logs:
if log.startswith('command: terraform apply'):
assert log.count('-var-file=') == 1
folder = "var_to_output"
tf.init(folder)
tf.apply(
folder,
var_file=os.path.join(current_path, "tfvar_files", "test.tfvars"),
)
for log in caplog.messages:
if log.startswith("Command: terraform apply"):
assert log.count("-var-file=") == 1
@pytest.mark.parametrize(
['cmd', 'args', 'options'],
["cmd", "args", "options"],
[
# bool value
('fmt', ['bad_fmt'], {'list': False, 'diff': False})
]
("fmt", ["bad_fmt"], {"list": False, "diff": False})
],
)
def test_options(self, cmd, args, options, fmt_test_file):
tf = Terraform(working_dir=current_path)
ret, out, err = getattr(tf, cmd)(*args, **options)
assert ret == 0
assert out == ''
assert out == ""
def test_state_data(self):
cwd = os.path.join(current_path, 'test_tfstate_file')
tf = Terraform(working_dir=cwd, state='tfstate.test')
cwd = os.path.join(current_path, "test_tfstate_file")
tf = Terraform(working_dir=cwd, state="tfstate.test")
tf.read_state_file()
assert tf.tfstate.modules[0]['path'] == ['root']
assert tf.tfstate.modules[0]["path"] == ["root"]
def test_state_default(self):
cwd = os.path.join(current_path, 'test_tfstate_file2')
cwd = os.path.join(current_path, "test_tfstate_file2")
tf = Terraform(working_dir=cwd)
tf.read_state_file()
assert tf.tfstate.modules[0]['path'] == ['default']
assert tf.tfstate.modules[0]["path"] == ["default"]
def test_state_default_backend(self):
cwd = os.path.join(current_path, 'test_tfstate_file3')
cwd = os.path.join(current_path, "test_tfstate_file3")
tf = Terraform(working_dir=cwd)
tf.read_state_file()
assert tf.tfstate.modules[0]['path'] == ['default_backend']
assert tf.tfstate.modules[0]["path"] == ["default_backend"]
def test_pre_load_state_data(self):
cwd = os.path.join(current_path, 'test_tfstate_file')
tf = Terraform(working_dir=cwd, state='tfstate.test')
assert tf.tfstate.modules[0]['path'] == ['root']
cwd = os.path.join(current_path, "test_tfstate_file")
tf = Terraform(working_dir=cwd, state="tfstate.test")
assert tf.tfstate.modules[0]["path"] == ["root"]
@pytest.mark.parametrize(
("folder", 'variables'),
[
("var_to_output", {'test_var': 'test'})
]
("folder", "variables"), [("var_to_output", {"test_var": "test"})]
)
def test_override_default(self, folder, variables):
tf = Terraform(working_dir=current_path, variables=variables)
tf.init(folder)
ret, out, err = tf.apply(folder, var={'test_var': 'test2'},
no_color=IsNotFlagged)
out = out.replace('\n', '')
assert '\x1b[0m\x1b[1m\x1b[32mApply' in out
out = tf.output('test_output')
assert 'test2' in out
@pytest.mark.parametrize(
("param"),
[
({}),
({'module': 'test2'}),
]
ret, out, err = tf.apply(
folder, var={"test_var": "test2"}, no_color=IsNotFlagged,
)
def test_output(self, param, string_logger):
tf = Terraform(working_dir=current_path, variables={'test_var': 'test'})
tf.init('var_to_output')
tf.apply('var_to_output')
result = tf.output('test_output', **param)
regex = re.compile("terraform output (-module=test2 -json|-json -module=test2) test_output")
log_str = string_logger()
if param:
assert re.search(regex, log_str), log_str
else:
assert result == 'test'
out = out.replace("\n", "")
assert "\x1b[0m\x1b[1m\x1b[32mApply" in out
out = tf.output("test_output")
assert "test2" in out
@pytest.mark.parametrize(
("param"),
[
({}),
({'module': 'test2'}),
]
@pytest.mark.parametrize("output_all", [True, False])
def test_output(self, caplog: LogCaptureFixture, output_all: bool):
expected_value = "test"
required_output = "test_output"
with caplog.at_level(logging.INFO):
tf = Terraform(
working_dir=current_path, variables={"test_var": expected_value}
)
def test_output_full_value(self, param, string_logger):
tf = Terraform(working_dir=current_path, variables={'test_var': 'test'})
tf.init('var_to_output')
tf.apply('var_to_output')
result = tf.output('test_output', **dict(param, full_value=True))
regex = re.compile("terraform output (-module=test2 -json|-json -module=test2) test_output")
log_str = string_logger()
if param:
assert re.search(regex, log_str), log_str
tf.init("var_to_output")
tf.apply("var_to_output")
params = tuple() if output_all else (required_output,)
result = tf.output(*params)
if output_all:
assert result[required_output]["value"] == expected_value
else:
assert result['value'] == 'test'
@pytest.mark.parametrize(
("param"),
[
({}),
({'module': 'test2'}),
]
)
def test_output_all(self, param, string_logger):
tf = Terraform(working_dir=current_path, variables={'test_var': 'test'})
tf.init('var_to_output')
tf.apply('var_to_output')
result = tf.output(**param)
regex = re.compile("terraform output (-module=test2 -json|-json -module=test2)")
log_str = string_logger()
if param:
assert re.search(regex, log_str), log_str
else:
assert result['test_output']['value'] == 'test'
assert result == expected_value
assert expected_value in caplog.messages[-1]
def test_destroy(self):
tf = Terraform(working_dir=current_path, variables={'test_var': 'test'})
tf.init('var_to_output')
ret, out, err = tf.destroy('var_to_output')
tf = Terraform(working_dir=current_path, variables={"test_var": "test"})
tf.init("var_to_output")
ret, out, err = tf.destroy("var_to_output")
assert ret == 0
assert 'Destroy complete! Resources: 0 destroyed.' in out
assert "Destroy complete! Resources: 0 destroyed." in out
@pytest.mark.parametrize(
("plan", "variables", "expected_ret"),
[
('vars_require_input', {}, 1)
]
("plan", "variables", "expected_ret"), [("vars_require_input", {}, 1)]
)
def test_plan(self, plan, variables, expected_ret):
tf = Terraform(working_dir=current_path, variables=variables)
ret, out, err = tf.plan(plan)
assert ret == expected_ret
tf.init(plan)
with pytest.raises(TerraformCommandError) as e:
tf.plan(plan)
assert (
e.value.err
== """\nError: Missing required argument\n\nThe argument "region" is required, but was not set.\n\n"""
)
def test_fmt(self, fmt_test_file):
tf = Terraform(working_dir=current_path, variables={'test_var': 'test'})
tf = Terraform(working_dir=current_path, variables={"test_var": "test"})
ret, out, err = tf.fmt(diff=True)
assert ret == 0
def test_import(self, string_logger):
tf = Terraform(working_dir=current_path)
tf.import_cmd('aws_instance.foo', 'i-abc1234', no_color=IsFlagged)
assert 'command: terraform import -no-color aws_instance.foo i-abc1234' in string_logger()
def test_create_workspace(self, workspace_setup_teardown):
workspace_name = 'test'
workspace_name = "test"
with workspace_setup_teardown(workspace_name, create=False) as tf:
ret, out, err = tf.create_workspace('test')
ret, out, err = tf.create_workspace("test")
assert ret == 0
assert err == ''
assert err == ""
def test_create_workspace_with_args(
self, workspace_setup_teardown, string_logger
):
workspace_name = 'test'
state_file_path = os.path.join(current_path, 'test_tfstate_file2', 'terraform.tfstate')
with workspace_setup_teardown(workspace_name, create=False) as tf:
ret, out, err = tf.create_workspace('test', current_path, no_color=IsFlagged)
def test_create_workspace_with_args(self, workspace_setup_teardown, caplog):
workspace_name = "test"
state_file_path = os.path.join(
current_path, "test_tfstate_file2", "terraform.tfstate"
)
with workspace_setup_teardown(
workspace_name, create=False
) as tf, caplog.at_level(logging.INFO):
ret, out, err = tf.create_workspace(
"test", current_path, no_color=IsFlagged
)
assert ret == 0
assert err == ''
logs = string_logger()
logs = logs.replace('\n', '')
expected_log = 'command: terraform workspace new -no-color test {}'.format(current_path)
assert expected_log in logs
assert err == ""
assert (
f"Command: terraform workspace new -no-color test {current_path}"
in caplog.messages
)
def test_set_workspace(self, workspace_setup_teardown):
workspace_name = 'test'
workspace_name = "test"
with workspace_setup_teardown(workspace_name) as tf:
ret, out, err = tf.set_workspace(workspace_name)
assert ret == 0
assert err == ''
assert err == ""
def test_set_workspace_with_args(
self, workspace_setup_teardown, string_logger):
workspace_name = 'test'
with workspace_setup_teardown(workspace_name) as tf:
ret, out, err = tf.set_workspace(workspace_name, current_path, no_color=IsFlagged)
def test_set_workspace_with_args(self, workspace_setup_teardown, caplog):
workspace_name = "test"
with workspace_setup_teardown(workspace_name) as tf, caplog.at_level(
logging.INFO
):
ret, out, err = tf.set_workspace(
workspace_name, current_path, no_color=IsFlagged
)
assert ret == 0
assert err == ''
logs = string_logger()
logs = logs.replace('\n', '')
expected_log = 'command: terraform workspace select -no-color test {}'.format(current_path)
assert expected_log in logs
assert err == ""
assert (
f"Command: terraform workspace select -no-color test {current_path}"
in caplog.messages
)
def test_show_workspace(self, workspace_setup_teardown):
workspace_name = 'test'
workspace_name = "test"
with workspace_setup_teardown(workspace_name) as tf:
ret, out, err = tf.show_workspace()
assert ret == 0
assert err == ''
assert err == ""
def test_show_workspace_with_no_color(
self, workspace_setup_teardown, string_logger
def test_show_workspace_with_no_color(self, workspace_setup_teardown, caplog):
workspace_name = "test"
with workspace_setup_teardown(workspace_name) as tf, caplog.at_level(
logging.INFO
):
workspace_name = 'test'
with workspace_setup_teardown(workspace_name) as tf:
ret, out, err = tf.show_workspace(no_color=IsFlagged)
assert ret == 0
assert err == ''
logs = string_logger()
logs = logs.replace('\n', '')
expected_log = 'command: terraform workspace show -no-color'
assert expected_log in logs
assert err == ""
assert "Command: terraform workspace show -no-color" in caplog.messages
def test_delete_workspace(self, workspace_setup_teardown):
workspace_name = 'test'
workspace_name = "test"
with workspace_setup_teardown(workspace_name, delete=False) as tf:
tf.set_workspace('default')
tf.set_workspace("default")
ret, out, err = tf.delete_workspace(workspace_name)
assert ret == 0
assert err == ''
assert err == ""
def test_delete_workspace_with_args(
self, workspace_setup_teardown, string_logger
):
workspace_name = 'test'
with workspace_setup_teardown(workspace_name, delete=False) as tf:
tf.set_workspace('default')
def test_delete_workspace_with_args(self, workspace_setup_teardown, caplog):
workspace_name = "test"
with workspace_setup_teardown(
workspace_name, delete=False
) as tf, caplog.at_level(logging.INFO):
tf.set_workspace("default")
ret, out, err = tf.delete_workspace(
workspace_name, current_path, force=IsFlagged,
)
assert ret == 0
assert err == ''
logs = string_logger()
logs = logs.replace('\n', '')
expected_log = 'command: terraform workspace delete -force test {}'.format(current_path)
assert expected_log in logs
assert err == ""
assert (
f"Command: terraform workspace delete -force test {current_path}"
in caplog.messages
)

10
test/var_to_output/test.tf
View file

@ -5,12 +5,12 @@ variable "test_var" {
provider "archive" {}
variable "test_list_var" {
type = "list"
type = list(string)
default = ["a", "b"]
}
variable "test_map_var" {
type = "map"
type = map
default = {
"a" = "a"
@ -19,13 +19,13 @@ variable "test_map_var" {
}
output "test_output" {
value = "${var.test_var}"
value = var.test_var
}
output "test_list_output" {
value = "${var.test_list_var}"
value = var.test_list_var
}
output "test_map_output" {
value = "${var.test_map_var}"
value = var.test_map_var
}

12
tox.ini
View file

@ -1,12 +0,0 @@
# content of: tox.ini , put in same dir as setup.py
[tox]
envlist = py27, py35, py36
[testenv]
deps=pytest
commands=py.test test
[travis]
python =
2.7: py27
3.5: py35
3.6: py36