Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
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.
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?
