How to use create_function_version method in localstack

Best Python code snippet using localstack_python

test_function_versions.py

Source:test_function_versions.py Github

copy

Full Screen

...23 self.wait_runtime_available(self.runtime_id)24 @decorators.idempotent_id('ce630c59-a79d-4b2d-89af-c7c5c8f8bd3f')25 def test_create(self):26 function_id = self.create_function()27 new_version = self.create_function_version(function_id)28 self.assertEqual(1, new_version)29 resp, body = self.client.get_resources(30 'functions/%s/versions' % function_id)31 self.assertEqual(200, resp.status)32 self.assertIn(33 new_version,34 [v['version_number'] for v in body['function_versions']]35 )36 @decorators.idempotent_id('9da2d24c-2ce4-4e6f-9e44-74ef1b9ec3cc')37 def test_create_function_no_change(self):38 function_id = self.create_function()39 self.create_function_version(function_id)40 self.assertRaises(41 exceptions.Forbidden,42 self.client.create_function_version,43 function_id44 )45 @decorators.idempotent_id('6864d134-fbb9-4738-9721-b541c4362789')46 def test_create_function_change(self):47 function_id = self.create_function()48 version_1 = self.create_function_version(function_id)49 self.update_function_package(function_id,50 "python/test_python_sleep.py")51 version_2 = self.create_function_version(function_id)52 self.assertGreater(version_2, version_1)53 resp, body = self.client.get_resources(54 'functions/%s/versions' % function_id)55 self.assertEqual(200, resp.status)56 numbers = [v['version_number'] for v in body['function_versions']]57 self.assertIn(version_1, numbers)58 self.assertIn(version_2, numbers)59 @decorators.idempotent_id('3f735ed4-64b0-4ec3-8bf2-507e38dcea19')60 def test_create_admin_not_allowed(self):61 """test_create_admin_not_allowed62 Even admin user can not create function version for normal user's63 function.64 """65 function_id = self.create_function()66 self.assertRaises(67 exceptions.NotFound,68 self.admin_client.create_function_version,69 function_id70 )71 # @decorators.idempotent_id('78dc5552-fcb8-4b27-86f7-5f3d96143934')72 # def test_create_version_lock_failed(self):73 # """test_create_version_lock_failed74 #75 # Creating a function requires a lock. If qinling failed to acquire the76 # lock then an error would be returned after some retries.77 #78 # In this test we acquire the lock manually, so that qinling will fail79 # to acquire the lock.80 # """81 # function_id = self.create_function()82 #83 # from qinling_tempest_plugin.tests import utils84 # etcd3_client = utils.get_etcd_client()85 # lock_id = "function_version_%s" % function_id86 # with etcd3_client.lock(id=lock_id):87 # self.assertRaises(88 # exceptions.ServerFault,89 # self.client.create_function_version,90 # function_id91 # )92 @decorators.idempotent_id('43c06f41-d116-43a7-a61c-115f7591b22e')93 def test_get_by_admin(self):94 """Admin user can get normal user's function version."""95 function_id = self.create_function()96 version = self.create_function_version(function_id)97 resp, body = self.admin_client.get_function_version(function_id,98 version)99 self.assertEqual(200, resp.status)100 self.assertEqual(version, body.get("version_number"))101 @decorators.idempotent_id('e6b865d8-ffa8-4cfc-8afb-820c64f9b2af')102 def test_get_all_by_admin(self):103 """Admin user can list normal user's function version."""104 function_id = self.create_function()105 version = self.create_function_version(function_id)106 resp, body = self.admin_client.get_function_versions(function_id)107 self.assertEqual(200, resp.status)108 self.assertIn(109 version,110 [v['version_number'] for v in body['function_versions']]111 )112 @decorators.idempotent_id('0e70ef18-687c-4ce4-ae29-aee2f88b4b9c')113 def test_delete(self):114 function_id = self.create_function()115 version = self.create_function_version(function_id)116 resp = self.client.delete_function_version(function_id, version)117 self.assertEqual(204, resp.status)118 resp, body = self.client.get_function_versions(function_id)119 self.assertEqual(200, resp.status)120 self.assertNotIn(121 version,122 [v['version_number'] for v in body['function_versions']]123 )124 @decorators.idempotent_id('c6717e2e-e80a-43d9-a25b-84f4b7453c76')125 def test_delete_by_admin(self):126 """test_delete_by_admin127 Admin user can not delete normal user's function version.128 """129 function_id = self.create_function()130 version = self.create_function_version(function_id)131 self.assertRaises(132 exceptions.NotFound,133 self.admin_client.delete_function_version,134 function_id,135 version136 )137 @decorators.idempotent_id('7898f89f-a490-42a3-8cf7-63cbd9543a06')138 def test_detach(self):139 """Admin only operation."""140 function_id = self.create_function()141 version = self.create_function_version(function_id)142 # Create execution to allocate worker143 resp, _ = self.client.create_execution(144 function_id, input='{"name": "Qinling"}', version=version145 )146 self.assertEqual(201, resp.status)147 resp, body = self.admin_client.get_function_workers(function_id,148 version=version)149 self.assertEqual(200, resp.status)150 self.assertEqual(1, len(body['workers']))151 # Detach function version from workers152 resp, _ = self.admin_client.detach_function(function_id,153 version=version)154 self.assertEqual(202, resp.status)155 def _assert_workers():...

Full Screen

Full Screen

direct.py

Source:direct.py Github

copy

Full Screen

...67def to_readable_size(size):68 for s, suffix in SUFFIXES.items():69 if size / s > 1:70 return f"{size / s:.3f}{suffix}"71def create_function_version(yc, config, config_filename):72 click.echo("Preparing package...")73 package_dir = prepare_package(config["requirements_file"],74 config["excluded_paths"],75 config_filename,76 )77 archive_path = make_archive(package_dir, 'zip', package_dir)78 archive_size = os.path.getsize(archive_path)79 try:80 if archive_size > MAX_DIRECT_ARCHIVE_SIZE:81 raise ClickException(82 f"Sorry. Looks like archive size ({to_readable_size(archive_size)}) is over the limit ({to_readable_size(MAX_DIRECT_ARCHIVE_SIZE)})."83 " Try deploying through s3 ($yappa deploy s3)")84 click.echo(f"Creating new function version for "85 + click.style(config["project_slug"], bold=True)86 + f" ({to_readable_size(archive_size)})")87 with open(archive_path, "rb") as f:88 content = f.read()89 yc.create_function_version(90 config["project_slug"],91 runtime=config["runtime"],92 description=config["description"],93 content=content,94 entrypoint=get_yc_entrypoint(config["application_type"],95 config["entrypoint"]),96 memory=config["memory_limit"],97 service_account_id=config["service_account_id"],98 timeout=config["timeout"],99 named_service_accounts=config["named_service_accounts"],100 environment=config["environment"],101 )102 click.echo(f"Created function version")103 if config["django_settings_module"]:104 click.echo("Creating new function version for management"105 " commands")106 yc.create_function_version(107 config["manage_function_name"],108 runtime=config["runtime"],109 description=config["description"],110 content=content,111 entrypoint=get_yc_entrypoint("manage",112 config["entrypoint"]),113 memory=config["memory_limit"],114 service_account_id=config["service_account_id"],115 timeout=60 * 10,116 named_service_accounts=config["named_service_accounts"],117 environment=config["environment"],118 )119 finally:120 os.remove(archive_path)...

Full Screen

Full Screen

function_service.py

Source:function_service.py Github

copy

Full Screen

...38 return response39 function_id = response.json()['metadata']['functionId']40 await asyncio.sleep(settings.CREATE_FUNCTION_TIMEOUT)41 logger.info('creating new function version...')42 response = await self.create_function_version(43 function_id=function_id,44 runtime=runtime,45 description=version_description,46 function_entrypoint=function_entrypoint,47 memory=memory,48 execution_timeout=execution_timeout,49 service_account_id=service_account_id,50 source_dir=source_dir,51 environment=environment52 )53 logger.info('done!')54 return response55 async def get_function_by_name(self, folder_id: str, name: str) -> Optional[dict]:56 params = {57 'folder_id': folder_id,58 'filter': f'name="{name}"'59 }60 async with AsyncClient() as client:61 response = await client.get(url=f'{settings.BASE_URL}/functions', params=params, auth=self._auth)62 functions = response.json().get('functions', None)63 return functions[0] if functions else None64 async def create_function(self, folder_id: str, name: str, description: str) -> Response:65 data = {66 'folderId': folder_id,67 'name': name,68 'description': description69 }70 async with AsyncClient() as client:71 response = await client.post(url=f'{settings.BASE_URL}/functions', json=data, auth=self._auth)72 logger.debug(response.text)73 return response74 async def create_function_version(75 self,76 function_id: str,77 runtime: str,78 description: str,79 function_entrypoint: str,80 memory: int,81 execution_timeout: int,82 service_account_id: str,83 source_dir: str,84 environment: Optional[dict[str, str]] = None85 ) -> Response:86 data = {87 'function_id': function_id,88 'runtime': runtime,...

Full Screen

Full Screen

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:

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!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful