KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use successful_task method in locust

Best Python code snippet using locust

wait_and_retry_test.py

Source:wait_and_retry_test.py

1# Copyright 2019 Verily Life Sciences Inc. All Rights Reserved.2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7#     http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14"""Unit tests for the _wait_and_retry loop."""15import unittest16from dsub.commands import dsub as dsub_command17from dsub.lib import job_model18from dsub.lib import param_util19from dsub.providers import stub20import fake_time21import parameterized22PREEMPTED_TASK = {23    'job-id': 'job-1',24    'status': ('FAILURE', 123),25    'task-id': '123',26    'status-message': 'preempted',27    'error-message': 'preempted'28}29FAILED_TASK = {30    'job-id': 'job-1',31    'status': ('FAILURE', 123),32    'task-id': '123',33    'status-message': 'non-preemption failure',34    'error-message': 'non-preemption failure'35}36SUCCESSFUL_TASK = {37    'job-id': 'job-1',38    'status': ('SUCCESS', 123),39    'task-id': '123',40    'status-message': 'Success!'41}42def establish_chronology(chronology):43  dsub_command.SLEEP_FUNCTION = fake_time.FakeTime(chronology).sleep44class TestWaitAndRetry(unittest.TestCase):45  def setUp(self):46    super(TestWaitAndRetry, self).setUp()47    self.provider = stub.StubJobProvider()48  @parameterized.parameterized.expand([49      # P == R, eventual success50      (1, 1, [[PREEMPTED_TASK], [SUCCESSFUL_TASK]], []),51      # P == R, all preempted52      (1, 1, [[PREEMPTED_TASK], [PREEMPTED_TASK,53                                 PREEMPTED_TASK]], [['preempted']]),54      # P == R, non-preemption failure55      (1, 1, [[FAILED_TASK], [FAILED_TASK,56                              FAILED_TASK]], [['non-preemption failure']]),57      # R > P, eventual success58      (2, 1, [[PREEMPTED_TASK], [PREEMPTED_TASK, PREEMPTED_TASK],59              [SUCCESSFUL_TASK]], []),60      # R > P, failure61      (2, 1, [[PREEMPTED_TASK], [FAILED_TASK, FAILED_TASK],62              [FAILED_TASK, FAILED_TASK,63               FAILED_TASK]], [['non-preemption failure']]),64      # no preemptible flag, R given, success65      (1, 0, [[SUCCESSFUL_TASK]], []),66      # no preemptible flag, R given, non-preemption failure67      (1, 0, [[FAILED_TASK], [FAILED_TASK,68                              FAILED_TASK]], [['non-preemption failure']]),69      # --preemptible, R given, eventual success70      (1, True, [[PREEMPTED_TASK], [SUCCESSFUL_TASK]], [])71  ])72  def test_foo(self, retries, max_preemptible_attempts,73               list_of_list_of_operations, expected):74    def chronology(list_of_list_of_operations):75      for operations in list_of_list_of_operations:76        self.provider.set_operations(operations)77        yield 178    establish_chronology(chronology(list_of_list_of_operations))79    job_descriptor = job_model.JobDescriptor(80        job_metadata={'create-time': 123456},81        job_params=None,82        job_resources=job_model.Resources(83            logging=job_model.LoggingParam('gs://buck/logs', job_model.P_GCS),84            max_preemptible_attempts=param_util.PreemptibleParam(85                max_preemptible_attempts)),86        task_descriptors=[87            job_model.TaskDescriptor(88                task_metadata={89                    'task-id': 123,90                    'create-time': 12345691                },92                task_params=None,93                task_resources=None)94        ])95    poll_interval = 196    ret = dsub_command._wait_and_retry(97        self.provider,98        'job-1',99        poll_interval,100        retries,101        job_descriptor,102        summary=False)103    tasks = self.provider.lookup_job_tasks({'*'})104    # First, the number of tasks returned by lookup_job_tasks should be equal105    # to the total number of task attempts.106    expected_num_of_tasks = len(list_of_list_of_operations[-1])107    self.assertEqual(len(tasks), expected_num_of_tasks)108    # Second, check that the return value of _wait_and_retry is correct.109    self.assertEqual(ret, expected)110if __name__ == '__main__':...

test_cronitor.py

Source:test_cronitor.py

...15RUN_LINK = _cronitor_url('secret', 'run')16FAIL_LINK = _cronitor_url('secret', 'fail')17COMPLETE_LINK = _cronitor_url('secret', 'complete')18@cronitor('hello')19def successful_task():20    return 121@cronitor('hello')22def crashing_task():23    raise ValueError24def test_cronitor_sends_run_and_complete(notify_api, rmock):25    rmock.get(RUN_LINK, status_code=200)26    rmock.get(COMPLETE_LINK, status_code=200)27    with set_config_values(notify_api, {28        'CRONITOR_ENABLED': True,29        'CRONITOR_KEYS': {'hello': 'secret'}30    }):31        assert successful_task() == 132    assert rmock.call_count == 233    assert rmock.request_history[0].url == RUN_LINK34    assert rmock.request_history[1].url == COMPLETE_LINK35def test_cronitor_sends_run_and_fail_if_exception(notify_api, rmock):36    rmock.get(RUN_LINK, status_code=200)37    rmock.get(FAIL_LINK, status_code=200)38    with set_config_values(notify_api, {39        'CRONITOR_ENABLED': True,40        'CRONITOR_KEYS': {'hello': 'secret'}41    }):42        with pytest.raises(ValueError):43            crashing_task()44    assert rmock.call_count == 245    assert rmock.request_history[0].url == RUN_LINK46    assert rmock.request_history[1].url == FAIL_LINK47def test_cronitor_does_nothing_if_cronitor_not_enabled(notify_api, rmock):48    with set_config_values(notify_api, {49        'CRONITOR_ENABLED': False,50        'CRONITOR_KEYS': {'hello': 'secret'}51    }):52        assert successful_task() == 153    assert rmock.called is False54def test_cronitor_does_nothing_if_name_not_recognised(notify_api, rmock, mocker):55    mock_logger = mocker.patch('app.cronitor.current_app.logger')56    with set_config_values(notify_api, {57        'CRONITOR_ENABLED': True,58        'CRONITOR_KEYS': {'not-hello': 'other'}59    }):60        assert successful_task() == 161    mock_logger.error.assert_called_with(62        'Cronitor enabled but task_name hello not found in environment'63    )64    assert rmock.called is False65def test_cronitor_doesnt_crash_if_request_fails(notify_api, rmock):66    rmock.get(RUN_LINK, exc=requests.exceptions.ConnectTimeout)67    rmock.get(COMPLETE_LINK, status_code=500)68    with set_config_values(notify_api, {69        'CRONITOR_ENABLED': True,70        'CRONITOR_KEYS': {'hello': 'secret'}71    }):72        assert successful_task() == 1...

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.