KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use set_type_default_version method in localstack
Best Python code snippet using localstack_python
test_project.py
Source:test_project.py 
1# fixture and parameter have the same name2# pylint: disable=redefined-outer-name,useless-super-delegation,protected-access3# pylint: disable=too-many-lines4import json5import logging6import os7import random8import string9import zipfile10from contextlib import contextmanager11from io import StringIO12from pathlib import Path13from unittest.mock import ANY, MagicMock, patch14import pytest15import yaml16from botocore.exceptions import ClientError, WaiterError17from rpdk.core.data_loaders import resource_json, resource_stream18from rpdk.core.exceptions import (19 DownstreamError,20 FragmentValidationError,21 InternalError,22 InvalidProjectError,23 SpecValidationError,24)25from rpdk.core.plugin_base import LanguagePlugin26from rpdk.core.project import (27 CFN_METADATA_FILENAME,28 LAMBDA_RUNTIMES,29 OVERRIDES_FILENAME,30 SCHEMA_UPLOAD_FILENAME,31 SETTINGS_FILENAME,32 Project,33 escape_markdown,34)35from rpdk.core.test import empty_override36from rpdk.core.upload import Uploader37from .utils import CONTENTS_UTF8, UnclosingBytesIO38ARTIFACT_TYPE_RESOURCE = "RESOURCE"39ARTIFACT_TYPE_MODULE = "MODULE"40LANGUAGE = "BQHDBC"41TYPE_NAME = "AWS::Color::Red"42MODULE_TYPE_NAME = "AWS::Color::Red::MODULE"43REGION = "us-east-1"44ENDPOINT = "cloudformation.beta.com"45RUNTIME = random.choice(list(LAMBDA_RUNTIMES))46BLANK_CLIENT_ERROR = {"Error": {"Code": "", "Message": ""}}47LOG = logging.getLogger(__name__)48REGISTRATION_TOKEN = "foo"49TYPE_ARN = "arn:aws:cloudformation:us-east-1:123456789012:type/resource/Foo-Bar-Foo"50TYPE_VERSION_ARN = (51 "arn:aws:cloudformation:us-east-1:123456789012:type/resource/Foo-Bar-Foo/00000001"52)53DESCRIBE_TYPE_COMPLETE_RETURN = {54 "TypeArn": TYPE_ARN,55 "TypeVersionArn": TYPE_VERSION_ARN,56 "Description": "Some detailed progress message.",57 "ProgressStatus": "COMPLETE",58}59DESCRIBE_TYPE_FAILED_RETURN = {60 "Description": "Some detailed progress message.",61 "ProgressStatus": "FAILED",62}63CREATE_INPUTS_FILE = "inputs/inputs_1_create.json"64UPDATE_INPUTS_FILE = "inputs/inputs_1_update.json"65INVALID_INPUTS_FILE = "inputs/inputs_1_invalid.json"66PLUGIN_INFORMATION = {67 "plugin-version": "2.1.3",68 "plugin-tool-version": "2.0.8",69 "plugin-name": "java",70}71@pytest.mark.parametrize("string", ["^[a-z]$", "([a-z])", ".*", "*."])72def test_escape_markdown_with_regex_names(string):73 assert escape_markdown(string).startswith("\\")74def test_escape_markdown_with_empty_string():75 assert escape_markdown("") == ""76 assert escape_markdown(None) is None77@pytest.mark.parametrize("string", ["Hello", "SomeProperty"])78def test_escape_markdown(string):79 assert escape_markdown(string) == string80@pytest.fixture81def project(tmpdir):82 unique_dir = "".join(random.choices(string.ascii_uppercase, k=12))83 return Project(root=tmpdir.mkdir(unique_dir))84@contextmanager85def patch_settings(project, data):86 with patch.object(project, "settings_path", autospec=True) as mock_path:87 mock_path.open.return_value.__enter__.return_value = StringIO(data)88 yield mock_path.open89def test_load_settings_invalid_json(project):90 with patch_settings(project, "") as mock_open:91 with pytest.raises(InvalidProjectError):92 project.load_settings()93 mock_open.assert_called_once_with("r", encoding="utf-8")94def test_load_settings_invalid_settings(project):95 with patch_settings(project, "{}") as mock_open:96 with pytest.raises(InvalidProjectError):97 project.load_settings()98 mock_open.assert_called_once_with("r", encoding="utf-8")99def test_load_settings_invalid_modules_settings(project):100 with patch_settings(project, '{"artifact_type": "MODULE"}') as mock_open:101 with pytest.raises(InvalidProjectError):102 project.load_settings()103 mock_open.assert_called_once_with("r", encoding="utf-8")104def test_load_settings_valid_json_for_resource(project):105 plugin = object()106 data = json.dumps(107 {108 "artifact_type": "RESOURCE",109 "typeName": TYPE_NAME,110 "language": LANGUAGE,111 "runtime": RUNTIME,112 "entrypoint": None,113 "testEntrypoint": None,114 "futureProperty": "value",115 }116 )117 patch_load = patch(118 "rpdk.core.project.load_plugin", autospec=True, return_value=plugin119 )120 with patch_settings(project, data) as mock_open, patch_load as mock_load:121 project.load_settings()122 mock_open.assert_called_once_with("r", encoding="utf-8")123 mock_load.assert_called_once_with(LANGUAGE)124 assert project.type_info == ("AWS", "Color", "Red")125 assert project.type_name == TYPE_NAME126 assert project.language == LANGUAGE127 assert project.artifact_type == ARTIFACT_TYPE_RESOURCE128 assert project._plugin is plugin129 assert project.settings == {}130def test_load_settings_valid_json_for_resource_backward_compatible(project):131 plugin = object()132 data = json.dumps(133 {134 "typeName": TYPE_NAME,135 "language": LANGUAGE,136 "runtime": RUNTIME,137 "entrypoint": None,138 "testEntrypoint": None,139 }140 )141 patch_load = patch(142 "rpdk.core.project.load_plugin", autospec=True, return_value=plugin143 )144 with patch_settings(project, data) as mock_open, patch_load as mock_load:145 project.load_settings()146 mock_open.assert_called_once_with("r", encoding="utf-8")147 mock_load.assert_called_once_with(LANGUAGE)148 assert project.type_info == ("AWS", "Color", "Red")149 assert project.type_name == TYPE_NAME150 assert project.language == LANGUAGE151 assert project.artifact_type == ARTIFACT_TYPE_RESOURCE152 assert project._plugin is plugin153 assert project.settings == {}154def test_load_settings_valid_json_for_module(project):155 plugin = object()156 data = json.dumps(157 {158 "artifact_type": "MODULE",159 "typeName": MODULE_TYPE_NAME,160 }161 )162 patch_load = patch(163 "rpdk.core.project.load_plugin", autospec=True, return_value=plugin164 )165 with patch_settings(project, data) as mock_open, patch_load as mock_load:166 project.load_settings()167 mock_open.assert_called_once_with("r", encoding="utf-8")168 mock_load.assert_not_called()169 assert project.type_info == ("AWS", "Color", "Red", "MODULE")170 assert project.type_name == MODULE_TYPE_NAME171 assert project.language is None172 assert project.artifact_type == ARTIFACT_TYPE_MODULE173 assert project._plugin is None174 assert project.settings == {}175def test_generate_for_modules_succeeds(project):176 project.type_info = ("AWS", "Color", "Red", "MODULE")177 project.artifact_type = ARTIFACT_TYPE_MODULE178 project.generate()179 project.generate_docs()180def test_load_schema_settings_not_loaded(project):181 with pytest.raises(InternalError):182 project.load_schema()183def test_load_schema_example(project):184 project.type_name = "AWS::Color::Blue"185 project._write_example_schema()186 project.load_schema()187def test_overwrite():188 mock_path = MagicMock(spec=Path)189 Project.overwrite(mock_path, LANGUAGE)190 mock_path.open.assert_called_once_with("w", encoding="utf-8")191 mock_f = mock_path.open.return_value.__enter__.return_value192 mock_f.write.assert_called_once_with(LANGUAGE)193def test_safewrite_overwrite(project):194 path = object()195 contents = object()196 patch_attr = patch.object(project, "overwrite_enabled", True)197 patch_meth = patch.object(project, "overwrite", autospec=True)198 with patch_attr, patch_meth as mock_overwrite:199 project.safewrite(path, contents)200 mock_overwrite.assert_called_once_with(path, contents)201def test_safewrite_doesnt_exist(project, tmpdir):202 path = Path(tmpdir.join("test")).resolve()203 with patch.object(project, "overwrite_enabled", False):204 project.safewrite(path, CONTENTS_UTF8)205 with path.open("r", encoding="utf-8") as f:206 assert f.read() == CONTENTS_UTF8207def test_safewrite_exists(project, tmpdir, caplog):208 caplog.set_level(logging.INFO)209 path = Path(tmpdir.join("test")).resolve()210 with path.open("w", encoding="utf-8") as f:211 f.write(CONTENTS_UTF8)212 with patch.object(project, "overwrite_enabled", False):213 project.safewrite(path, CONTENTS_UTF8)214 last_record = caplog.records[-1]215 assert last_record.levelname == "INFO"216 assert str(path) in last_record.message217def test_generate_no_handlers(project):218 project.schema = {}219 mock_plugin = MagicMock(spec=["generate"])220 with patch.object(project, "_plugin", mock_plugin):221 project.generate()222 project.generate_docs()223 mock_plugin.generate.assert_called_once_with(project)224@pytest.mark.parametrize(225 "schema_path,path",226 [227 ("data/schema/valid/valid_no_type.json", "generate_with_no_type_defined"),228 (229 "data/schema/valid/valid_type_complex.json",230 "generate_with_docs_type_complex",231 ),232 (233 "data/schema/valid/valid_pattern_properties.json",234 "generate_with_docs_pattern_properties",235 ),236 (237 "data/schema/valid/valid_no_properties.json",238 "generate_with_docs_no_properties",239 ),240 (241 "data/schema/valid/valid_nested_property_object.json",242 "generate_with_docs_nested_object",243 ),244 (245 "data/schema/valid/valid_type_composite_primary_identifier.json",246 "generate_with_docs_composite_primary_identifier",247 ),248 ],249)250def test_generate_with_docs(project, tmp_path_factory, schema_path, path):251 project.schema = resource_json(__name__, schema_path)252 project.type_name = "AWS::Color::Red"253 # tmpdir conflicts with other tests, make a unique one254 project.root = tmp_path_factory.mktemp(path)255 mock_plugin = MagicMock(spec=["generate"])256 with patch.object(project, "_plugin", mock_plugin):257 project.generate()258 project.generate_docs()259 mock_plugin.generate.assert_called_once_with(project)260 docs_dir = project.root / "docs"261 readme_file = project.root / "docs" / "README.md"262 assert docs_dir.is_dir()263 assert readme_file.is_file()264 with patch.object(project, "_plugin", mock_plugin):265 project.generate()266 readme_contents = readme_file.read_text(encoding="utf-8")267 assert project.type_name in readme_contents268def test_generate_docs_with_multityped_property(project, tmp_path_factory):269 project.schema = resource_json(270 __name__, "data/schema/valid/valid_multityped_property.json"271 )272 project.type_name = "AWS::Color::Red"273 # tmpdir conflicts with other tests, make a unique one274 project.root = tmp_path_factory.mktemp("generate_with_docs_type_complex")275 mock_plugin = MagicMock(spec=["generate"])276 with patch.object(project, "_plugin", mock_plugin):277 project.generate()278 project.generate_docs()279 mock_plugin.generate.assert_called_once_with(project)280 docs_dir = project.root / "docs"281 readme_file = project.root / "docs" / "README.md"282 assert docs_dir.is_dir()283 assert readme_file.is_file()284 with patch.object(project, "_plugin", mock_plugin):285 project.generate()286 readme_contents = readme_file.read_text(encoding="utf-8")287 readme_contents_target = resource_stream(288 __name__, "data/schema/target_output/multityped.md"289 )290 read_me_stripped = readme_contents.strip().replace(" ", "")291 read_me_target_stripped = readme_contents_target.read().strip().replace(" ", "")292 LOG.debug("read_me_stripped %s", read_me_stripped)293 LOG.debug("read_me_target_stripped %s", read_me_target_stripped)294 assert project.type_name in readme_contents295 assert read_me_stripped == read_me_target_stripped296def test_generate_docs_with_multiref_property(project, tmp_path_factory):297 project.schema = resource_json(298 __name__, "data/schema/valid/valid_multiref_property.json"299 )300 project.type_name = "AWS::Color::Red"301 # tmpdir conflicts with other tests, make a unique one302 project.root = tmp_path_factory.mktemp("generate_with_docs_type_complex")303 mock_plugin = MagicMock(spec=["generate"])304 with patch.object(project, "_plugin", mock_plugin):305 project.generate()306 project.generate_docs()307 mock_plugin.generate.assert_called_once_with(project)308 docs_dir = project.root / "docs"309 readme_file = project.root / "docs" / "README.md"310 assert docs_dir.is_dir()311 assert readme_file.is_file()312 with patch.object(project, "_plugin", mock_plugin):313 project.generate()314 readme_contents = readme_file.read_text(encoding="utf-8")315 readme_contents_target = resource_stream(316 __name__, "data/schema/target_output/multiref.md"317 )318 read_me_stripped = readme_contents.strip().replace(" ", "")319 read_me_target_stripped = readme_contents_target.read().strip().replace(" ", "")320 LOG.debug("read_me_stripped %s", read_me_stripped)321 LOG.debug("read_me_target_stripped %s", read_me_target_stripped)322 assert project.type_name in readme_contents323 assert read_me_stripped == read_me_target_stripped324def test_generate_with_docs_invalid_property_type(project, tmp_path_factory):325 project.schema = resource_json(326 __name__, "data/schema/invalid/invalid_property_type_invalid.json"327 )328 project.type_name = "AWS::Color::Red"329 # tmpdir conflicts with other tests, make a unique one330 project.root = tmp_path_factory.mktemp("generate_with_docs_invalid_property_type")331 mock_plugin = MagicMock(spec=["generate"])332 with patch.object(project, "_plugin", mock_plugin):333 # skip actual generation334 project.generate_docs()335 docs_dir = project.root / "docs"336 readme_file = project.root / "docs" / "README.md"337 assert docs_dir.is_dir()338 assert readme_file.is_file()339 with patch.object(project, "_plugin", mock_plugin):340 project.generate()341 project.generate_docs()342 readme_contents = readme_file.read_text(encoding="utf-8")343 assert project.type_name in readme_contents344def test_generate_with_docs_no_type(project, tmp_path_factory):345 project.schema = {"properties": {}}346 # tmpdir conflicts with other tests, make a unique one347 project.root = tmp_path_factory.mktemp("generate_with_docs_no_type")348 mock_plugin = MagicMock(spec=["generate"])349 with patch.object(project, "_plugin", mock_plugin):350 project.generate()351 project.generate_docs()352 mock_plugin.generate.assert_called_once_with(project)353 docs_dir = project.root / "docs"354 assert not docs_dir.is_dir()355def test_generate_with_docs_twice(project, tmp_path_factory):356 project.schema = {"properties": {}}357 project.type_name = "AWS::Color::Red"358 # tmpdir conflicts with other tests, make a unique one359 project.root = tmp_path_factory.mktemp("generate_with_docs_twice")360 mock_plugin = MagicMock(spec=["generate"])361 with patch.object(project, "_plugin", mock_plugin):362 project.generate()363 project.generate_docs()364 mock_plugin.generate.assert_called_once_with(project)365 docs_dir = project.root / "docs"366 readme_file = docs_dir / "README.md"367 assert docs_dir.is_dir()368 assert readme_file.is_file()369 with patch.object(project, "_plugin", mock_plugin):370 project.generate()371 project.generate_docs()372 assert docs_dir.is_dir()373 assert readme_file.is_file()374 with patch.object(project, "_plugin", mock_plugin):375 project.generate()376 project.generate_docs()377 readme_contents = readme_file.read_text(encoding="utf-8")378 assert project.type_name in readme_contents379def test_generate_handlers(project, tmpdir):380 project.type_name = "Test::Handler::Test"381 expected_actions = {"createAction", "readAction"}382 project.schema = {383 "handlers": {384 "create": {"permissions": ["createAction", "readAction"]},385 "read": {"permissions": ["readAction", ""]},386 }387 }388 project.root = tmpdir389 mock_plugin = MagicMock(spec=["generate"])390 with patch.object(project, "_plugin", mock_plugin):391 project.generate()392 role_path = project.root / "resource-role.yaml"393 with role_path.open("r", encoding="utf-8") as f:394 template = yaml.safe_load(f.read())395 action_list = template["Resources"]["ExecutionRole"]["Properties"]["Policies"][0][396 "PolicyDocument"397 ]["Statement"][0]["Action"]398 assert all(action in expected_actions for action in action_list)399 assert len(action_list) == len(expected_actions)400 assert template["Outputs"]["ExecutionRoleArn"]401 mock_plugin.generate.assert_called_once_with(project)402@pytest.mark.parametrize(403 "schema",404 ({"handlers": {"create": {"permissions": [""]}}}, {"handlers": {"create": {}}}),405)406def test_generate_handlers_deny_all(project, tmpdir, schema):407 project.type_name = "Test::Handler::Test"408 project.schema = schema409 project.root = tmpdir410 mock_plugin = MagicMock(spec=["generate"])411 with patch.object(project, "_plugin", mock_plugin):412 project.generate()413 role_path = project.root / "resource-role.yaml"414 with role_path.open("r", encoding="utf-8") as f:415 template = yaml.safe_load(f.read())416 statement = template["Resources"]["ExecutionRole"]["Properties"]["Policies"][0][417 "PolicyDocument"418 ]["Statement"][0]419 assert statement["Effect"] == "Deny"420 assert statement["Action"][0] == "*"421 mock_plugin.generate.assert_called_once_with(project)422@pytest.mark.parametrize(423 "schema,result",424 (425 ({"handlers": {"create": {"timeoutInMinutes": 720}}}, 43200),426 ({"handlers": {"create": {"timeoutInMinutes": 2}}}, 3600),427 ({"handlers": {"create": {"timeoutInMinutes": 90}}}, 6300),428 (429 {430 "handlers": {431 "create": {"timeoutInMinutes": 70},432 "update": {"timeoutInMinutes": 90},433 }434 },435 6300,436 ),437 ({"handlers": {"create": {}}}, 8400),438 ({"handlers": {"create": {"timeoutInMinutes": 90}, "read": {}}}, 8400),439 ),440)441def test_generate_handlers_role_session_timeout(project, tmpdir, schema, result):442 project.type_name = "Test::Handler::Test"443 project.schema = schema444 project.root = tmpdir445 mock_plugin = MagicMock(spec=["generate"])446 with patch.object(project, "_plugin", mock_plugin):447 project.generate()448 role_path = project.root / "resource-role.yaml"449 with role_path.open("r", encoding="utf-8") as f:450 template = yaml.safe_load(f.read())451 max_session_timeout = template["Resources"]["ExecutionRole"]["Properties"][452 "MaxSessionDuration"453 ]454 assert max_session_timeout == result455 mock_plugin.generate.assert_called_once_with(project)456def test_init_resource(project):457 type_name = "AWS::Color::Red"458 mock_plugin = MagicMock(spec=["init"])459 patch_load_plugin = patch(460 "rpdk.core.project.load_plugin", autospec=True, return_value=mock_plugin461 )462 with patch_load_plugin as mock_load_plugin:463 project.init(type_name, LANGUAGE)464 mock_load_plugin.assert_called_once_with(LANGUAGE)465 mock_plugin.init.assert_called_once_with(project)466 assert project.type_info == ("AWS", "Color", "Red")467 assert project.type_name == type_name468 assert project.language == LANGUAGE469 assert project.artifact_type == ARTIFACT_TYPE_RESOURCE470 assert project._plugin is mock_plugin471 assert project.settings == {}472 with project.settings_path.open("r", encoding="utf-8") as f:473 assert json.load(f)474 # ends with newline475 with project.settings_path.open("rb") as f:476 f.seek(-1, os.SEEK_END)477 assert f.read() == b"\n"478 with project.schema_path.open("r", encoding="utf-8") as f:479 assert json.load(f)480 for file_inputs in (481 "inputs_1_create.json",482 "inputs_1_update.json",483 "inputs_1_invalid.json",484 ):485 path_file = project.example_inputs_path / file_inputs486 with path_file.open("r", encoding="utf-8") as f:487 assert json.load(f)488 # ends with newline489 with project.schema_path.open("rb") as f:490 f.seek(-1, os.SEEK_END)491 assert f.read() == b"\n"492def test_init_module(project):493 type_name = "AWS::Color::Red"494 mock_plugin = MagicMock(spec=["init"])495 patch_load_plugin = patch(496 "rpdk.core.project.load_plugin", autospec=True, return_value=mock_plugin497 )498 with patch_load_plugin as mock_load_plugin:499 project.init_module(type_name)500 mock_load_plugin.assert_not_called()501 mock_plugin.init.assert_not_called()502 assert project.type_info == ("AWS", "Color", "Red")503 assert project.type_name == type_name504 assert project.language is None505 assert project.artifact_type == ARTIFACT_TYPE_MODULE506 assert project._plugin is None507 assert project.settings == {}508 with project.settings_path.open("r", encoding="utf-8") as f:509 assert json.load(f)510 # ends with newline511 with project.settings_path.open("rb") as f:512 f.seek(-1, os.SEEK_END)513 assert f.read() == b"\n"514def test_load_invalid_schema(project):515 patch_settings = patch.object(project, "load_settings")516 patch_schema = patch.object(517 project, "load_schema", side_effect=SpecValidationError("")518 )519 with patch_settings as mock_settings, patch_schema as mock_schema, pytest.raises(520 InvalidProjectError521 ) as excinfo:522 project.load()523 mock_settings.assert_called_once_with()524 mock_schema.assert_called_once_with()525 assert "invalid" in str(excinfo.value)526def test_load_module_project_succeeds(project):527 project.artifact_type = "MODULE"528 project.type_name = "Unit::Test::Malik::MODULE"529 patch_load_settings = patch.object(530 project, "load_settings", return_value={"artifact_type": "MODULE"}531 )532 with patch_load_settings:533 project.load()534def test_load_resource_succeeds(project):535 project.artifact_type = "Resource"536 project.type_name = "Unit::Test::Resource"537 patch_load_settings = patch.object(538 project, "load_settings", return_value={"artifact_type": "RESOURCE"}539 )540 project._write_example_schema()541 with patch_load_settings:542 project.load()543def test_load_module_project_with_invalid_fragments(project):544 project.artifact_type = "MODULE"545 project.type_name = "Unit::Test::Malik::MODULE"546 patch_load_settings = patch.object(547 project, "load_settings", return_value={"artifact_type": "MODULE"}548 )549 patch_validate = patch.object(550 project, "_validate_fragments", side_effect=FragmentValidationError551 )552 with patch_load_settings, patch_validate, pytest.raises(InvalidProjectError):553 project.load()554def test_schema_not_found(project):555 patch_settings = patch.object(project, "load_settings")556 patch_schema = patch.object(project, "load_schema", side_effect=FileNotFoundError)557 with patch_settings as mock_settings, patch_schema as mock_schema, pytest.raises(558 InvalidProjectError559 ) as excinfo:560 project.load()561 mock_settings.assert_called_once_with()562 mock_schema.assert_called_once_with()563 assert "not found" in str(excinfo.value)564def test_settings_not_found(project):565 patch_settings = patch.object(566 project, "load_settings", side_effect=FileNotFoundError567 )568 patch_schema = patch.object(project, "load_schema")569 with patch_settings as mock_settings, patch_schema as mock_schema, pytest.raises(570 InvalidProjectError571 ) as excinfo:572 project.load()573 mock_settings.assert_called_once_with()574 mock_schema.assert_not_called()575 assert "not found" in str(excinfo.value)576 assert "init" in str(excinfo.value)577def create_input_file(base):578 path = base / "inputs"579 os.mkdir(path, mode=0o777)580 path_create = base / CREATE_INPUTS_FILE581 with path_create.open("w", encoding="utf-8") as f:582 f.write("{}")583 path_update = base / UPDATE_INPUTS_FILE584 with path_update.open("w", encoding="utf-8") as f:585 f.write("{}")586 path_invalid = base / INVALID_INPUTS_FILE587 with path_invalid.open("w", encoding="utf-8") as f:588 f.write("{}")589# pylint: disable=too-many-arguments, too-many-locals590def test_submit_dry_run(project):591 project.type_name = TYPE_NAME592 project.runtime = RUNTIME593 project.language = LANGUAGE594 project.artifact_type = ARTIFACT_TYPE_RESOURCE595 zip_path = project.root / "test.zip"596 with project.schema_path.open("w", encoding="utf-8") as f:597 f.write(CONTENTS_UTF8)598 with project.overrides_path.open("w", encoding="utf-8") as f:599 f.write(json.dumps(empty_override()))600 create_input_file(project.root)601 project.write_settings()602 patch_plugin = patch.object(project, "_plugin", spec=LanguagePlugin)603 patch_upload = patch.object(project, "_upload", autospec=True)604 patch_path = patch("rpdk.core.project.Path", return_value=zip_path)605 patch_temp = patch("rpdk.core.project.TemporaryFile", autospec=True)606 # fmt: off607 # these context managers can't be wrapped by black, but it removes the \608 with patch_plugin as mock_plugin, patch_path as mock_path, \609 patch_temp as mock_temp, patch_upload as mock_upload:610 mock_plugin.get_plugin_information = MagicMock(return_value=PLUGIN_INFORMATION)611 project.submit(612 True,613 endpoint_url=ENDPOINT,614 region_name=REGION,615 role_arn=None,616 use_role=True,617 set_default=False618 )619 # fmt: on620 mock_temp.assert_not_called()621 mock_path.assert_called_once_with("{}.zip".format(project.hypenated_name))622 mock_plugin.package.assert_called_once_with(project, ANY)623 mock_upload.assert_not_called()624 with zipfile.ZipFile(zip_path, mode="r") as zip_file:625 assert set(zip_file.namelist()) == {626 SCHEMA_UPLOAD_FILENAME,627 SETTINGS_FILENAME,628 OVERRIDES_FILENAME,629 CREATE_INPUTS_FILE,630 INVALID_INPUTS_FILE,631 UPDATE_INPUTS_FILE,632 CFN_METADATA_FILENAME,633 }634 schema_contents = zip_file.read(SCHEMA_UPLOAD_FILENAME).decode("utf-8")635 assert schema_contents == CONTENTS_UTF8636 settings = json.loads(zip_file.read(SETTINGS_FILENAME).decode("utf-8"))637 assert settings["runtime"] == RUNTIME638 overrides = json.loads(zip_file.read(OVERRIDES_FILENAME).decode("utf-8"))639 assert "CREATE" in overrides640 # https://docs.python.org/3/library/zipfile.html#zipfile.ZipFile.testzip641 input_create = json.loads(zip_file.read(CREATE_INPUTS_FILE).decode("utf-8"))642 assert input_create == {}643 input_invalid = json.loads(zip_file.read(UPDATE_INPUTS_FILE).decode("utf-8"))644 assert input_invalid == {}645 input_update = json.loads(zip_file.read(INVALID_INPUTS_FILE).decode("utf-8"))646 assert input_update == {}647 assert zip_file.testzip() is None648 metadata_info = json.loads(zip_file.read(CFN_METADATA_FILENAME).decode("utf-8"))649 assert "cli-version" in metadata_info650 assert "plugin-version" in metadata_info651 assert "plugin-tool-version" in metadata_info652def test_submit_dry_run_modules(project):653 project.type_name = MODULE_TYPE_NAME654 project.runtime = RUNTIME655 project.language = LANGUAGE656 project.artifact_type = ARTIFACT_TYPE_MODULE657 project.fragment_dir = project.root / "fragments"658 zip_path = project.root / "test.zip"659 schema_path = project.root / "schema.json"660 fragment_path = project.root / "fragments" / "fragment.json"661 with project.schema_path.open("w", encoding="utf-8") as f:662 f.write(CONTENTS_UTF8)663 with schema_path.open("w", encoding="utf-8") as f:664 f.write(CONTENTS_UTF8)665 if not os.path.exists(project.root / "fragments"):666 os.mkdir(project.root / "fragments")667 with fragment_path.open("w", encoding="utf-8") as f:668 f.write(CONTENTS_UTF8)669 with project.overrides_path.open("w", encoding="utf-8") as f:670 f.write(json.dumps(empty_override()))671 project.write_settings()672 patch_plugin = patch.object(project, "_plugin", spec=LanguagePlugin)673 patch_upload = patch.object(project, "_upload", autospec=True)674 patch_path = patch("rpdk.core.project.Path", return_value=zip_path)675 patch_temp = patch("rpdk.core.project.TemporaryFile", autospec=True)676 # fmt: off677 # these context managers can't be wrapped by black, but it removes the \678 with patch_plugin as mock_plugin, patch_path as mock_path, \679 patch_temp as mock_temp, patch_upload as mock_upload:680 project.submit(681 True,682 endpoint_url=ENDPOINT,683 region_name=REGION,684 role_arn=None,685 use_role=True,686 set_default=False687 )688 # fmt: on689 mock_temp.assert_not_called()690 mock_path.assert_called_once_with("{}.zip".format(project.hypenated_name))691 mock_plugin.package.assert_not_called()692 mock_upload.assert_not_called()693 fragment_file_name = "fragments/fragment.json"694 with zipfile.ZipFile(zip_path, mode="r") as zip_file:695 assert set(zip_file.namelist()) == {696 fragment_file_name,697 SCHEMA_UPLOAD_FILENAME,698 SETTINGS_FILENAME,699 OVERRIDES_FILENAME,700 }701 schema_contents = zip_file.read(SCHEMA_UPLOAD_FILENAME).decode("utf-8")702 assert schema_contents == CONTENTS_UTF8703 overrides = json.loads(zip_file.read(OVERRIDES_FILENAME).decode("utf-8"))704 assert "CREATE" in overrides705 # https://docs.python.org/3/library/zipfile.html#zipfile.ZipFile.testzip706 assert zip_file.testzip() is None707def test_submit_live_run(project):708 project.type_name = TYPE_NAME709 project.runtime = RUNTIME710 project.language = LANGUAGE711 project.artifact_type = ARTIFACT_TYPE_RESOURCE712 with project.schema_path.open("w", encoding="utf-8") as f:713 f.write(CONTENTS_UTF8)714 project.write_settings()715 temp_file = UnclosingBytesIO()716 patch_plugin = patch.object(project, "_plugin", spec=LanguagePlugin)717 patch_upload = patch.object(project, "_upload", autospec=True)718 patch_path = patch("rpdk.core.project.Path", autospec=True)719 patch_temp = patch("rpdk.core.project.TemporaryFile", return_value=temp_file)720 # fmt: off721 # these context managers can't be wrapped by black, but it removes the \722 with patch_plugin as mock_plugin, patch_path as mock_path, \723 patch_temp as mock_temp, patch_upload as mock_upload:724 project.submit(725 False,726 endpoint_url=ENDPOINT,727 region_name=REGION,728 role_arn=None,729 use_role=True,730 set_default=True731 )732 # fmt: on733 mock_path.assert_not_called()734 mock_temp.assert_called_once_with("w+b")735 mock_plugin.package.assert_called_once_with(project, ANY)736 # zip file construction is tested by the dry-run test737 assert temp_file.tell() == 0 # file was rewound before upload738 mock_upload.assert_called_once_with(739 temp_file,740 region_name=REGION,741 endpoint_url=ENDPOINT,742 role_arn=None,743 use_role=True,744 set_default=True,745 )746 assert temp_file._was_closed747 temp_file._close()748def test_submit_live_run_for_module(project):749 project.type_name = MODULE_TYPE_NAME750 project.runtime = RUNTIME751 project.language = LANGUAGE752 project.artifact_type = ARTIFACT_TYPE_MODULE753 with project.schema_path.open("w", encoding="utf-8") as f:754 f.write(CONTENTS_UTF8)755 project.write_settings()756 temp_file = UnclosingBytesIO()757 patch_plugin = patch.object(project, "_plugin", spec=LanguagePlugin)758 patch_path = patch("rpdk.core.project.Path", autospec=True)759 patch_temp = patch("rpdk.core.project.TemporaryFile", return_value=temp_file)760 # fmt: off761 # these context managers can't be wrapped by black, but it removes the \762 with patch_plugin as mock_plugin, patch_path as mock_path, \763 patch_temp as mock_temp, \764 pytest.raises(InternalError):765 project.submit(766 False,767 endpoint_url=ENDPOINT,768 region_name=REGION,769 role_arn=None,770 use_role=True,771 set_default=True772 )773 # fmt: on774 mock_path.assert_not_called()775 mock_temp.assert_called_once_with("w+b")776 mock_plugin.package.assert_not_called()777 temp_file._close()778def test__upload_good_path_create_role_and_set_default(project):779 project.type_name = TYPE_NAME780 project.artifact_type = ARTIFACT_TYPE_RESOURCE781 project.schema = {"handlers": {}}782 mock_cfn_client = MagicMock(spec=["register_type"])783 mock_cfn_client.register_type.return_value = {"RegistrationToken": "foo"}784 fileobj = object()785 patch_sdk = patch("rpdk.core.project.create_sdk_session", autospec=True)786 patch_uploader = patch.object(Uploader, "upload", return_value="url")787 patch_exec_role_arn = patch.object(788 Uploader, "create_or_update_role", return_value="some-execution-role-arn"789 )790 patch_logging_role_arn = patch.object(791 Uploader, "get_log_delivery_role_arn", return_value="some-log-role-arn"792 )793 patch_uuid = patch("rpdk.core.project.uuid4", autospec=True, return_value="foo")794 patch_wait = patch.object(project, "_wait_for_registration", autospec=True)795 with patch_sdk as mock_sdk, patch_uploader as mock_upload_method, patch_logging_role_arn as mock_role_arn_method, patch_exec_role_arn as mock_exec_role_method: # noqa: B950 as it conflicts with formatting rules # pylint: disable=C0301796 mock_sdk.return_value.client.side_effect = [mock_cfn_client, MagicMock()]797 with patch_uuid as mock_uuid, patch_wait as mock_wait:798 project._upload(799 fileobj,800 endpoint_url=None,801 region_name=None,802 role_arn=None,803 use_role=True,804 set_default=True,805 )806 mock_sdk.assert_called_once_with(None)807 mock_exec_role_method.assert_called_once_with(808 project.root / "resource-role.yaml", project.hypenated_name809 )810 mock_upload_method.assert_called_once_with(project.hypenated_name, fileobj)811 mock_role_arn_method.assert_called_once_with()812 mock_uuid.assert_called_once_with()813 mock_cfn_client.register_type.assert_called_once_with(814 Type="RESOURCE",815 TypeName=project.type_name,816 SchemaHandlerPackage="url",817 ClientRequestToken=mock_uuid.return_value,818 LoggingConfig={819 "LogRoleArn": "some-log-role-arn",820 "LogGroupName": "aws-color-red-logs",821 },822 ExecutionRoleArn="some-execution-role-arn",823 )824 mock_wait.assert_called_once_with(mock_cfn_client, "foo", True)825@pytest.mark.parametrize(826 ("use_role,expected_additional_args"),827 [(True, {"ExecutionRoleArn": "someArn"}), (False, {})],828)829def test__upload_good_path_skip_role_creation(830 project, use_role, expected_additional_args831):832 project.type_name = TYPE_NAME833 project.artifact_type = ARTIFACT_TYPE_RESOURCE834 project.schema = {"handlers": {}}835 mock_cfn_client = MagicMock(spec=["register_type"])836 fileobj = object()837 mock_cfn_client.register_type.return_value = {"RegistrationToken": "foo"}838 patch_sdk = patch("rpdk.core.project.create_sdk_session", autospec=True)839 patch_uploader = patch.object(Uploader, "upload", return_value="url")840 patch_logging_role_arn = patch.object(841 Uploader, "get_log_delivery_role_arn", return_value="some-log-role-arn"842 )843 patch_uuid = patch("rpdk.core.project.uuid4", autospec=True, return_value="foo")844 patch_wait = patch.object(project, "_wait_for_registration", autospec=True)845 with patch_sdk as mock_sdk, patch_uploader as mock_upload_method, patch_logging_role_arn as mock_role_arn_method: # noqa: B950 as it conflicts with formatting rules # pylint: disable=C0301846 mock_sdk.return_value.client.side_effect = [mock_cfn_client, MagicMock()]847 with patch_uuid as mock_uuid, patch_wait as mock_wait:848 project._upload(849 fileobj,850 endpoint_url=None,851 region_name=None,852 role_arn="someArn",853 use_role=use_role,854 set_default=True,855 )856 mock_sdk.assert_called_once_with(None)857 mock_upload_method.assert_called_once_with(project.hypenated_name, fileobj)858 mock_role_arn_method.assert_called_once_with()859 mock_uuid.assert_called_once_with()860 mock_wait.assert_called_once_with(mock_cfn_client, "foo", True)861 mock_cfn_client.register_type.assert_called_once_with(862 Type="RESOURCE",863 TypeName=project.type_name,864 SchemaHandlerPackage="url",865 ClientRequestToken=mock_uuid.return_value,866 LoggingConfig={867 "LogRoleArn": "some-log-role-arn",868 "LogGroupName": "aws-color-red-logs",869 },870 **expected_additional_args871 )872def test__upload_clienterror(project):873 project.type_name = TYPE_NAME874 project.artifact_type = ARTIFACT_TYPE_RESOURCE875 project.schema = {}876 mock_cfn_client = MagicMock(spec=["register_type"])877 mock_cfn_client.register_type.side_effect = ClientError(878 BLANK_CLIENT_ERROR, "RegisterType"879 )880 fileobj = object()881 patch_sdk = patch("rpdk.core.project.create_sdk_session", autospec=True)882 patch_uploader = patch.object(Uploader, "upload", return_value="url")883 patch_role_arn = patch.object(884 Uploader, "get_log_delivery_role_arn", return_value="some-log-role-arn"885 )886 patch_uuid = patch("rpdk.core.project.uuid4", autospec=True, return_value="foo")887 with patch_sdk as mock_sdk, patch_uploader as mock_upload_method, patch_role_arn as mock_role_arn_method: # noqa: B950 as it conflicts with formatting rules # pylint: disable=C0301888 mock_session = mock_sdk.return_value889 mock_session.client.side_effect = [mock_cfn_client, MagicMock()]890 with patch_uuid as mock_uuid, pytest.raises(DownstreamError):891 project._upload(892 fileobj,893 endpoint_url=None,894 region_name=None,895 role_arn=None,896 use_role=False,897 set_default=True,898 )899 mock_sdk.assert_called_once_with(None)900 mock_upload_method.assert_called_once_with(project.hypenated_name, fileobj)901 mock_role_arn_method.assert_called_once_with()902 mock_uuid.assert_called_once_with()903 mock_cfn_client.register_type.assert_called_once_with(904 Type="RESOURCE",905 TypeName=project.type_name,906 SchemaHandlerPackage="url",907 ClientRequestToken=mock_uuid.return_value,908 LoggingConfig={909 "LogRoleArn": "some-log-role-arn",910 "LogGroupName": "aws-color-red-logs",911 },912 )913def test__upload_clienterror_module(project):914 project.type_name = MODULE_TYPE_NAME915 project.artifact_type = ARTIFACT_TYPE_MODULE916 project.schema = {}917 mock_cfn_client = MagicMock(spec=["register_type"])918 mock_cfn_client.register_type.side_effect = ClientError(919 BLANK_CLIENT_ERROR, "RegisterType"920 )921 fileobj = object()922 patch_sdk = patch("rpdk.core.project.create_sdk_session", autospec=True)923 patch_uploader = patch.object(Uploader, "upload", return_value="url")924 patch_role_arn = patch.object(925 Uploader, "get_log_delivery_role_arn", return_value="some-log-role-arn"926 )927 patch_uuid = patch("rpdk.core.project.uuid4", autospec=True, return_value="foo")928 with patch_sdk as mock_sdk, patch_uploader as mock_upload_method, patch_role_arn as mock_role_arn_method: # noqa: B950 as it conflicts with formatting rules # pylint: disable=C0301929 mock_session = mock_sdk.return_value930 mock_session.client.side_effect = [mock_cfn_client, MagicMock()]931 with patch_uuid as mock_uuid, pytest.raises(DownstreamError):932 project._upload(933 fileobj,934 endpoint_url=None,935 region_name=None,936 role_arn=None,937 use_role=False,938 set_default=True,939 )940 mock_sdk.assert_called_once_with(None)941 mock_upload_method.assert_called_once_with(project.hypenated_name, fileobj)942 mock_role_arn_method.assert_called_once_with()943 mock_uuid.assert_called_once_with()944 mock_cfn_client.register_type.assert_called_once_with(945 Type="MODULE",946 TypeName=project.type_name,947 SchemaHandlerPackage="url",948 ClientRequestToken=mock_uuid.return_value,949 LoggingConfig={950 "LogRoleArn": "some-log-role-arn",951 "LogGroupName": "aws-color-red-module-logs",952 },953 )954def test__wait_for_registration_set_default(project):955 mock_cfn_client = MagicMock(956 spec=["describe_type_registration", "set_type_default_version", "get_waiter"]957 )958 mock_cfn_client.describe_type_registration.return_value = (959 DESCRIBE_TYPE_COMPLETE_RETURN960 )961 mock_waiter = MagicMock(spec=["wait"])962 mock_cfn_client.get_waiter.return_value = mock_waiter963 project._wait_for_registration(mock_cfn_client, REGISTRATION_TOKEN, True)964 mock_cfn_client.describe_type_registration.assert_called_once_with(965 RegistrationToken=REGISTRATION_TOKEN966 )967 mock_cfn_client.set_type_default_version.assert_called_once_with(968 Arn=TYPE_VERSION_ARN969 )970 mock_waiter.wait.assert_called_once_with(RegistrationToken=REGISTRATION_TOKEN)971def test__wait_for_registration_set_default_fails(project):972 mock_cfn_client = MagicMock(973 spec=["describe_type_registration", "set_type_default_version", "get_waiter"]974 )975 mock_cfn_client.describe_type_registration.return_value = (976 DESCRIBE_TYPE_COMPLETE_RETURN977 )978 mock_cfn_client.set_type_default_version.side_effect = ClientError(979 BLANK_CLIENT_ERROR, "SetTypeDefaultVersion"980 )981 mock_waiter = MagicMock(spec=["wait"])982 mock_cfn_client.get_waiter.return_value = mock_waiter983 with pytest.raises(DownstreamError):984 project._wait_for_registration(mock_cfn_client, REGISTRATION_TOKEN, True)985 mock_cfn_client.describe_type_registration.assert_called_once_with(986 RegistrationToken=REGISTRATION_TOKEN987 )988 mock_cfn_client.set_type_default_version.assert_called_once_with(989 Arn=TYPE_VERSION_ARN990 )991 mock_waiter.wait.assert_called_once_with(RegistrationToken=REGISTRATION_TOKEN)992def test__wait_for_registration_no_set_default(project):993 mock_cfn_client = MagicMock(994 spec=["describe_type_registration", "set_type_default_version", "get_waiter"]995 )996 mock_cfn_client.describe_type_registration.return_value = (997 DESCRIBE_TYPE_COMPLETE_RETURN998 )999 mock_waiter = MagicMock(spec=["wait"])1000 mock_cfn_client.get_waiter.return_value = mock_waiter1001 project._wait_for_registration(mock_cfn_client, REGISTRATION_TOKEN, False)1002 mock_cfn_client.describe_type_registration.assert_called_once_with(1003 RegistrationToken=REGISTRATION_TOKEN1004 )1005 mock_cfn_client.set_type_default_version.assert_not_called()1006 mock_waiter.wait.assert_called_once_with(RegistrationToken=REGISTRATION_TOKEN)1007def test__wait_for_registration_waiter_fails(project):1008 mock_cfn_client = MagicMock(1009 spec=["describe_type_registration", "set_type_default_version", "get_waiter"]1010 )1011 mock_cfn_client.describe_type_registration.return_value = (1012 DESCRIBE_TYPE_FAILED_RETURN1013 )1014 mock_waiter = MagicMock(spec=["wait"])1015 mock_waiter.wait.side_effect = WaiterError(1016 "TypeRegistrationComplete",1017 "Waiter encountered a terminal failure state",1018 DESCRIBE_TYPE_FAILED_RETURN,1019 )1020 mock_cfn_client.get_waiter.return_value = mock_waiter1021 with pytest.raises(DownstreamError):1022 project._wait_for_registration(mock_cfn_client, REGISTRATION_TOKEN, True)1023 mock_cfn_client.describe_type_registration.assert_called_once_with(1024 RegistrationToken=REGISTRATION_TOKEN1025 )1026 mock_cfn_client.set_type_default_version.assert_not_called()1027 mock_waiter.wait.assert_called_once_with(RegistrationToken=REGISTRATION_TOKEN)1028def test__wait_for_registration_waiter_fails_describe_fails(project):1029 mock_cfn_client = MagicMock(1030 spec=["describe_type_registration", "set_type_default_version", "get_waiter"]1031 )1032 mock_cfn_client.describe_type_registration.side_effect = ClientError(1033 BLANK_CLIENT_ERROR, "DescribeTypeRegistration"1034 )1035 mock_waiter = MagicMock(spec=["wait"])1036 mock_waiter.wait.side_effect = WaiterError(1037 "TypeRegistrationComplete",1038 "Waiter encountered a terminal failure state",1039 DESCRIBE_TYPE_FAILED_RETURN,1040 )1041 mock_cfn_client.get_waiter.return_value = mock_waiter1042 with pytest.raises(DownstreamError):1043 project._wait_for_registration(mock_cfn_client, REGISTRATION_TOKEN, False)1044 mock_cfn_client.describe_type_registration.assert_called_once_with(1045 RegistrationToken=REGISTRATION_TOKEN1046 )1047 mock_cfn_client.set_type_default_version.assert_not_called()1048 mock_waiter.wait.assert_called_once_with(RegistrationToken=REGISTRATION_TOKEN)1049def test__write_settings_invalid_runtime(project):1050 project.runtime = "foo"1051 project.language = LANGUAGE1052 with pytest.raises(InternalError):1053 project.write_settings()1054@pytest.mark.parametrize(1055 "docs_schema",1056 (1057 {},1058 {"primaryIdentifier": ["/properties/Id1", "/properties/Id1"]},1059 {"primaryIdentifier": ["/properties/Nested/Id1"]},1060 ),1061)1062def test__get_docs_primary_identifier_bad_path(docs_schema):1063 ref = Project._get_docs_primary_identifier(docs_schema)1064 assert ref is None1065def test__get_docs_primary_identifier_good_path():1066 ref = Project._get_docs_primary_identifier(1067 {"primaryIdentifier": ["/properties/Id1"]}1068 )1069 assert ref == "Id1"1070def test__get_docs_gettable_atts_empty():1071 getatt = Project._get_docs_gettable_atts({})1072 assert getatt == []1073@pytest.mark.parametrize(1074 "docs_schema",1075 (1076 {"readOnlyProperties": ["/properties/Id2"]},1077 {"properties": {}, "readOnlyProperties": ["/properties/Id2"]},1078 {"properties": {"Id2": {}}, "readOnlyProperties": ["/properties/Id2"]},1079 ),1080)1081def test__get_docs_gettable_atts_bad_path(docs_schema):1082 getatt = Project._get_docs_gettable_atts(docs_schema)1083 assert getatt == [1084 {"name": "Id2", "description": "Returns the <code>Id2</code> value."}1085 ]1086def test__get_docs_gettable_atts_good_path():1087 getatt = Project._get_docs_gettable_atts(1088 {1089 "properties": {"Id2": {"description": "foo"}},1090 "readOnlyProperties": ["/properties/Id2"],1091 }1092 )1093 assert getatt == [{"name": "Id2", "description": "foo"}]1094def test_generate_image_build_config(project):1095 project.schema = {}1096 mock_plugin = MagicMock(spec=["generate_image_build_config"])1097 with patch.object(project, "_plugin", mock_plugin):1098 project.generate_image_build_config()1099 mock_plugin.generate_image_build_config.assert_called_once()1100def test_generate_image_build_config_plugin_not_supported(project):1101 project.schema = {}1102 mock_plugin = MagicMock(spec=[])1103 with patch.object(project, "_plugin", mock_plugin):1104 try:1105 project.generate_image_build_config()1106 except InvalidProjectError:1107 pass1108def test__write_settings_null_executable_entrypoint(project):1109 project.type_name = TYPE_NAME1110 project.artifact_type = ARTIFACT_TYPE_RESOURCE1111 project.runtime = RUNTIME1112 project.language = LANGUAGE1113 project.executable_entrypoint = None1114 project.write_settings()1115 with project.settings_path.open("r", encoding="utf-8") as f:1116 settings = json.load(f)1117 assert "executableEntrypoint" not in settings1118def test__write_settings_nonnull_executable_entrypoint(project):1119 project.type_name = TYPE_NAME1120 project.artifact_type = ARTIFACT_TYPE_RESOURCE1121 project.runtime = RUNTIME1122 project.language = LANGUAGE1123 project.executable_entrypoint = "executable_entrypoint"1124 project.write_settings()1125 with project.settings_path.open("r", encoding="utf-8") as f:1126 settings = json.load(f)1127 assert "executableEntrypoint" in settings...
lambda_function.py
Source:lambda_function.py
...49 retries -= 150 logger.error(e, exc_info=True)51 sleep(60)52 if version_arn:53 cfn.set_type_default_version(Arn=version_arn)54 return version_arn55def delete_oldest(name):56 versions = cfn.list_type_versions(Type='RESOURCE', TypeName=name)['TypeVersionSummaries']57 if len(versions) < 2:58 return59 try:60 try:61 cfn.deregister_type(Arn=versions[0]['Arn'])62 except cfn.exceptions.CFNRegistryException as e:63 if "is the default version" not in str(e):64 raise65 cfn.deregister_type(Arn=versions[1]['Arn'])66 except cfn.exceptions.TypeNotFoundException:67 print("version already deleted...")68@helper.delete69def delete(event, _):70 if not event['PhysicalResourceId'].startswith("arn:"):71 print("no valid arn to delete")72 return73 retries = 074 while True:75 try:76 try:77 cfn.deregister_type(Arn=event['PhysicalResourceId'])78 except cfn.exceptions.CFNRegistryException as e:79 if "is the default version" not in str(e):80 raise81 versions = cfn.list_type_versions(Type='RESOURCE', TypeName=event['ResourceProperties']['TypeName'])['TypeVersionSummaries']82 if len(versions) > 1:83 versions = [v['Arn'] for v in versions if v['Arn'] != event['PhysicalResourceId']]84 versions.sort(reverse=True)85 cfn.set_type_default_version(Arn=versions[0])86 cfn.deregister_type(Arn=event['PhysicalResourceId'])87 else:88 cfn.deregister_type(Type='RESOURCE', TypeName=event['ResourceProperties']['TypeName'])89 return90 except cfn.exceptions.TypeNotFoundException:91 print("type already deleted...")92 return93 except Exception as e:94 retries += 195 if retries > 5:96 raise97 logger.error(e, exc_info=True)98 sleep(random.choice([1, 2, 3, 4, 5]))99def lambda_handler(event, context):...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
