Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use count_open_workflow_executions method in localstack

Best Python code snippet using localstack_python

workflow_test.py

Source:workflow_test.py

...129            workflow_client,130            mocked_boto_client,131            oldest_start_date132    ):133        response = workflow_client.count_open_workflow_executions(134            oldest_start_date=oldest_start_date,135        )136        assert response.count == 123137        assert not response.truncated138        mocked_boto_client.count_open_workflow_executions.assert_called_with(139            domain=workflow_config.domain,140            startTimeFilter=dict(oldestDate=oldest_start_date),141        )142    def test_count_open_workflow_executions_with_start_time_range(143            self,144            workflow_config,145            workflow_client,146            mocked_boto_client,147            oldest_start_date,148            latest_start_date149    ):150        workflow_client.count_open_workflow_executions(151            oldest_start_date=oldest_start_date,152            latest_start_date=latest_start_date,153        )154        mocked_boto_client.count_open_workflow_executions.assert_called_with(155            domain=workflow_config.domain,156            startTimeFilter=dict(157                oldestDate=oldest_start_date,158                latestDate=latest_start_date,159            ),160        )161    def test_count_open_workflow_executions_with_oldest_start_time_and_workflow_name(162            self,163            workflow_config,164            workflow_client,165            mocked_boto_client,166            oldest_start_date,167    ):168        workflow_client.count_open_workflow_executions(169            oldest_start_date=oldest_start_date,170            workflow_name='test',171            version='test',172        )173        mocked_boto_client.count_open_workflow_executions.assert_called_with(174            domain=workflow_config.domain,175            startTimeFilter=dict(oldestDate=oldest_start_date),176            typeFilter=dict(177                name='test',178                version='test',179            )180        )181    def test_count_open_workflow_executions_with_oldest_start_time_and_tag(182            self,183            workflow_config,184            workflow_client,185            mocked_boto_client,186            oldest_start_date,187    ):188        workflow_client.count_open_workflow_executions(189            oldest_start_date=oldest_start_date,190            tag='test',191        )192        mocked_boto_client.count_open_workflow_executions.assert_called_with(193            domain=workflow_config.domain,194            startTimeFilter=dict(oldestDate=oldest_start_date),195            tagFilter=dict(tag='test'),196        )197    def test_count_open_workflow_executions_with_oldest_start_time_and_workflow_id(198            self,199            workflow_config,200            workflow_client,201            mocked_boto_client,202            oldest_start_date,203    ):204        workflow_client.count_open_workflow_executions(205            oldest_start_date=oldest_start_date,206            workflow_id='test'207        )208        mocked_boto_client.count_open_workflow_executions.assert_called_with(209            domain=workflow_config.domain,210            startTimeFilter=dict(oldestDate=oldest_start_date),211            executionFilter=dict(workflowId='test'),212        )213class TestCountClosedWorkflowExecutions:214    @pytest.fixture215    def mocked_boto_client(self):216        mocked_boto_client = mock.Mock()217        mocked_boto_client.count_closed_workflow_executions.return_value = {218            'count': 321,...

workflow.py

Source:workflow.py

...72            domain=self.workflow_client_config.domain,73            workflowId=workflow_id,74            reason=reason,75        )76    def count_open_workflow_executions(77            self,78            oldest_start_date,79            latest_start_date=None,80            workflow_name=None,81            version=None,82            tag=None,83            workflow_id=None,84    ):85        """86        Count the number of open workflows for a domain. You can pass in filtering criteria.87        This operation is eventually consistent. The results are best effort and may not exactly reflect recent updates88        and changes. Worklfows started or closed near the time when calling count_open_workflow_executions may not be reflected89        Passthrough to :meth:~SWF.Client.count_open_workflow_executions``90        executionFilter , typeFilter and tagFilter are mutually exclusive. You can specify at most one of these in a request.91        :param oldest_start_date: datetime. Specifies the oldest start date and time to return.92        :param latest_start_date: datetime. Specifies the latest start date and time to return.93        :param workflow_name: string. Required for typeFilter. Specifies the type of the workflow executions to be counted.94        :param version: string. Optional for typeFilter. Version of the workflow type.95        :param tag: string. Required for tagFilter. Specifies the tag that must be associated with the execution for it96            to meet the filter criteria.97        :param workflow_id: string. Required for executionFilter. The workflowId to pass of match the criteria of this filter.98        :return: number of open workflows within time range99        """100        workflow_filter_dict = _build_workflow_filter_dict(101            workflow_name=workflow_name,102            version=version,103            tag=tag,104            workflow_id=workflow_id,105        )106        start_time_filter_dict = _build_time_filter_dict(107            oldest_start_date=oldest_start_date,108            latest_start_date=latest_start_date,109        )110        response = self.boto_client.count_open_workflow_executions(111            domain=self.workflow_client_config.domain,112            startTimeFilter=start_time_filter_dict['startTimeFilter'],113            **workflow_filter_dict114        )115        return CountWorkflowsResult(count=response['count'], truncated=response['truncated'])116    def count_closed_workflow_executions(117            self,118            oldest_start_date=None,119            latest_start_date=None,120            oldest_close_date=None,121            latest_close_date=None,122            workflow_name=None,123            version=None,124            tag=None,...

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.