diff --git a/test/fixtures/prevent_destroy/main.tf b/test/fixtures/prevent_destroy/main.tf new file mode 100644 index 0000000..2af55aa --- /dev/null +++ b/test/fixtures/prevent_destroy/main.tf @@ -0,0 +1,33 @@ +/** + * Copyright 2019 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + + +resource "null_resource" "sample" { + lifecycle { + prevent_destroy = true + } +} + + +resource "null_resource" "sample_bad_format" { + lifecycle { + prevent_destroy = true + } +} + +module "with_prevent_destroy" { + source = "./module" +} diff --git a/test/fixtures/prevent_destroy/module/main.tf b/test/fixtures/prevent_destroy/module/main.tf new file mode 100644 index 0000000..024db15 --- /dev/null +++ b/test/fixtures/prevent_destroy/module/main.tf @@ -0,0 +1,30 @@ +/** + * Copyright 2019 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + + +resource "null_resource" "sample" { + lifecycle { + prevent_destroy = true + } +} + + +resource "null_resource" "sample_bad_format" { + lifecycle { + prevent_destroy = true + } +} + diff --git a/test/test_prevent_destroy.py b/test/test_prevent_destroy.py new file mode 100644 index 0000000..f84a8c4 --- /dev/null +++ b/test/test_prevent_destroy.py @@ -0,0 +1,57 @@ +# Copyright 2019 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"Test init and plan with prevent_destroy lifecycles" +import logging +from unittest.mock import mock_open, patch + +import pytest +import tftest + + +def test_with_no_lifecycle_override(fixtures_dir): + tf = tftest.TerraformTest('prevent_destroy', fixtures_dir) + tf.setup() + tf.apply(auto_approve=True) + with pytest.raises(tftest.TerraformTestError): + tf.destroy(auto_approve=True) + tf.output() + + +def test_with_with_lifecycle_override(fixtures_dir): + tf = tftest.TerraformTest('prevent_destroy', fixtures_dir) + tf.setup(disable_prevent_destroy=True) + tf.apply(auto_approve=True) + tf.destroy(auto_approve=True) + + +@patch('tftest.shutil.copy') +def test_raises_exception_if_file_backup_fails(mock_copy, fixtures_dir, caplog): + caplog.set_level(logging.ERROR) + tf = tftest.TerraformTest('prevent_destroy', fixtures_dir) + mock_copy.side_effect = [FileNotFoundError('[Errno 2] No such file or directory: "sddsdsds"')] + with pytest.raises(tftest.TerraformTestError): + tf.setup(disable_prevent_destroy=True) + assert caplog.messages[0].startswith('Unable to backup terraform file ') + + +@patch('tftest.shutil.copy') +def test_raises_exception_when_read_or_write_fails(mock_copy, fixtures_dir, caplog): + caplog.set_level(logging.ERROR) + tf = tftest.TerraformTest('prevent_destroy', fixtures_dir) + with patch('builtins.open', mock_open(read_data='')) as mock_file: + mock_file.return_value.write.side_effect = [IOError('[Errno 13] Permission denied: "sddsdsds"')] + with pytest.raises(tftest.TerraformTestError): + tf.setup(disable_prevent_destroy=True) + assert caplog.messages[0].startswith('Unable to update prevent_destroy in file ') diff --git a/tftest.py b/tftest.py index 956b1c1..341aab3 100644 --- a/tftest.py +++ b/tftest.py @@ -33,10 +33,12 @@ import shutil import stat import subprocess +import sys import tempfile import weakref import re from functools import partial +from pathlib import Path from typing import List __version__ = '1.6.3' @@ -299,14 +301,13 @@ def __init__(self, tfdir, basedir=None, binary='terraform', env=None): self.env.update(env) @classmethod - def _cleanup(cls, tfdir, filenames, deep=True): + def _cleanup(cls, tfdir, filenames, deep=True, restore_files=False): """Remove linked files, .terraform and/or .terragrunt-cache folder at instance deletion.""" def remove_readonly(func, path, excinfo): _LOGGER.warning(f'Issue deleting file {path}, caused by {excinfo}') os.chmod(path, stat.S_IWRITE) func(path) - _LOGGER.debug('cleaning up %s %s', tfdir, filenames) for filename in filenames: path = os.path.join(tfdir, filename) @@ -323,13 +324,21 @@ def remove_readonly(func, path, excinfo): for tg_dir in glob.glob(path, recursive=True): if os.path.isdir(tg_dir): shutil.rmtree(tg_dir, onerror=remove_readonly) + _LOGGER.debug('Restoring original TF files after prevent destroy changes') + if restore_files: + for bkp_file in Path(tfdir).rglob('*.bkp'): + try: + shutil.copy(str(bkp_file), f'{str(bkp_file).strip(".bkp")}') + except (IOError, OSError): + _LOGGER.exception(f'Unable to restore terraform file {bkp_file.resolve()}') + raise TerraformTestError(f'Restore of terraform file ({bkp_file.resolve()}) failed') def _abspath(self, path): """Make relative path absolute from base dir.""" return path if path.startswith('/') else os.path.join(self._basedir, path) def setup(self, extra_files=None, plugin_dir=None, init_vars=None, - backend=True, cleanup_on_exit=True, **kw): + backend=True, cleanup_on_exit=True, disable_prevent_destroy=False, **kw): """Setup method to use in test fixtures. This method prepares a new Terraform environment for testing the module @@ -343,10 +352,33 @@ def setup(self, extra_files=None, plugin_dir=None, init_vars=None, init_vars: Terraform backend configuration variables backend: Terraform backend argument cleanup_on_exit: remove .terraform and terraform.tfstate files on exit + disable_prevent_destroy: set all prevent destroy to false Returns: Terraform init output. """ + # remove lifecycle prevent destroy + if disable_prevent_destroy: + min_python = (3, 5) + if sys.version_info < min_python: + raise TerraformTestError('The disable_prevent_destroy flag requires at least Python 3.5') + for tf_file in Path(self.tfdir).rglob('*.tf'): + try: + shutil.copy(str(tf_file), f'{str(tf_file)}.bkp') + # except (OSError, IOError) as exc: + except (OSError, IOError): + _LOGGER.exception(f'Unable to backup terraform file {tf_file.resolve()}') + raise TerraformTestError(f'Backup of terraform file ({tf_file.resolve()}) failed') + try: + with open(tf_file, 'r') as src: + terraform = src.read() + with open(tf_file, 'w') as src: + terraform = re.sub(r'prevent_destroy\s+=\s+true', 'prevent_destroy = false', terraform) + src.write(terraform) + except (OSError, IOError): + _LOGGER.exception(f'Unable to update prevent_destroy in file {tf_file.resolve()}') + raise TerraformTestError(f'Unable to update prevent_destroy in file ({tf_file.resolve()}) failed') + # link extra files inside dir filenames = [] for link_src in (extra_files or []): @@ -367,7 +399,7 @@ def setup(self, extra_files=None, plugin_dir=None, init_vars=None, else: _LOGGER.warning('no such file {}'.format(link_src)) self._finalizer = weakref.finalize( - self, self._cleanup, self.tfdir, filenames, deep=cleanup_on_exit) + self, self._cleanup, self.tfdir, filenames, deep=cleanup_on_exit, restore_files=disable_prevent_destroy) return self.init(plugin_dir=plugin_dir, init_vars=init_vars, backend=backend, **kw) def init(self, input=False, color=False, force_copy=False, plugin_dir=None,