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

...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():...

direct.py

Source:direct.py

...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)...

function_service.py

Source:function_service.py

...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,...

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.