diff --git a/.bumpversion.cfg b/.bumpversion.cfg deleted file mode 100644 index 1c1f0f9..0000000 --- a/.bumpversion.cfg +++ /dev/null @@ -1,7 +0,0 @@ -[bumpversion] -current_version = 0.10.1 -commit = True -tag = False - -[bumpversion:file:setup.py] - diff --git a/.gitignore b/.gitignore index a99ba67..d330498 100644 --- a/.gitignore +++ b/.gitignore @@ -14,6 +14,7 @@ Icon # virtualenv .virtualenv/ +venv/ # Intellij @@ -21,3 +22,4 @@ Icon # VSCode .vscode/ +pyrightconfig.json diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml new file mode 100644 index 0000000..4c7fa54 --- /dev/null +++ b/.pre-commit-config.yaml @@ -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 diff --git a/.travis.yml b/.travis.yml index bb62e9c..7ab948e 100644 --- a/.travis.yml +++ b/.travis.yml @@ -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 diff --git a/CHANGELOG.md b/CHANGELOG.md index f7fa6ec..9020c8f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,4 +16,4 @@ ## [0.10.1] 1. [#48] adding extension for temp file to adopt the change in terraform 0.12.0 -1. [#49] add workspace support \ No newline at end of file +1. [#49] add workspace support diff --git a/DESCRIPTION.rst b/DESCRIPTION.rst index f9deb59..f94e562 100644 --- a/DESCRIPTION.rst +++ b/DESCRIPTION.rst @@ -1,3 +1,3 @@ Please see README at github_ -.. _github: https://github.com/beelit94/python-terraform/blob/master/README.md \ No newline at end of file +.. _github: https://github.com/beelit94/python-terraform/blob/master/README.md diff --git a/LICENSE.txt b/LICENSE.txt index 70afd37..5cb3228 100644 --- a/LICENSE.txt +++ b/LICENSE.txt @@ -4,4 +4,4 @@ Permission is hereby granted, free of charge, to any person obtaining a copy of The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. -THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. \ No newline at end of file +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. diff --git a/README.md b/README.md index 1540092..56dee7a 100644 --- a/README.md +++ b/README.md @@ -1,14 +1,17 @@ ## Introduction -python-terraform is a python module provide a wrapper of `terraform` command line tool. +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 - + ## Usage #### For any terraform command @@ -29,44 +32,44 @@ or just call cmd method directly from python_terraform import * t = Terraform() return_code, stdout, stderr = t.cmd(, *arguments, **options) - + #### For any argument simply pass the string to arguments of the method, for example, - terraform apply target_dir + terraform apply target_dir --> .apply('target_dir') - terraform import aws_instance.foo i-abcd1234 + terraform import aws_instance.foo i-abcd1234 --> .import('aws_instance.foo', 'i-abcd1234') #### For any options - + * dash to underscore remove first dash, and then use underscore to replace dash symbol as option name - + ex. -no-color --> no_color * for a simple flag option - use ```IsFlagged/None``` as value for raising/not raising flag, for example, - - terraform taint -allow-missing + use ```IsFlagged/None``` as value for raising/not raising flag, for example, + + terraform taint -allow-missing --> .taint(allow＿missing=IsFlagged) - terraform taint + terraform taint --> .taint(allow＿missing=None) or .taint() terraform apply -no-color --> .apply(no_color=IsFlagged) - + * for a boolean value option - + assign True or False, for example, - + terraform apply -refresh=true --> .apply(refresh=True) - + * if a flag could be used multiple times, assign a list to it's value - + terraform apply -target=aws_instance.foo[1] -target=aws_instance.foo[2] - ---> + ---> .apply(target=['aws_instance.foo[1]', 'aws_instance.foo[2]']) * for the "var" flag, assign dictionary to it @@ -84,19 +87,19 @@ By default, stdout and stderr are captured and returned. This causes the applica return_code, stdout, stderr = t.(capture_output=False) ## Examples -### Have a test.tf file under folder "/home/test" +### Have a test.tf file under folder "/home/test" #### 1. apply with variables a=b, c=d, refresh=false, no color in the output In shell: cd /home/test terraform apply -var='a=b' -var='c=d' -refresh=false -no-color - + In python-terraform: from python_terraform import * tf = Terraform(working_dir='/home/test') tf.apply(no_color=IsFlagged, refresh=False, var={'a':'b', 'c':'d'}) - + or from python_terraform import * @@ -108,40 +111,32 @@ or from python_terraform import * tf = Terraform(working_dir='/home/test', variables={'a':'b', 'c':'d'}) tf.apply(no_color=IsFlagged, refresh=False) - + #### 2. fmt command, diff=true In shell: cd /home/test - terraform fmt -diff=true - + terraform fmt -diff=true + In python-terraform: - + from python_terraform import * tf = terraform(working_dir='/home/test') tf.fmt(diff=True) - + ## default values -for apply/plan/destroy command, assign with following default value to make +for apply/plan/destroy command, assign with following default value to make caller easier in python 1. ```input=False```, in this case process won't hang because you missing a variable 1. ```no_color=IsFlagged```, in this case, stdout of result is easier for parsing ## Implementation -IMHO, how terraform design boolean options is confusing. +IMHO, how terraform design boolean options is confusing. Take `input=True` and `-no-color` option of `apply` command for example, -they're all boolean value but with different option type. -This make api caller don't have a general rule to follow but to do +they're all boolean value but with different option type. +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 +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` - - - - - - - - diff --git a/python_terraform/__init__.py b/python_terraform/__init__.py index 33ab114..c2e2c10 100644 --- a/python_terraform/__init__.py +++ b/python_terraform/__init__.py @@ -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 diff --git a/python_terraform/terraform.py b/python_terraform/terraform.py new file mode 100644 index 0000000..8117da1 --- /dev/null +++ b/python_terraform/terraform.py @@ -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 = [] diff --git a/python_terraform/tfstate.py b/python_terraform/tfstate.py index 497e757..f39af0f 100644 --- a/python_terraform/tfstate.py +++ b/python_terraform/tfstate.py @@ -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): - """ - Read the tfstate file and load its contents, parses then as JSON and put the result into the object + 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. """ - 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() \ No newline at end of file + return Tfstate() diff --git a/requirements.txt b/requirements.txt index 0740785..398f64e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1 @@ -tox-pyenv -pytest -tox \ No newline at end of file +/bin/bash: q : commande introuvable diff --git a/setup.cfg b/setup.cfg index 5e40900..0710cda 100644 --- a/setup.cfg +++ b/setup.cfg @@ -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 diff --git a/setup.py b/setup.py index 2659f1e..f3bace4 100644 --- a/setup.py +++ b/setup.py @@ -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", + ], ) diff --git a/test/bad_fmt/test.tf b/test/bad_fmt/test.tf index b756909..09b9e4d 100644 --- a/test/bad_fmt/test.tf +++ b/test/bad_fmt/test.tf @@ -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 } diff --git a/test/test_terraform.py b/test/test_terraform.py index b8a79a5..221f6d3 100644 --- a/test/test_terraform.py +++ b/test/test_terraform.py @@ -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() @@ -18,78 +19,92 @@ 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('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("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", + ], +] 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): - tf = Terraform(working_dir=current_path) - tf.init(folder) - try: - ret, out, err = 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 + 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, _ = method(tf) + assert not expected_exception + except TerraformCommandError as e: + assert expected_exception + ret = e.returncode + out = e.out + + 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 == '' - - def test_apply_with_var_file(self, string_logger): - 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 + assert expected_output in out.replace("\n", "").replace(" ", "") + assert err == "" + + def test_apply_with_var_file(self, caplog: LogCaptureFixture): + with caplog.at_level(logging.INFO): + tf = Terraform(working_dir=current_path) + + 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'}), - ] - ) - 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' - - @pytest.mark.parametrize( - ("param"), - [ - ({}), - ({'module': 'test2'}), - ] - ) - 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 - 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 + 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("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} + ) + 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['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 == '' - - 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) + assert err == "" + + 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 == '' - - 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) + assert err == "" + + 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 - ): - workspace_name = 'test' - with workspace_setup_teardown(workspace_name) as tf: + 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 + ): 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 == '' - - 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') + assert err == "" + + 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 + ) diff --git a/test/var_to_output/test.tf b/test/var_to_output/test.tf index 71c2e8c..761d0bb 100644 --- a/test/var_to_output/test.tf +++ b/test/var_to_output/test.tf @@ -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 } diff --git a/test/var_to_output/test_map_var.json b/test/var_to_output/test_map_var.json index 0673f77..78d35a0 100644 --- a/test/var_to_output/test_map_var.json +++ b/test/var_to_output/test_map_var.json @@ -4,4 +4,4 @@ "e": "e", "f": "f" } -} \ No newline at end of file +} diff --git a/tox.ini b/tox.ini deleted file mode 100644 index 1f9a9cd..0000000 --- a/tox.ini +++ /dev/null @@ -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 \ No newline at end of file