How to use set_running method in autotest

Best Python code snippet using autotest_python

test_task_planner.py

Source:test_task_planner.py

...38def test_task_finish_with_ephemerals():39  pure_ephemeral = empty_task(processes=[p1(ephemeral=True)])40  p = TaskPlanner(pure_ephemeral)41  assert p.is_complete()42  p.set_running('p1')43  assert p.is_complete()44  p.add_failure('p1')45  assert p.is_complete()46  assert not p.failed47  assert p.finished == _('p1')48  with_ephemeral = empty_task(processes=[p1, p2(ephemeral=True)])49  p = TaskPlanner(with_ephemeral)50  assert not p.is_complete()51  assert p.runnable == _('p1', 'p2')52  p.set_running('p1')53  assert not p.is_complete()54  p.add_failure('p1')55  assert p.is_complete()56  p.set_running('p2')57  assert p.is_complete()58  p.add_failure('p2')59  assert p.is_complete()60  assert p.failed == _('p1')61def test_task_finish_with_daemons():62  # Daemon is still restricted to the failure limit63  p = TaskPlanner(empty_task(processes=[p1(daemon=True)]))64  assert not p.is_complete()65  p.set_running('p1')66  assert not p.is_complete()67  p.add_failure('p1')68  assert p.is_complete()69  # Resilient to two failures70  p = TaskPlanner(empty_task(processes=[p1(daemon=True, max_failures=2)]))71  assert not p.is_complete()72  p.set_running('p1')73  assert not p.is_complete()74  p.add_failure('p1')75  assert not p.is_complete()76  p.set_running('p1')77  assert not p.is_complete()78  p.add_failure('p1')79  assert p.is_complete()80  # Can swallow successes81  p = TaskPlanner(empty_task(processes=[p1(daemon=True, max_failures=2)]))82  assert not p.is_complete()83  p.set_running('p1')84  assert not p.is_complete()85  p.add_failure('p1')86  assert not p.is_complete()87  p.set_running('p1')88  assert not p.is_complete()89  p.add_success('p1')90  assert not p.is_complete()91  p.set_running('p1')92  assert not p.is_complete()93  p.add_failure('p1')94  assert p.is_complete()95def test_task_finish_with_daemon_ephemerals():96  p = TaskPlanner(empty_task(processes=[p1, p2(daemon=True, ephemeral=True, max_failures=2)]))97  assert not p.is_complete()98  p.set_running('p1')99  p.set_running('p2')100  assert not p.is_complete()101  p.add_success('p1')102  assert p.is_complete()103def test_task_process_cannot_depend_upon_daemon():104  with pytest.raises(TaskPlanner.InvalidSchedule):105    TaskPlanner(empty_task(processes=[p1(daemon=True), p2], constraints=[{'order': ['p1', 'p2']}]))106def test_task_non_ephemeral_process_cannot_depend_on_ephemeral_process():107  with pytest.raises(TaskPlanner.InvalidSchedule):108    TaskPlanner(empty_task(processes=[p1(ephemeral=True), p2],109                           constraints=[{'order': ['p1', 'p2']}]))110def test_task_failed_predecessor_does_not_make_process_runnable():111  p = TaskPlanner(empty_task(processes=[p1, p2], constraints=[{'order': ['p1', 'p2']}]))112  p.set_running('p1')113  p.add_success('p1')114  assert 'p2' in p.runnable115  assert not p.is_complete()116  p = TaskPlanner(empty_task(processes=[p1, p2], constraints=[{'order': ['p1', 'p2']}]))117  p.set_running('p1')118  p.add_failure('p1')119  assert 'p2' not in p.runnable120  assert not p.is_complete()121def test_task_daemon_duration():122  p = TaskPlanner(empty_task(processes=[p1(daemon=True, max_failures=2, min_duration=10)]))123  assert 'p1' in p.runnable124  p.set_running('p1')125  p.add_success('p1', timestamp=5)126  assert 'p1' not in p.runnable_at(timestamp=5)127  assert 'p1' not in p.runnable_at(timestamp=10)128  assert 'p1' in p.runnable_at(timestamp=15)129  assert 'p1' in p.runnable_at(timestamp=20)130  p.set_running('p1')131  p.add_failure('p1', timestamp=10)132  assert 'p1' not in p.runnable_at(timestamp=10)133  assert 'p1' not in p.runnable_at(timestamp=15)134  assert 'p1' in p.runnable_at(timestamp=20)135  assert 'p1' in p.runnable_at(timestamp=25)136  p.set_running('p1')137  p.add_failure('p1', timestamp=15)138  assert 'p1' not in p.runnable_at(timestamp=15)139  assert 'p1' not in p.runnable_at(timestamp=20)140  assert 'p1' not in p.runnable_at(timestamp=25)  # task past maximum failure limit141  assert 'p1' not in p.runnable_at(timestamp=30)142def test_task_waits():143  dt = p1(daemon=True, max_failures=0)144  p = TaskPlanner(empty_task(processes=[dt(name='d3', min_duration=3),145                                        dt(name='d5', min_duration=5),146                                        dt(name='d7', min_duration=7)]))147  assert p.runnable_at(timestamp=0) == _('d3', 'd5', 'd7')148  assert p.min_wait(timestamp=0) == 0149  p.set_running('d3')150  p.add_success('d3', timestamp=0)151  assert p.runnable_at(timestamp=0) == _('d5', 'd7')152  assert p.waiting_at(timestamp=0) == _('d3')153  assert approx_equal(p.get_wait('d3', timestamp=0), 3)154  assert approx_equal(p.min_wait(timestamp=0), 0)155  assert approx_equal(p.min_wait(timestamp=1), 0)156  assert p.waiting_at(timestamp=3) == empty157  assert p.runnable_at(timestamp=3) == _('d3', 'd5', 'd7')158  p.set_running('d3')159  p.set_running('d7')160  p.add_success('d7', timestamp=1)161  assert approx_equal(p.min_wait(timestamp=2), 0)162  p.set_running('d5')163  assert approx_equal(p.min_wait(timestamp=2), 6)164  p.add_success('d5', timestamp=2)165  p.add_success('d3', timestamp=2)166  assert approx_equal(p.min_wait(timestamp=3), 2)167  assert p.runnable_at(timestamp=2) == empty168  assert p.runnable_at(timestamp=5) == _('d3')169  assert p.runnable_at(timestamp=7) == _('d3', 'd5')170  assert p.runnable_at(timestamp=8) == _('d3', 'd5', 'd7')171def test_task_fails():172  dt = p1(max_failures=1, min_duration=1)173  p = TaskPlanner(empty_task(processes=[dt(name='d1'), dt(name='d2')]))174  assert p.runnable_at(timestamp=0) == _('d1', 'd2')175  p.set_running('d1')176  p.set_running('d2')177  assert p.runnable_at(timestamp=0) == empty178  p.add_failure('d1', timestamp=0)179  p.add_failure('d2', timestamp=0)180  assert p.runnable_at(timestamp=0) == empty181  assert p.min_wait(timestamp=0) == TaskPlanner.INFINITY182  p = TaskPlanner(empty_task(processes=[dt(name='d1'), dt(name='d2')]))183  assert p.runnable_at(timestamp=0) == _('d1', 'd2')184  p.set_running('d1')185  p.set_failed('d1')186  assert p.runnable_at(timestamp=0) == _('d2')187  p.set_running('d2')188  assert p.runnable_at(timestamp=0) == empty189  p.add_failure('d2', timestamp=0)190  assert p.runnable_at(timestamp=0) == empty191  assert p.min_wait(timestamp=0) == TaskPlanner.INFINITY192  # test max_failures=0 && daemon==True ==> retries forever193  p = TaskPlanner(empty_task(processes=[dt(name='d1', max_failures=0, daemon=True)]))194  for k in range(10):195    p.set_running('d1')196    assert 'd1' in p.running197    assert 'd1' not in p.failed198    p.add_failure('d1')199    assert 'd1' not in p.running200    assert 'd1' not in p.failed201    p.set_running('d1')202    assert 'd1' in p.running203    assert 'd1' not in p.failed204    p.add_success('d1')205    assert 'd1' not in p.running206    assert 'd1' not in p.failed207    assert 'd1' not in p.finished208  p = TaskPlanner(empty_task(processes=[dt(name='d1', max_failures=0)]))209  p.set_running('d1')210  assert 'd1' in p.running211  assert 'd1' not in p.failed212  p.add_failure('d1')213  assert 'd1' not in p.running214  assert 'd1' not in p.failed215  p.set_running('d1')216  assert 'd1' in p.running217  assert 'd1' not in p.failed218  p.add_success('d1')219  assert 'd1' not in p.running220  assert 'd1' not in p.failed221  assert 'd1' in p.finished222def test_task_lost():223  dt = p1(max_failures=2, min_duration=1)224  # regular success behavior225  p = TaskPlanner(empty_task(processes=[dt(name='d1')]))226  assert p.runnable_at(timestamp=0) == _('d1')227  p.set_running('d1')228  p.add_success('d1', timestamp=0)229  assert p.min_wait(timestamp=0) == TaskPlanner.INFINITY230  # regular failure behavior231  p = TaskPlanner(empty_task(processes=[dt(name='d1')]))232  assert p.runnable_at(timestamp=0) == _('d1')233  p.set_running('d1')234  p.add_failure('d1', timestamp=1)235  assert approx_equal(p.min_wait(timestamp=1), 1)236  p.set_running('d1')237  p.add_failure('d1', timestamp=3)238  assert p.min_wait(timestamp=3) == TaskPlanner.INFINITY239  # lost behavior240  p = TaskPlanner(empty_task(processes=[dt(name='d1')]))241  assert p.runnable_at(timestamp=0) == _('d1')242  p.set_running('d1')243  p.add_failure('d1', timestamp=1)244  assert approx_equal(p.min_wait(timestamp=1), 1)245  p.set_running('d1')246  p.lost('d1')247  assert approx_equal(p.min_wait(timestamp=1), 1)248  p.set_running('d1')249  p.add_failure('d1', timestamp=3)250  assert p.min_wait(timestamp=3) == TaskPlanner.INFINITY251def test_task_filters():252  t = p1253  task = empty_task(254      processes=[255          t(name='p1'), t(name='p2'), t(name='p3'),256          t(name='f1'), t(name='f2'), t(name='f3')],257      constraints=[258          Constraint(order=['p1', 'p2', 'p3']),259          Constraint(order=['f1', 'f2', 'f3'])])260  assert TaskPlanner(task, process_filter=lambda proc: proc.name().get().startswith('p'))261  assert TaskPlanner(task, process_filter=lambda proc: proc.name().get().startswith('f'))262  with pytest.raises(TaskPlanner.InvalidSchedule):263    TaskPlanner(task(constraints=[Constraint(order=['p1', 'f1'])]),264                process_filter=lambda proc: proc.name().get().startswith('p'))265  with pytest.raises(TaskPlanner.InvalidSchedule):266    TaskPlanner(task(constraints=[Constraint(order=['p1', 'f1'])]),267                process_filter=lambda proc: proc.name().get().startswith('f'))268def test_task_max_runs():269  class CappedTaskPlanner(TaskPlanner):270    TOTAL_RUN_LIMIT = 2271  dt = p1(daemon=True, max_failures=0)272  p = CappedTaskPlanner(empty_task(processes=[dt(name='d1', max_failures=100, daemon=False)]))273  p.set_running('d1')274  p.add_failure('d1', timestamp=1)275  assert 'd1' in p.runnable276  p.set_running('d1')277  p.add_failure('d1', timestamp=2)278  assert 'd1' not in p.runnable279  p = CappedTaskPlanner(empty_task(processes=[dt(name='d1', max_failures=100)]))280  p.set_running('d1')281  p.add_failure('d1', timestamp=1)282  assert 'd1' in p.runnable283  p.set_running('d1')284  p.add_success('d1', timestamp=2)...

ut_runable.py

Source:ut_runable.py

...4        self.paint_calls = 05        Rectangle.__init__(self, parent, color=(255, 0, 0), **kwargs)6        Runable.__init__(self, parent)7        self.console = None8    def set_running(self, val):9        if val == self.is_running:10            return11        super().set_running(val)12        if self.is_running:13            self.console.set_text(self.console.text + "\nMethod called : set_running(True)")14        else:15            self.console.set_text(self.console.text + "\nMethod called : set_running(False)")16        self.console.parent.pack(adapt=True)17    def paint(self):18        self.color.set_hue(self.paint_calls)19        super().paint()20    def run(self):21        self.paint_calls = (self.paint_calls + .003) % 36022        self.send_paint_request()23class UT_Runable_Zone(Zone):24    def __init__(self, *args, **kwargs):25        Zone.__init__(self, *args, **kwargs)26        Layer(self, name="zones_layer", spacing=10)27        z1 = Zone(self, background_color=(150, 150, 150), padding=10, spacing=10)28        self.default_layer.pack()29        rainbow = RainbowRect(z1)30        buttons_zone = Zone(z1, spacing=10)31        buttons_zone.set_style_for(Button, width=200)32        Button(buttons_zone, text="start running", command=PrefilledFunction(rainbow.set_running, True))33        Button(buttons_zone, text="stop running", command=PrefilledFunction(rainbow.set_running, False))34        buttons_zone.pack(axis="horizontal", adapt=True)35        rainbow.resize(*buttons_zone.rect.size)36        rainbow.console = Text(z1, text="Console:", max_width=rainbow.rect.w)37        z1.pack(adapt=True)38    def load_sections(self):39        self.parent.add_section(40            title="Paintable.set_running()",41            tests=[42                "After set_running(True), the Runable is running",43                "After set_running(False), the Runable is stopped",44            ]45        )46# For the PresentationScene import47ut_zone_class = UT_Runable_Zone48if __name__ == "__main__":49    from baopig.prefabs.testerscene import TesterScene50    app = Application()51    TesterScene(app, ut_zone_class)...

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.