jasima/majuna
3
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Use disposable temporary directories to run automation jobs

Browse source
This commit is contained in:
Owen 2022-11-16 15:47:36 +00:00 • committed by irl
parent 9e5280280f
commit 0ebfe28b89
5 changed files with 36 additions and 42 deletions
Download patch file Download diff file Expand all files Collapse all files

36
app/terraform/terraform.py
View file

@ -27,7 +27,7 @@ class TerraformAutomation(BaseAutomation):
Short name for the provider used by this module.
"""
def automate(self, full: bool = False) -> Tuple[bool, str]:
def automate(self, working_dir: str, full: bool = False) -> Tuple[bool, str]:
"""
Runs the Terraform automation module. The run will follow these steps:
@ -41,17 +41,19 @@ class TerraformAutomation(BaseAutomation):
5. The :func:`tf_posthook` hook is run.
6. The logs from the apply step are returned as a string.
:param working_dir: temporary directory used to run the automation
:param full: include a Terraform refresh in the automation module run
:return: success status and Terraform apply logs
"""
prehook_result = self.tf_prehook() # pylint: disable=assignment-from-no-return
self.tf_generate()
self.tf_init()
returncode, logs = self.tf_apply(refresh=self.always_refresh or full)
self.tf_generate(working_dir)
self.tf_init(working_dir)
returncode, logs = self.tf_apply(working_dir, refresh=self.always_refresh or full)
self.tf_posthook(prehook_result=prehook_result)
return returncode == 0, logs
def tf_apply(self, *,
def tf_apply(self, working_dir: str, *,
refresh: bool = True,
parallelism: Optional[int] = None,
lock_timeout: int = 15) -> Tuple[int, str]:
@ -69,15 +71,15 @@ class TerraformAutomation(BaseAutomation):
f'-parallelism={str(parallelism)}',
f'-lock-timeout={str(lock_timeout)}m',
],
cwd=self.working_directory(),
cwd=working_dir,
stdout=subprocess.PIPE)
return tfcmd.returncode, tfcmd.stdout.decode('utf-8')
@abstractmethod
def tf_generate(self) -> None:
def tf_generate(self, working_dir) -> None:
raise NotImplementedError()
def tf_init(self, *,
def tf_init(self, working_dir: str, *,
lock_timeout: int = 15) -> None:
# The init command does not support JSON output.
# The following subprocess call takes external input, but is providing
@ -88,17 +90,17 @@ class TerraformAutomation(BaseAutomation):
'init',
f'-lock-timeout={str(lock_timeout)}m',
],
cwd=self.working_directory())
cwd=working_dir)
def tf_output(self) -> Any:
def tf_output(self, working_dir) -> Any:
# The following subprocess call does not take any user input.
tfcmd = subprocess.run( # nosec
['terraform', 'output', '-json'],
cwd=self.working_directory(),
cwd=working_dir,
stdout=subprocess.PIPE)
return json.loads(tfcmd.stdout)
def tf_plan(self, *,
def tf_plan(self, working_dir: str, *,
refresh: bool = True,
parallelism: Optional[int] = None,
lock_timeout: int = 15) -> Tuple[int, str]:
@ -113,7 +115,7 @@ class TerraformAutomation(BaseAutomation):
f'-parallelism={str(parallelism)}',
f'-lock-timeout={str(lock_timeout)}m',
],
cwd=self.working_directory())
cwd=working_dir)
return tfcmd.returncode, tfcmd.stdout.decode('utf-8')
def tf_posthook(self, *, prehook_result: Any = None) -> None:
@ -138,13 +140,13 @@ class TerraformAutomation(BaseAutomation):
:return: state that is useful to :func:`tf_posthook`, if required
"""
def tf_show(self) -> Any:
def tf_show(self, working_dir) -> Any:
# This subprocess call doesn't take any user input.
terraform = subprocess.run( # nosec
['terraform', 'show', '-json'],
cwd=self.working_directory(),
cwd=working_dir,
stdout=subprocess.PIPE)
return json.loads(terraform.stdout)
def tf_write(self, template: str, **kwargs: Any) -> None:
self.tmpl_write("main.tf", template, **kwargs)
def tf_write(self, template: str, working_dir: str, **kwargs: Any) -> None:
self.tmpl_write("main.tf", template, working_dir, **kwargs)