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)...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!