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

1. let variables able to accept map and list

Browse source 
2. refactor to be able to accept command as a instance attribute
3. add test cases
This commit is contained in:
Freddy Tan 2016-12-20 18:53:01 +08:00
parent 2a13455db9
commit 2d63fa8716
7 changed files with 165 additions and 64 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

55
README.md
View file

@ -9,30 +9,36 @@ python-terraform is a python module provide a wrapper of `terraform` command lin
## Installation
pip install python-terraform
## Implementation
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
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
## Usage
For any terraform command
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
:return: ret_code, out, err
from python_terraform import Terraform
t = Terraform()
return_code, stdout, stderr = t.<cmd_name>(*arguments, **options)
For any options
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 like "-refresh=true", 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
## Examples
### Have a test.tf file under folder "/home/test"
#### apply with variables a=b, c=d, refresh=False, no color in the output
#### 1. apply with variables a=b, c=d, refresh=false, no color in the output
In shell:
cd /home/test
@ -41,19 +47,26 @@ In shell:
In python-terraform:
from python_terraform import Terraform
tf = terraform(working_dir='/home/test')
tf = Terraform(working_dir='/home/test')
tf.apply(no_color=IsFlagged, refresh=False, var={'a':'b', 'c':'d'})
#### taint command, allow-missing and no color
or
from python_terraform import Terraform
tf = Terraform()
tf.apply('/home/test', no_color=IsFlagged, refresh=False, var={'a':'b', 'c':'d'})
#### 2. fmt command, diff=true
In shell:
cd /home/test
terraform taint -allow-missing -no-color
terraform fmt -diff=true
In python-terraform:
from python_terraform import Terraform
tf = terraform(working_dir='/home/test')
tf.cmd('taint', allow_missing=IsFlagged, no_color=IsFlagged)
tf.fmt(diff=True)

53
python_terraform/__init__.py
View file

@ -5,6 +5,7 @@ import subprocess
import os
import json
import logging
import tempfile
from python_terraform.tfstate import Tfstate
@ -19,7 +20,7 @@ class IsNotFlagged:
pass
class Terraform:
class Terraform(object):
"""
Wrapper of terraform command line tool
https://www.terraform.io/
@ -38,7 +39,8 @@ class Terraform:
:param state: path of state file relative to working folder
:param variables: variables for apply/destroy/plan command
:param parallelism: parallelism for apply/destroy command
:param var_file: if specified, variables will not be used
: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
"""
self.working_dir = working_dir
@ -49,9 +51,18 @@ class Terraform:
self.terraform_bin_path = terraform_bin_path \
if terraform_bin_path else 'terraform'
self.var_file = var_file
self.temp_var_files = VaribleFiles()
# store the tfstate data
self.tfstate = dict()
self.tfstate = None
self.read_state_file(self.state)
def __getattr__(self, item):
def wrapper(*args, **kwargs):
print('called with %r and %r' % (args, kwargs))
return self.cmd(item, *args, **kwargs)
return wrapper
def apply(self, dir_or_plan=None, **kwargs):
"""
@ -74,7 +85,7 @@ class Terraform:
option_dict['var_file'] = self.var_file
option_dict['parallelism'] = self.parallelism
option_dict['no_color'] = IsFlagged
option_dict['input'] = True
option_dict['input'] = False
option_dict.update(kwargs)
args = [dir_or_plan] if dir_or_plan else []
return args, option_dict
@ -125,11 +136,12 @@ class Terraform:
cmds += ['-{k}={v}'.format(k=k, v=sub_v)]
continue
# right now we assume only variables will be passed as dict
# since map type sent in string won't work, create temp var file for
# variables, and clean it up later
if type(v) is dict:
for sub_k, sub_v in v.items():
cmds += ["-{k}='{var_k}={var_v}'".format(k=k,
var_k=sub_k,
var_v=sub_v)]
filename = self.temp_var_files.create(v)
cmds += ['-var-file={0}'.format(filename)]
continue
# simple flag,
@ -185,6 +197,8 @@ class Terraform:
self.read_state_file()
else:
log.warn('error: {e}'.format(e=err))
self.temp_var_files.clean_up()
return ret_code, out.decode('utf-8'), err.decode('utf-8')
def output(self, name):
@ -220,3 +234,26 @@ class Terraform:
file_path = os.path.join(self.working_dir, file_path)
self.tfstate = Tfstate.load_file(file_path)
def __exit__(self, exc_type, exc_value, traceback):
self.temp_var_files.clean_up()
class VaribleFiles(object):
def __init__(self):
self.files = []
def create(self, variables):
with tempfile.NamedTemporaryFile('w+t', delete=False) as temp:
logging.debug('{0} is created'.format(temp.name))
self.files.append(temp)
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 = []

2
python_terraform/tfstate.py
View file

@ -29,6 +29,6 @@ class Tfstate(object):
tf_state.tfstate_file = file_path
return tf_state
log.warn('{0} is not exist'.format(file_path))
log.debug('{0} is not exist'.format(file_path))
return Tfstate()

8
test/apply_tf/test.tf
View file

@ -1,8 +0,0 @@
variable "test_var" {}
provider "archive" {
}
output "test_output" {
value = "${var.test_var}"
}

69
test/test_terraform.py
View file

@ -4,22 +4,20 @@ import os
import logging
import re
logging.basicConfig(level=logging.WARN)
logging.basicConfig(level=logging.DEBUG)
current_path = os.path.dirname(os.path.realpath(__file__))
STRING_CASES = [
[
lambda x: x.generate_cmd_string('apply', 'the_folder',
no_color=IsFlagged,
var={'a': 'b', 'c': 'd'}),
"terraform apply -var='a=b' -var='c=d' -no-color the_folder"
no_color=IsFlagged),
"terraform apply -no-color the_folder"
],
[
lambda x: x.generate_cmd_string('push', 'path',
var={'a': 'b'}, vcs=True,
lambda x: x.generate_cmd_string('push', 'path', vcs=True,
token='token',
atlas_address='url'),
"terraform push -var='a=b' -vcs=true -token=token -atlas-address=url path"
"terraform push -vcs=true -token=token -atlas-address=url path"
],
]
@ -27,14 +25,14 @@ CMD_CASES = [
['method', 'expected_output'],
[
[
lambda x: x.cmd('plan', 'apply_tf', no_color=IsFlagged, var={'test_var': 'test'}) ,
lambda x: x.cmd('plan', 'var_to_output', no_color=IsFlagged, var={'test_var': 'test'}) ,
"doesn't need to do anything"
]
]
]
class TestTerraform:
class TestTerraform(object):
def teardown_method(self, method):
""" teardown any state that was previously setup with a setup_method
call.
@ -66,35 +64,58 @@ class TestTerraform:
assert expected_output in out
assert ret == 0
@pytest.mark.parametrize(
("folder", "variables", "var_files", "expected_output"),
[
("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}")
])
def test_apply(self, folder, variables, var_files, expected_output):
tf = Terraform(variables=variables, var_file=var_files)
ret, out, err = tf.apply(folder)
assert ret == 0
assert expected_output in out.replace('\n', '').replace(' ', '')
assert err == ''
def test_state_data(self):
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']
def test_apply(self):
cwd = os.path.join(current_path, 'apply_tf')
tf = Terraform(working_dir=cwd, variables={'test_var': 'test'})
ret, out, err = tf.apply(var={'test_var': 'test2'})
assert ret == 0
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']
def test_override_no_color(self):
cwd = os.path.join(current_path, 'apply_tf')
tf = Terraform(working_dir=cwd, variables={'test_var': 'test'})
ret, out, err = tf.apply(var={'test_var': 'test2'},
@pytest.mark.parametrize(
("folder", 'variables'),
[
("var_to_output", {'test_var': 'test'})
]
)
def test_override_default(self, folder, variables):
tf = Terraform(variables=variables)
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
def test_get_output(self):
cwd = os.path.join(current_path, 'apply_tf')
tf = Terraform(working_dir=cwd, variables={'test_var': 'test'})
tf.apply()
tf = Terraform(variables={'test_var': 'test'})
tf.apply('var_to_output')
assert tf.output('test_output') == 'test'
def test_destroy(self):
cwd = os.path.join(current_path, 'apply_tf')
tf = Terraform(working_dir=cwd, variables={'test_var': 'test'})
ret, out, err = tf.destroy()
tf = Terraform(variables={'test_var': 'test'})
ret, out, err = tf.destroy('var_to_output')
assert ret == 0
assert 'Destroy complete! Resources: 0 destroyed.' in out
def test_fmt(self):
tf = Terraform(variables={'test_var': 'test'})
ret, out, err = tf.fmt(diff=True)
assert ret == 0

31
test/var_to_output/test.tf Normal file
View file

@ -0,0 +1,31 @@
variable "test_var" {
default = ""
}
provider "archive" {}
variable "test_list_var" {
type = "list"
default = ["a", "b"]
}
variable "test_map_var" {
type = "map"
default = {
"a" = "a"
"b" = "b"
}
}
output "test_output" {
value = "${var.test_var}"
}
output "test_list_output" {
value = "${var.test_list_var}"
}
output "test_map_output" {
value = "${var.test_map_var}"
}

7
test/var_to_output/test_map_var.json Normal file
View file

@ -0,0 +1,7 @@
{
"test_map_var":
{
"e": "e",
"f": "f"
}
}