Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use invoke_function method in localstack

Best Python code snippet using localstack_python

trigger_pb2.pyi

Source:trigger_pb2.pyi

...193        INVOKE_CONTAINER_WITH_RETRY_FIELD_NUMBER: builtins.int194        cron_expression: builtins.str195        """Description of a schedule as a [cron expression](/docs/functions/concepts/trigger/timer)."""196        @property197        def invoke_function(self) -> global___InvokeFunctionOnce:198            """Instructions for invoking a function once."""199        @property200        def invoke_function_with_retry(self) -> global___InvokeFunctionWithRetry:201            """Instructions for invoking a function with retry."""202        @property203        def invoke_container_with_retry(self) -> global___InvokeContainerWithRetry:204            """Instructions for invoking a container with retry."""205        def __init__(206            self,207            *,208            cron_expression: builtins.str = ...,209            invoke_function: global___InvokeFunctionOnce | None = ...,210            invoke_function_with_retry: global___InvokeFunctionWithRetry | None = ...,211            invoke_container_with_retry: global___InvokeContainerWithRetry | None = ...,212        ) -> None: ...213        def HasField(self, field_name: typing_extensions.Literal["action", b"action", "invoke_container_with_retry", b"invoke_container_with_retry", "invoke_function", b"invoke_function", "invoke_function_with_retry", b"invoke_function_with_retry"]) -> builtins.bool: ...214        def ClearField(self, field_name: typing_extensions.Literal["action", b"action", "cron_expression", b"cron_expression", "invoke_container_with_retry", b"invoke_container_with_retry", "invoke_function", b"invoke_function", "invoke_function_with_retry", b"invoke_function_with_retry"]) -> None: ...215        def WhichOneof(self, oneof_group: typing_extensions.Literal["action", b"action"]) -> typing_extensions.Literal["invoke_function", "invoke_function_with_retry", "invoke_container_with_retry"] | None: ...216    class MessageQueue(google.protobuf.message.Message):217        """Rule for activating a message queue trigger."""218        DESCRIPTOR: google.protobuf.descriptor.Descriptor219        QUEUE_ID_FIELD_NUMBER: builtins.int220        SERVICE_ACCOUNT_ID_FIELD_NUMBER: builtins.int221        BATCH_SETTINGS_FIELD_NUMBER: builtins.int222        VISIBILITY_TIMEOUT_FIELD_NUMBER: builtins.int223        INVOKE_FUNCTION_FIELD_NUMBER: builtins.int224        INVOKE_CONTAINER_FIELD_NUMBER: builtins.int225        queue_id: builtins.str226        """ID of the message queue in Message Queue."""227        service_account_id: builtins.str228        """ID of the service account which has read access to the message queue."""229        @property230        def batch_settings(self) -> global___BatchSettings:231            """Batch settings for processing messages in the queue."""232        @property233        def visibility_timeout(self) -> google.protobuf.duration_pb2.Duration:234            """Queue visibility timeout override."""235        @property236        def invoke_function(self) -> global___InvokeFunctionOnce:237            """Instructions for invoking a function once."""238        @property239        def invoke_container(self) -> global___InvokeContainerOnce:240            """Instructions for invoking a container once."""241        def __init__(242            self,243            *,244            queue_id: builtins.str = ...,245            service_account_id: builtins.str = ...,246            batch_settings: global___BatchSettings | None = ...,247            visibility_timeout: google.protobuf.duration_pb2.Duration | None = ...,248            invoke_function: global___InvokeFunctionOnce | None = ...,249            invoke_container: global___InvokeContainerOnce | None = ...,250        ) -> None: ...251        def HasField(self, field_name: typing_extensions.Literal["action", b"action", "batch_settings", b"batch_settings", "invoke_container", b"invoke_container", "invoke_function", b"invoke_function", "visibility_timeout", b"visibility_timeout"]) -> builtins.bool: ...252        def ClearField(self, field_name: typing_extensions.Literal["action", b"action", "batch_settings", b"batch_settings", "invoke_container", b"invoke_container", "invoke_function", b"invoke_function", "queue_id", b"queue_id", "service_account_id", b"service_account_id", "visibility_timeout", b"visibility_timeout"]) -> None: ...253        def WhichOneof(self, oneof_group: typing_extensions.Literal["action", b"action"]) -> typing_extensions.Literal["invoke_function", "invoke_container"] | None: ...254    class IoTMessage(google.protobuf.message.Message):255        """Rule for activating a IoT Core trigger."""256        DESCRIPTOR: google.protobuf.descriptor.Descriptor257        REGISTRY_ID_FIELD_NUMBER: builtins.int258        DEVICE_ID_FIELD_NUMBER: builtins.int259        MQTT_TOPIC_FIELD_NUMBER: builtins.int260        INVOKE_FUNCTION_FIELD_NUMBER: builtins.int261        INVOKE_CONTAINER_FIELD_NUMBER: builtins.int262        registry_id: builtins.str263        """ID of the IoT Core registry."""264        device_id: builtins.str265        """ID of the IoT Core device in the registry."""266        mqtt_topic: builtins.str267        """MQTT topic whose messages activate the trigger."""268        @property269        def invoke_function(self) -> global___InvokeFunctionWithRetry:270            """Instructions for invoking a function with retries as needed."""271        @property272        def invoke_container(self) -> global___InvokeContainerWithRetry:273            """Instructions for invoking a container with retries as needed."""274        def __init__(275            self,276            *,277            registry_id: builtins.str = ...,278            device_id: builtins.str = ...,279            mqtt_topic: builtins.str = ...,280            invoke_function: global___InvokeFunctionWithRetry | None = ...,281            invoke_container: global___InvokeContainerWithRetry | None = ...,282        ) -> None: ...283        def HasField(self, field_name: typing_extensions.Literal["action", b"action", "invoke_container", b"invoke_container", "invoke_function", b"invoke_function"]) -> builtins.bool: ...284        def ClearField(self, field_name: typing_extensions.Literal["action", b"action", "device_id", b"device_id", "invoke_container", b"invoke_container", "invoke_function", b"invoke_function", "mqtt_topic", b"mqtt_topic", "registry_id", b"registry_id"]) -> None: ...285        def WhichOneof(self, oneof_group: typing_extensions.Literal["action", b"action"]) -> typing_extensions.Literal["invoke_function", "invoke_container"] | None: ...286    class IoTBrokerMessage(google.protobuf.message.Message):287        """Rule for activating a IoT Core Broker trigger."""288        DESCRIPTOR: google.protobuf.descriptor.Descriptor289        BROKER_ID_FIELD_NUMBER: builtins.int290        MQTT_TOPIC_FIELD_NUMBER: builtins.int291        INVOKE_FUNCTION_FIELD_NUMBER: builtins.int292        INVOKE_CONTAINER_FIELD_NUMBER: builtins.int293        broker_id: builtins.str294        """ID of the IoT Core broker."""295        mqtt_topic: builtins.str296        """MQTT topic whose messages activate the trigger."""297        @property298        def invoke_function(self) -> global___InvokeFunctionWithRetry:299            """Instructions for invoking a function with retries as needed."""300        @property301        def invoke_container(self) -> global___InvokeContainerWithRetry:302            """Instructions for invoking a container with retries as needed."""303        def __init__(304            self,305            *,306            broker_id: builtins.str = ...,307            mqtt_topic: builtins.str = ...,308            invoke_function: global___InvokeFunctionWithRetry | None = ...,309            invoke_container: global___InvokeContainerWithRetry | None = ...,310        ) -> None: ...311        def HasField(self, field_name: typing_extensions.Literal["action", b"action", "invoke_container", b"invoke_container", "invoke_function", b"invoke_function"]) -> builtins.bool: ...312        def ClearField(self, field_name: typing_extensions.Literal["action", b"action", "broker_id", b"broker_id", "invoke_container", b"invoke_container", "invoke_function", b"invoke_function", "mqtt_topic", b"mqtt_topic"]) -> None: ...313        def WhichOneof(self, oneof_group: typing_extensions.Literal["action", b"action"]) -> typing_extensions.Literal["invoke_function", "invoke_container"] | None: ...314    class ObjectStorage(google.protobuf.message.Message):315        DESCRIPTOR: google.protobuf.descriptor.Descriptor316        EVENT_TYPE_FIELD_NUMBER: builtins.int317        BUCKET_ID_FIELD_NUMBER: builtins.int318        PREFIX_FIELD_NUMBER: builtins.int319        SUFFIX_FIELD_NUMBER: builtins.int320        INVOKE_FUNCTION_FIELD_NUMBER: builtins.int321        INVOKE_CONTAINER_FIELD_NUMBER: builtins.int322        @property323        def event_type(self) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[global___Trigger.ObjectStorageEventType.ValueType]:324            """Type (name) of events, at least one value is required."""325        bucket_id: builtins.str326        """ID of the bucket."""327        prefix: builtins.str328        """Prefix of the object key. Filter, optional."""329        suffix: builtins.str330        """Suffix of the object key. Filter, optional."""331        @property332        def invoke_function(self) -> global___InvokeFunctionWithRetry:333            """Instructions for invoking a function with retries as needed."""334        @property335        def invoke_container(self) -> global___InvokeContainerWithRetry:336            """Instructions for invoking a container with retries as needed."""337        def __init__(338            self,339            *,340            event_type: collections.abc.Iterable[global___Trigger.ObjectStorageEventType.ValueType] | None = ...,341            bucket_id: builtins.str = ...,342            prefix: builtins.str = ...,343            suffix: builtins.str = ...,344            invoke_function: global___InvokeFunctionWithRetry | None = ...,345            invoke_container: global___InvokeContainerWithRetry | None = ...,346        ) -> None: ...347        def HasField(self, field_name: typing_extensions.Literal["action", b"action", "invoke_container", b"invoke_container", "invoke_function", b"invoke_function"]) -> builtins.bool: ...348        def ClearField(self, field_name: typing_extensions.Literal["action", b"action", "bucket_id", b"bucket_id", "event_type", b"event_type", "invoke_container", b"invoke_container", "invoke_function", b"invoke_function", "prefix", b"prefix", "suffix", b"suffix"]) -> None: ...349        def WhichOneof(self, oneof_group: typing_extensions.Literal["action", b"action"]) -> typing_extensions.Literal["invoke_function", "invoke_container"] | None: ...350    class ContainerRegistry(google.protobuf.message.Message):351        DESCRIPTOR: google.protobuf.descriptor.Descriptor352        EVENT_TYPE_FIELD_NUMBER: builtins.int353        REGISTRY_ID_FIELD_NUMBER: builtins.int354        IMAGE_NAME_FIELD_NUMBER: builtins.int355        TAG_FIELD_NUMBER: builtins.int356        INVOKE_FUNCTION_FIELD_NUMBER: builtins.int357        INVOKE_CONTAINER_FIELD_NUMBER: builtins.int358        @property359        def event_type(self) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[global___Trigger.ContainerRegistryEventType.ValueType]:360            """Type (name) of events, at least one value is required."""361        registry_id: builtins.str362        """ID of the registry."""363        image_name: builtins.str364        """Docker-image name. Filter, optional."""365        tag: builtins.str366        """Docker-image tag. Filter, optional."""367        @property368        def invoke_function(self) -> global___InvokeFunctionWithRetry:369            """Instructions for invoking a function with retries as needed."""370        @property371        def invoke_container(self) -> global___InvokeContainerWithRetry:372            """Instructions for invoking a container with retries as needed."""373        def __init__(374            self,375            *,376            event_type: collections.abc.Iterable[global___Trigger.ContainerRegistryEventType.ValueType] | None = ...,377            registry_id: builtins.str = ...,378            image_name: builtins.str = ...,379            tag: builtins.str = ...,380            invoke_function: global___InvokeFunctionWithRetry | None = ...,381            invoke_container: global___InvokeContainerWithRetry | None = ...,382        ) -> None: ...383        def HasField(self, field_name: typing_extensions.Literal["action", b"action", "invoke_container", b"invoke_container", "invoke_function", b"invoke_function"]) -> builtins.bool: ...384        def ClearField(self, field_name: typing_extensions.Literal["action", b"action", "event_type", b"event_type", "image_name", b"image_name", "invoke_container", b"invoke_container", "invoke_function", b"invoke_function", "registry_id", b"registry_id", "tag", b"tag"]) -> None: ...385        def WhichOneof(self, oneof_group: typing_extensions.Literal["action", b"action"]) -> typing_extensions.Literal["invoke_function", "invoke_container"] | None: ...386    class CloudLogs(google.protobuf.message.Message):387        DESCRIPTOR: google.protobuf.descriptor.Descriptor388        LOG_GROUP_ID_FIELD_NUMBER: builtins.int389        BATCH_SETTINGS_FIELD_NUMBER: builtins.int390        INVOKE_FUNCTION_FIELD_NUMBER: builtins.int391        INVOKE_CONTAINER_FIELD_NUMBER: builtins.int392        @property393        def log_group_id(self) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[builtins.str]:394            """Log group identifiers, at least one value is required."""395        @property396        def batch_settings(self) -> global___CloudLogsBatchSettings:397            """Batch settings for processing log events."""398        @property399        def invoke_function(self) -> global___InvokeFunctionWithRetry:400            """Instructions for invoking a function with retries as needed."""401        @property402        def invoke_container(self) -> global___InvokeContainerWithRetry:403            """Instructions for invoking a container with retries as needed."""404        def __init__(405            self,406            *,407            log_group_id: collections.abc.Iterable[builtins.str] | None = ...,408            batch_settings: global___CloudLogsBatchSettings | None = ...,409            invoke_function: global___InvokeFunctionWithRetry | None = ...,410            invoke_container: global___InvokeContainerWithRetry | None = ...,411        ) -> None: ...412        def HasField(self, field_name: typing_extensions.Literal["action", b"action", "batch_settings", b"batch_settings", "invoke_container", b"invoke_container", "invoke_function", b"invoke_function"]) -> builtins.bool: ...413        def ClearField(self, field_name: typing_extensions.Literal["action", b"action", "batch_settings", b"batch_settings", "invoke_container", b"invoke_container", "invoke_function", b"invoke_function", "log_group_id", b"log_group_id"]) -> None: ...414        def WhichOneof(self, oneof_group: typing_extensions.Literal["action", b"action"]) -> typing_extensions.Literal["invoke_function", "invoke_container"] | None: ...415    class Logging(google.protobuf.message.Message):416        DESCRIPTOR: google.protobuf.descriptor.Descriptor417        LOG_GROUP_ID_FIELD_NUMBER: builtins.int418        RESOURCE_TYPE_FIELD_NUMBER: builtins.int419        RESOURCE_ID_FIELD_NUMBER: builtins.int420        LEVELS_FIELD_NUMBER: builtins.int421        BATCH_SETTINGS_FIELD_NUMBER: builtins.int422        INVOKE_FUNCTION_FIELD_NUMBER: builtins.int423        INVOKE_CONTAINER_FIELD_NUMBER: builtins.int424        log_group_id: builtins.str425        """Log events filter settings."""426        @property427        def resource_type(self) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[builtins.str]: ...428        @property429        def resource_id(self) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[builtins.str]: ...430        @property431        def levels(self) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[yandex.cloud.logging.v1.log_entry_pb2.LogLevel.Level.ValueType]: ...432        @property433        def batch_settings(self) -> global___LoggingBatchSettings:434            """Batch settings for processing log events."""435        @property436        def invoke_function(self) -> global___InvokeFunctionWithRetry:437            """Instructions for invoking a function with retries as needed."""438        @property439        def invoke_container(self) -> global___InvokeContainerWithRetry:440            """Instructions for invoking a container with retries as needed."""441        def __init__(442            self,443            *,444            log_group_id: builtins.str = ...,445            resource_type: collections.abc.Iterable[builtins.str] | None = ...,446            resource_id: collections.abc.Iterable[builtins.str] | None = ...,447            levels: collections.abc.Iterable[yandex.cloud.logging.v1.log_entry_pb2.LogLevel.Level.ValueType] | None = ...,448            batch_settings: global___LoggingBatchSettings | None = ...,449            invoke_function: global___InvokeFunctionWithRetry | None = ...,450            invoke_container: global___InvokeContainerWithRetry | None = ...,451        ) -> None: ...452        def HasField(self, field_name: typing_extensions.Literal["action", b"action", "batch_settings", b"batch_settings", "invoke_container", b"invoke_container", "invoke_function", b"invoke_function"]) -> builtins.bool: ...453        def ClearField(self, field_name: typing_extensions.Literal["action", b"action", "batch_settings", b"batch_settings", "invoke_container", b"invoke_container", "invoke_function", b"invoke_function", "levels", b"levels", "log_group_id", b"log_group_id", "resource_id", b"resource_id", "resource_type", b"resource_type"]) -> None: ...454        def WhichOneof(self, oneof_group: typing_extensions.Literal["action", b"action"]) -> typing_extensions.Literal["invoke_function", "invoke_container"] | None: ...455    ID_FIELD_NUMBER: builtins.int456    FOLDER_ID_FIELD_NUMBER: builtins.int457    CREATED_AT_FIELD_NUMBER: builtins.int458    NAME_FIELD_NUMBER: builtins.int459    DESCRIPTION_FIELD_NUMBER: builtins.int460    LABELS_FIELD_NUMBER: builtins.int461    RULE_FIELD_NUMBER: builtins.int462    STATUS_FIELD_NUMBER: builtins.int463    id: builtins.str464    """ID of the trigger. Generated at creation time."""465    folder_id: builtins.str466    """ID of the folder that the trigger belongs to."""467    @property468    def created_at(self) -> google.protobuf.timestamp_pb2.Timestamp:469        """Creation timestamp for the trigger."""470    name: builtins.str471    """Name of the trigger."""472    description: builtins.str473    """Description of the trigger."""474    @property475    def labels(self) -> google.protobuf.internal.containers.ScalarMap[builtins.str, builtins.str]:476        """Trigger labels as `key:value` pairs."""477    @property478    def rule(self) -> global___Trigger.Rule:479        """Rule for trigger activation (always consistent with the trigger type)."""480    status: global___Trigger.Status.ValueType481    """Trigger status."""482    def __init__(483        self,484        *,485        id: builtins.str = ...,486        folder_id: builtins.str = ...,487        created_at: google.protobuf.timestamp_pb2.Timestamp | None = ...,488        name: builtins.str = ...,489        description: builtins.str = ...,490        labels: collections.abc.Mapping[builtins.str, builtins.str] | None = ...,491        rule: global___Trigger.Rule | None = ...,492        status: global___Trigger.Status.ValueType = ...,493    ) -> None: ...494    def HasField(self, field_name: typing_extensions.Literal["created_at", b"created_at", "rule", b"rule"]) -> builtins.bool: ...495    def ClearField(self, field_name: typing_extensions.Literal["created_at", b"created_at", "description", b"description", "folder_id", b"folder_id", "id", b"id", "labels", b"labels", "name", b"name", "rule", b"rule", "status", b"status"]) -> None: ...496global___Trigger = Trigger497class InvokeFunctionOnce(google.protobuf.message.Message):498    """A single function invocation."""499    DESCRIPTOR: google.protobuf.descriptor.Descriptor500    FUNCTION_ID_FIELD_NUMBER: builtins.int501    FUNCTION_TAG_FIELD_NUMBER: builtins.int502    SERVICE_ACCOUNT_ID_FIELD_NUMBER: builtins.int503    function_id: builtins.str504    """ID of the function to invoke."""505    function_tag: builtins.str506    """Version tag of the function to execute."""507    service_account_id: builtins.str508    """ID of the service account that should be used to invoke the function."""509    def __init__(510        self,511        *,512        function_id: builtins.str = ...,513        function_tag: builtins.str = ...,514        service_account_id: builtins.str = ...,515    ) -> None: ...516    def ClearField(self, field_name: typing_extensions.Literal["function_id", b"function_id", "function_tag", b"function_tag", "service_account_id", b"service_account_id"]) -> None: ...517global___InvokeFunctionOnce = InvokeFunctionOnce518class InvokeFunctionWithRetry(google.protobuf.message.Message):519    """A function invocation with retries."""520    DESCRIPTOR: google.protobuf.descriptor.Descriptor521    FUNCTION_ID_FIELD_NUMBER: builtins.int522    FUNCTION_TAG_FIELD_NUMBER: builtins.int523    SERVICE_ACCOUNT_ID_FIELD_NUMBER: builtins.int524    RETRY_SETTINGS_FIELD_NUMBER: builtins.int525    DEAD_LETTER_QUEUE_FIELD_NUMBER: builtins.int526    function_id: builtins.str527    """ID of the function to invoke."""528    function_tag: builtins.str529    """Version tag of the function to execute."""530    service_account_id: builtins.str531    """ID of the service account which has permission to invoke the function."""532    @property533    def retry_settings(self) -> global___RetrySettings:534        """Retry policy. If the field is not specified, or the value is empty, no retries will be attempted."""535    @property536    def dead_letter_queue(self) -> global___PutQueueMessage:537        """DLQ policy (no value means discarding a message)."""538    def __init__(539        self,540        *,541        function_id: builtins.str = ...,542        function_tag: builtins.str = ...,543        service_account_id: builtins.str = ...,544        retry_settings: global___RetrySettings | None = ...,545        dead_letter_queue: global___PutQueueMessage | None = ...,546    ) -> None: ...547    def HasField(self, field_name: typing_extensions.Literal["dead_letter_queue", b"dead_letter_queue", "retry_settings", b"retry_settings"]) -> builtins.bool: ...548    def ClearField(self, field_name: typing_extensions.Literal["dead_letter_queue", b"dead_letter_queue", "function_id", b"function_id", "function_tag", b"function_tag", "retry_settings", b"retry_settings", "service_account_id", b"service_account_id"]) -> None: ...549global___InvokeFunctionWithRetry = InvokeFunctionWithRetry550class InvokeContainerOnce(google.protobuf.message.Message):551    """A single container invocation."""552    DESCRIPTOR: google.protobuf.descriptor.Descriptor553    CONTAINER_ID_FIELD_NUMBER: builtins.int554    PATH_FIELD_NUMBER: builtins.int555    SERVICE_ACCOUNT_ID_FIELD_NUMBER: builtins.int556    container_id: builtins.str557    """ID of the container to invoke."""558    path: builtins.str559    """Endpoint HTTP path to invoke."""560    service_account_id: builtins.str561    """ID of the service account which has permission to invoke the container."""562    def __init__(563        self,564        *,565        container_id: builtins.str = ...,566        path: builtins.str = ...,567        service_account_id: builtins.str = ...,568    ) -> None: ...569    def ClearField(self, field_name: typing_extensions.Literal["container_id", b"container_id", "path", b"path", "service_account_id", b"service_account_id"]) -> None: ...570global___InvokeContainerOnce = InvokeContainerOnce571class InvokeContainerWithRetry(google.protobuf.message.Message):572    """A container invocation with retries."""573    DESCRIPTOR: google.protobuf.descriptor.Descriptor574    CONTAINER_ID_FIELD_NUMBER: builtins.int575    PATH_FIELD_NUMBER: builtins.int576    SERVICE_ACCOUNT_ID_FIELD_NUMBER: builtins.int577    RETRY_SETTINGS_FIELD_NUMBER: builtins.int578    DEAD_LETTER_QUEUE_FIELD_NUMBER: builtins.int579    container_id: builtins.str580    """ID of the container to invoke."""581    path: builtins.str582    """Endpoint HTTP path to invoke."""583    service_account_id: builtins.str584    """ID of the service account which has permission to invoke the container."""585    @property586    def retry_settings(self) -> global___RetrySettings:587        """Retry policy. If the field is not specified, or the value is empty, no retries will be attempted."""588    @property589    def dead_letter_queue(self) -> global___PutQueueMessage:590        """DLQ policy (no value means discarding a message)."""591    def __init__(592        self,593        *,594        container_id: builtins.str = ...,595        path: builtins.str = ...,596        service_account_id: builtins.str = ...,597        retry_settings: global___RetrySettings | None = ...,598        dead_letter_queue: global___PutQueueMessage | None = ...,599    ) -> None: ...600    def HasField(self, field_name: typing_extensions.Literal["dead_letter_queue", b"dead_letter_queue", "retry_settings", b"retry_settings"]) -> builtins.bool: ...601    def ClearField(self, field_name: typing_extensions.Literal["container_id", b"container_id", "dead_letter_queue", b"dead_letter_queue", "path", b"path", "retry_settings", b"retry_settings", "service_account_id", b"service_account_id"]) -> None: ...602global___InvokeContainerWithRetry = InvokeContainerWithRetry603class PutQueueMessage(google.protobuf.message.Message):604    DESCRIPTOR: google.protobuf.descriptor.Descriptor605    QUEUE_ID_FIELD_NUMBER: builtins.int606    SERVICE_ACCOUNT_ID_FIELD_NUMBER: builtins.int607    queue_id: builtins.str608    """ID of the queue."""609    service_account_id: builtins.str610    """Service account which has write permission on the queue."""611    def __init__(612        self,613        *,614        queue_id: builtins.str = ...,615        service_account_id: builtins.str = ...,616    ) -> None: ...617    def ClearField(self, field_name: typing_extensions.Literal["queue_id", b"queue_id", "service_account_id", b"service_account_id"]) -> None: ...618global___PutQueueMessage = PutQueueMessage619class BatchSettings(google.protobuf.message.Message):620    """Settings for batch processing of messages in a queue."""621    DESCRIPTOR: google.protobuf.descriptor.Descriptor622    SIZE_FIELD_NUMBER: builtins.int623    CUTOFF_FIELD_NUMBER: builtins.int624    size: builtins.int625    """Batch size. Trigger will send the batch of messages to the function626    when the number of messages in the queue reaches [size], or the [cutoff] time has passed.627    """628    @property629    def cutoff(self) -> google.protobuf.duration_pb2.Duration:630        """Maximum wait time. Trigger will send the batch of messages to the function when631        the number of messages in the queue reaches [size], or the [cutoff] time has passed.632        """633    def __init__(634        self,635        *,636        size: builtins.int = ...,637        cutoff: google.protobuf.duration_pb2.Duration | None = ...,638    ) -> None: ...639    def HasField(self, field_name: typing_extensions.Literal["cutoff", b"cutoff"]) -> builtins.bool: ...640    def ClearField(self, field_name: typing_extensions.Literal["cutoff", b"cutoff", "size", b"size"]) -> None: ...641global___BatchSettings = BatchSettings642class CloudLogsBatchSettings(google.protobuf.message.Message):643    DESCRIPTOR: google.protobuf.descriptor.Descriptor644    SIZE_FIELD_NUMBER: builtins.int645    CUTOFF_FIELD_NUMBER: builtins.int646    size: builtins.int647    """Batch size. Trigger will send the batch of messages to the function648    when the number of messages in the log group reaches [size], or the [cutoff] time has passed.649    """650    @property651    def cutoff(self) -> google.protobuf.duration_pb2.Duration:652        """Maximum wait time. Trigger will send the batch of messages to the function when653        the number of messages in the log group reaches [size], or the [cutoff] time has passed.654        """655    def __init__(656        self,657        *,658        size: builtins.int = ...,659        cutoff: google.protobuf.duration_pb2.Duration | None = ...,660    ) -> None: ...661    def HasField(self, field_name: typing_extensions.Literal["cutoff", b"cutoff"]) -> builtins.bool: ...662    def ClearField(self, field_name: typing_extensions.Literal["cutoff", b"cutoff", "size", b"size"]) -> None: ...663global___CloudLogsBatchSettings = CloudLogsBatchSettings664class LoggingBatchSettings(google.protobuf.message.Message):665    DESCRIPTOR: google.protobuf.descriptor.Descriptor666    SIZE_FIELD_NUMBER: builtins.int667    CUTOFF_FIELD_NUMBER: builtins.int668    size: builtins.int669    """Batch size. Trigger will send the batch of messages to the associated function670    when the number of log events reaches this value, or the [cutoff] time has passed.671    """672    @property673    def cutoff(self) -> google.protobuf.duration_pb2.Duration:674        """Maximum wait time. Trigger will send the batch of messages the time since the last batch675        exceeds the `cutoff` value, regardless of the amount of log events.676        """677    def __init__(678        self,679        *,680        size: builtins.int = ...,681        cutoff: google.protobuf.duration_pb2.Duration | None = ...,682    ) -> None: ...683    def HasField(self, field_name: typing_extensions.Literal["cutoff", b"cutoff"]) -> builtins.bool: ...684    def ClearField(self, field_name: typing_extensions.Literal["cutoff", b"cutoff", "size", b"size"]) -> None: ...685global___LoggingBatchSettings = LoggingBatchSettings686class RetrySettings(google.protobuf.message.Message):687    """Settings for retrying to invoke a function."""688    DESCRIPTOR: google.protobuf.descriptor.Descriptor689    RETRY_ATTEMPTS_FIELD_NUMBER: builtins.int690    INTERVAL_FIELD_NUMBER: builtins.int691    retry_attempts: builtins.int692    """Maximum number of retries (extra invokes) before the action is considered failed."""693    @property694    def interval(self) -> google.protobuf.duration_pb2.Duration:695        """Time in seconds to wait between individual retries."""696    def __init__(697        self,698        *,699        retry_attempts: builtins.int = ...,700        interval: google.protobuf.duration_pb2.Duration | None = ...,701    ) -> None: ...702    def HasField(self, field_name: typing_extensions.Literal["interval", b"interval"]) -> builtins.bool: ...703    def ClearField(self, field_name: typing_extensions.Literal["interval", b"interval", "retry_attempts", b"retry_attempts"]) -> None: ...704global___RetrySettings = RetrySettings705class BillingBudget(google.protobuf.message.Message):706    DESCRIPTOR: google.protobuf.descriptor.Descriptor707    BILLING_ACCOUNT_ID_FIELD_NUMBER: builtins.int708    BUDGET_ID_FIELD_NUMBER: builtins.int709    INVOKE_FUNCTION_FIELD_NUMBER: builtins.int710    INVOKE_CONTAINER_FIELD_NUMBER: builtins.int711    billing_account_id: builtins.str712    budget_id: builtins.str713    @property714    def invoke_function(self) -> global___InvokeFunctionWithRetry: ...715    @property716    def invoke_container(self) -> global___InvokeContainerWithRetry: ...717    def __init__(718        self,719        *,720        billing_account_id: builtins.str = ...,721        budget_id: builtins.str = ...,722        invoke_function: global___InvokeFunctionWithRetry | None = ...,723        invoke_container: global___InvokeContainerWithRetry | None = ...,724    ) -> None: ...725    def HasField(self, field_name: typing_extensions.Literal["action", b"action", "invoke_container", b"invoke_container", "invoke_function", b"invoke_function"]) -> builtins.bool: ...726    def ClearField(self, field_name: typing_extensions.Literal["action", b"action", "billing_account_id", b"billing_account_id", "budget_id", b"budget_id", "invoke_container", b"invoke_container", "invoke_function", b"invoke_function"]) -> None: ...727    def WhichOneof(self, oneof_group: typing_extensions.Literal["action", b"action"]) -> typing_extensions.Literal["invoke_function", "invoke_container"] | None: ...728global___BillingBudget = BillingBudget729class DataStreamBatchSettings(google.protobuf.message.Message):730    DESCRIPTOR: google.protobuf.descriptor.Descriptor731    SIZE_FIELD_NUMBER: builtins.int732    CUTOFF_FIELD_NUMBER: builtins.int733    size: builtins.int734    """Batch size in bytes. Trigger will send the batch of messages to the associated function735    when size of log events reaches this value, or the [cutoff] time has passed.736    """737    @property738    def cutoff(self) -> google.protobuf.duration_pb2.Duration:739        """Maximum wait time. Trigger will send the batch of messages the time since the last batch740        exceeds the `cutoff` value, regardless of the amount of log events.741        """742    def __init__(743        self,744        *,745        size: builtins.int = ...,746        cutoff: google.protobuf.duration_pb2.Duration | None = ...,747    ) -> None: ...748    def HasField(self, field_name: typing_extensions.Literal["cutoff", b"cutoff"]) -> builtins.bool: ...749    def ClearField(self, field_name: typing_extensions.Literal["cutoff", b"cutoff", "size", b"size"]) -> None: ...750global___DataStreamBatchSettings = DataStreamBatchSettings751class DataStream(google.protobuf.message.Message):752    DESCRIPTOR: google.protobuf.descriptor.Descriptor753    ENDPOINT_FIELD_NUMBER: builtins.int754    DATABASE_FIELD_NUMBER: builtins.int755    STREAM_FIELD_NUMBER: builtins.int756    SERVICE_ACCOUNT_ID_FIELD_NUMBER: builtins.int757    BATCH_SETTINGS_FIELD_NUMBER: builtins.int758    INVOKE_FUNCTION_FIELD_NUMBER: builtins.int759    INVOKE_CONTAINER_FIELD_NUMBER: builtins.int760    endpoint: builtins.str761    """Data stream endpoint."""762    database: builtins.str763    """Data stream database."""764    stream: builtins.str765    """Stream name."""766    service_account_id: builtins.str767    """ID of the service account which has permission to read data stream."""768    @property769    def batch_settings(self) -> global___DataStreamBatchSettings:770        """Batch settings for processing events."""771    @property772    def invoke_function(self) -> global___InvokeFunctionWithRetry: ...773    @property774    def invoke_container(self) -> global___InvokeContainerWithRetry: ...775    def __init__(776        self,777        *,778        endpoint: builtins.str = ...,779        database: builtins.str = ...,780        stream: builtins.str = ...,781        service_account_id: builtins.str = ...,782        batch_settings: global___DataStreamBatchSettings | None = ...,783        invoke_function: global___InvokeFunctionWithRetry | None = ...,784        invoke_container: global___InvokeContainerWithRetry | None = ...,785    ) -> None: ...786    def HasField(self, field_name: typing_extensions.Literal["action", b"action", "batch_settings", b"batch_settings", "invoke_container", b"invoke_container", "invoke_function", b"invoke_function"]) -> builtins.bool: ...787    def ClearField(self, field_name: typing_extensions.Literal["action", b"action", "batch_settings", b"batch_settings", "database", b"database", "endpoint", b"endpoint", "invoke_container", b"invoke_container", "invoke_function", b"invoke_function", "service_account_id", b"service_account_id", "stream", b"stream"]) -> None: ...788    def WhichOneof(self, oneof_group: typing_extensions.Literal["action", b"action"]) -> typing_extensions.Literal["invoke_function", "invoke_container"] | None: ...789global___DataStream = DataStream790class Mail(google.protobuf.message.Message):791    DESCRIPTOR: google.protobuf.descriptor.Descriptor792    EMAIL_FIELD_NUMBER: builtins.int793    INVOKE_FUNCTION_FIELD_NUMBER: builtins.int794    INVOKE_CONTAINER_FIELD_NUMBER: builtins.int795    email: builtins.str796    """Address to receive emails for trigger activation.797    Field is ignored for write requests and populated on trigger creation.798    """799    @property800    def invoke_function(self) -> global___InvokeFunctionWithRetry: ...801    @property802    def invoke_container(self) -> global___InvokeContainerWithRetry: ...803    def __init__(804        self,805        *,806        email: builtins.str = ...,807        invoke_function: global___InvokeFunctionWithRetry | None = ...,808        invoke_container: global___InvokeContainerWithRetry | None = ...,809    ) -> None: ...810    def HasField(self, field_name: typing_extensions.Literal["action", b"action", "invoke_container", b"invoke_container", "invoke_function", b"invoke_function"]) -> builtins.bool: ...811    def ClearField(self, field_name: typing_extensions.Literal["action", b"action", "email", b"email", "invoke_container", b"invoke_container", "invoke_function", b"invoke_function"]) -> None: ...812    def WhichOneof(self, oneof_group: typing_extensions.Literal["action", b"action"]) -> typing_extensions.Literal["invoke_function", "invoke_container"] | None: ......

test_main.py

Source:test_main.py

...34	35	def test_01(self):36		result = False37		logger.open("test_01.log", "test_01")38		(result, response) = invoke_function(self.m_contract_address, "int_to_int", "", argvs = [{"type": "int","value": "0"}], node_index = 0)39		logger.close(result and response["result"]["Result"] == '01')40		41	def test_02(self):42		result = False43		logger.open("test_02.log", "test_02")44		(result, response) = invoke_function(self.m_contract_address, "int_to_int", "", argvs = [{"type": "int","value": "-1"}], node_index = 0)45		logger.close(result and response["result"]["Result"] == '01')46		# logger.close(not result or response["result"]["Result"] != '01')47	48	def test_03(self):49		result = False50		logger.open("test_03.log", "test_03")51		(result, response) = invoke_function(self.m_contract_address, "int_to_int", "", argvs = [{"type": "int","value": "65535"}], node_index = 0)52		logger.close(result and response["result"]["Result"] == '01')53		54	def test_04(self):55		result = False56		logger.open("test_04.log", "test_04")57		(result, response) = invoke_function(self.m_contract_address, "int_to_int", "", argvs = [{"type": "int","value": "65536"}], node_index = 0)58		logger.close(result and response["result"]["Result"] == '01')59		#logger.close(not result or response["result"]["Result"] != '01')60		61	def test_05(self):62		result = False63		logger.open("test_05.log", "test_05")64		(result, response) = invoke_function(self.m_contract_address, "int_to_int", "", argvs = [{"type": "string","value": "abc"}], node_index = 0)65		logger.close(result and response["result"]["Result"] == '01')66		# logger.close(not result or response["result"]["Result"] != '01')		67	68	def test_06(self):69		result = False70		logger.open("test_06.log", "test_06")71		(result, response) = invoke_function(self.m_contract_address, "int_to_int", "", argvs = [{"type": "string","value": ""}], node_index = 0)72		logger.close(result and response["result"]["Result"] == '01')73		#logger.close(not result or response["result"]["Result"] != '01')	74	def test_07(self):75		result = False76		logger.open("test_07.log", "test_07")77		(result, response) = invoke_function(self.m_contract_address, "int_to_int", "", argvs = [{"type": "bytearray","value": "111122223333"}], node_index = 0)78		logger.close(result and response["result"]["Result"] == '01')79		#logger.close(not result or response["result"]["Result"] != '01')	80	def test_08(self):81		result = False82		logger.open("test_08.log", "test_08")83		(result, response) = invoke_function(self.m_contract_address, "int_to_int", "", argvs = [{"type": "bool","value": "true"}], node_index = 0)84		logger.close(result and response["result"]["Result"] == '01')85		#logger.close(not result or response["result"]["Result"] != '01')	86	def test_09(self):87		result = False88		logger.open("test_09.log", "test_09")89		(result, response) = invoke_function(self.m_contract_address, "int_to_int", "", argvs = [{"type": "bool","value": "false"}], node_index = 0)90		logger.close(result and response["result"]["Result"] == '01')91		#logger.close(not result or response["result"]["Result"] != '01')	92	93	#string to int94	def test_10(self):95		result = False96		logger.open("test_10.log", "test_10")97		(result, response) = invoke_function(self.m_contract_address, "string_to_int", "", argvs = [{"type": "string","value": "qwertyuiop"}], node_index = 0)98		logger.close(result and response["result"]["Result"] == '01')	99	def test_11(self):100		result = False101		logger.open("test_11.log", "test_11")102		(result, response) = invoke_function(self.m_contract_address, "string_to_int", "", argvs = [{"type": "string","value": "~!@@#$$%^&*()_+-={}|:\"<>?;'[] \\,./"}], node_index = 0)103		logger.close(result and response["result"]["Result"] == '01')	104	def test_12(self):105		result = False106		logger.open("test_12.log", "test_12")107		(result, response) = invoke_function(self.m_contract_address, "string_to_int", "", argvs = [{"type": "string","value": "abcd1234_))()*(&^$^%#"}], node_index = 0)108		logger.close(result and response["result"]["Result"] == '01')109	110	def test_13(self):111		result = False112		logger.open("test_13.log", "test_13")113		(result, response) = invoke_function(self.m_contract_address, "string_to_int", "", argvs = [{"type": "int","value": "123"}], node_index = 0)114		logger.close(result and response["result"]["Result"] == '01')115		# logger.close(not result or response["result"]["Result"] != '01')116	def test_14(self):117		result = False118		logger.open("test_14.log", "test_14")119		(result, response) = invoke_function(self.m_contract_address, "string_to_int", "", argvs = [{"type": "bytearray","value": "111122223333"}], node_index = 0)120		logger.close(result and response["result"]["Result"] == '01')121		#logger.close(not result or response["result"]["Result"] != '01')122	def test_15(self):123		result = False124		logger.open("test_15.log", "test_15")125		(result, response) = invoke_function(self.m_contract_address, "string_to_int", "", argvs = [{"type": "bool","value": "true"}], node_index = 0)126		logger.close(result and response["result"]["Result"] == '01')127		# logger.close(not result or response["result"]["Result"] != '01')		128	def test_16(self):129		result = False130		logger.open("test_16.log", "test_16")131		(result, response) = invoke_function(self.m_contract_address, "string_to_int", "", argvs = [{"type": "bool","value": "false"}], node_index = 0)132		logger.close(result and response["result"]["Result"] == '01')133		# logger.close(not result or response["result"]["Result"] != '01')		134	def test_24(self):135		result = False136		logger.open("test_24.log", "test_24")137		(result, response) = invoke_function(self.m_contract_address, "bool_to_int", "", argvs = [{"type": "bool","value": "true"}], node_index = 0)138		logger.close(result and response["result"]["Result"] == '01')				139	def test_25(self):140		result = False141		logger.open("test_25.log", "test_25")142		(result, response) = invoke_function(self.m_contract_address, "bool_to_int", "", argvs = [{"type": "int","value": "0"}], node_index = 0)143		logger.close(result and response["result"]["Result"] == '01')				144	145	def test_26(self):146		result = False147		logger.open("test_26.log", "test_26")148		(result, response) = invoke_function(self.m_contract_address, "bool_to_int", "", argvs = [{"type": "bool","value": "false"}], node_index = 0)149		logger.close(result and response["result"]["Result"] == '01')	150		151	def test_27(self):152		result = False153		logger.open("test_27.log", "test_27")154		(result, response) = invoke_function(self.m_contract_address, "bool_to_int", "", argvs = [{"type": "int","value": "1"}], node_index = 0)155		logger.close(result and response["result"]["Result"] == '01')156		# logger.close(not result or response["result"]["Result"] != '01')157		158	def test_28(self):159		result = False160		logger.open("test_28.log", "test_28")161		(result, response) = invoke_function(self.m_contract_address, "bool_to_int", "", argvs = [{"type": "string","value": "abc"}], node_index = 0)162		logger.close(result and response["result"]["Result"] == '01')163		# logger.close(not result or response["result"]["Result"] != '01')164	165	def test_29(self):166		result = False167		logger.open("test_29.log", "test_29")168		(result, response) = invoke_function(self.m_contract_address, "bool_to_int", "", argvs = [{"type": "string","value": ""}], node_index = 0)169		logger.close(result and response["result"]["Result"] == '01')170		# logger.close(not result or response["result"]["Result"] != '01')171	172	def test_30(self):173		result = False174		logger.open("test_30.log", "test_30")175		(result, response) = invoke_function(self.m_contract_address, "bool_to_int", "", argvs = [{"type": "bytearray","value": "111122223333"}], node_index = 0)176		logger.close(result and response["result"]["Result"] == '01')177		# logger.close(not result or response["result"]["Result"] != '01')178	def test_31(self):179		result = False180		logger.open("test_31.log", "test_31")181		(result, response) = invoke_function(self.m_contract_address, "byte_to_int", "", argvs = [{"type": "bytearray","value": "1234567890abcdef"}], node_index = 0)182		logger.close(result and response["result"]["Result"] == '01')183		184	def test_32(self):185		result = False186		logger.open("test_32.log", "test_32")187		(result, response) = invoke_function(self.m_contract_address, "byte_to_int", "", argvs = [{"type": "bytearray","value": ""}], node_index = 0)188		logger.close(result and response["result"]["Result"] == '01')189		# logger.close(not result or response["result"]["Result"] != '01')190	191	def test_33(self):192		result = False193		logger.open("test_33.log", "test_33")194		(result, response) = invoke_function(self.m_contract_address, "byte_to_int", "", argvs = [{"type": "int","value": "123"}], node_index = 0)195		logger.close(result and response["result"]["Result"] == '01')196		# logger.close(not result or response["result"]["Result"] != '01')197		198	def test_34(self):199		result = False200		logger.open("test_34.log", "test_34")201		(result, response) = invoke_function(self.m_contract_address, "byte_to_int", "", argvs = [{"type": "string","value": "zxcvbnm!@"}], node_index = 0)202		logger.close(result and response["result"]["Result"] == '01')203		# logger.close(not result or response["result"]["Result"] != '01')204	def test_35(self):205		result = False206		logger.open("test_35.log", "test_35")207		(result, response) = invoke_function(self.m_contract_address, "byte_to_int", "", argvs = [{"type": "bool","value": "true"}], node_index = 0)208		logger.close(result and response["result"]["Result"] == '01')209		# logger.close(not result or response["result"]["Result"] != '01')210		211	def test_36(self):212		result = False213		logger.open("test_36.log", "test_36")214		(result, response) = invoke_function(self.m_contract_address, "byte_to_int", "", argvs = [{"type": "bool","value": "false"}], node_index = 0)215		logger.close(result and response["result"]["Result"] == '01')216		# logger.close(not result or response["result"]["Result"] != '01')217		218	def test_37(self):219		result = False220		logger.open("test_37.log", "test_37")221		(result, response) = invoke_function(self.m_contract_address, "test_37", "", argvs = [{"type": "string","value": ""}], node_index = 0)222		logger.close(result and response["result"]["Result"] == '00')223		224	def test_38(self):225		result = False226		logger.open("test_38.log", "test_38")227		(result, response) = invoke_function(self.m_contract_address, "test_38", "", argvs = [{"type": "string","value": ""}], node_index = 0)228		logger.close(result and response["result"]["Result"] == 'ff')		229	230	def test_39(self):231		result = False232		logger.open("test_39.log", "test_39")233		(result, response) = invoke_function(self.m_contract_address, "test_39", "", argvs = [{"type": "string","value": ""}], node_index = 0)234		logger.close(result and response["result"]["Result"] == 'ffff00')	235	236	def test_46(self):237		result = False238		logger.open("test_46.log", "test_46")239		(result, response) = invoke_function(self.m_contract_address, "test_46", "", argvs = [{"type": "string","value": ""}], node_index = 0)240		logger.close(result and response["result"]["Result"] == '71776572747975696f70')	241	242	def test_47(self):243		result = False244		logger.open("test_47.log", "test_47")245		(result, response) = invoke_function(self.m_contract_address, "test_47", "", argvs = [{"type": "string","value": ""}], node_index = 0)246		logger.close(result and response["result"]["Result"] == '7e214040232424255e262a28295f2b2d3d7b7d7c3a223c3e3f3b275b5d205c2c2e2f')	247	248	def test_48(self):249		result = False250		logger.open("test_48.log", "test_48")251		(result, response) = invoke_function(self.m_contract_address, "test_48", "", argvs = [{"type": "string","value": ""}], node_index = 0)252		logger.close(result and response["result"]["Result"] == '61626364313233345f292928292a28265e245e2523')	253	254	def test_60(self):255		result = False256		logger.open("test_60.log", "test_60")257		(result, response) = invoke_function(self.m_contract_address, "test_60", "", argvs = [{"type": "string","value": ""}], node_index = 0)258		logger.close(result and response["result"]["Result"] == '01')259		260	def test_62(self):261		result = False262		logger.open("test_62.log", "test_62")263		(result, response) = invoke_function(self.m_contract_address, "test_62", "", argvs = [{"type": "string","value": ""}], node_index = 0)264		logger.close(result and response["result"]["Result"] == '00')	265		266	def test_67(self):267		result = False268		logger.open("test_67.log", "test_67")269		(result, response) = invoke_function(self.m_contract_address, "test_67", "", argvs = [{"type": "string","value": ""}], node_index = 0)270		logger.close(result and response["result"]["Result"] == '1234567890abcdef')	271if __name__ == '__main__':...

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.