KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use test_exec method in pytest-play
Best Python code snippet using pytest-play_python
test_definite_assignment.py
Source:test_definite_assignment.py 
1from __future__ import annotations2from textwrap import dedent3from typing import final, Optional, Sequence4from _strictmodule import StrictAnalysisResult, StrictModuleLoader5from .common import StrictTestBase6@final7class DefiniteAssignmentTests(StrictTestBase):8 def analyze(9 self,10 code: str,11 mod_name: str = "mod",12 import_path: Optional[Sequence[str]] = None,13 allow_list_prefix: Optional[Sequence[str]] = None,14 stub_root: str = "",15 ) -> StrictAnalysisResult:16 code = dedent(code)17 compiler = StrictModuleLoader(18 import_path or [], stub_root, allow_list_prefix or [], [], True19 )20 module = compiler.check_source(code, f"{mod_name}.py", mod_name, [])21 return module22 def assertNoError(23 self,24 code: str,25 mod_name: str = "mod",26 import_path: Optional[Sequence[str]] = None,27 allow_list_prefix: Optional[Sequence[str]] = None,28 stub_root: str = "",29 ):30 m = self.analyze(code, mod_name, import_path, allow_list_prefix, stub_root)31 self.assertEqual(m.is_valid, True)32 self.assertEqual(m.errors, [])33 def assertError(34 self,35 code: str,36 err: str,37 mod_name: str = "mod",38 import_path: Optional[Sequence[str]] = None,39 allow_list_prefix: Optional[Sequence[str]] = None,40 stub_root: str = "",41 ):42 m = self.analyze(code, mod_name, import_path, allow_list_prefix, stub_root)43 self.assertEqual(m.is_valid, True)44 self.assertTrue(len(m.errors) > 0)45 self.assertTrue(err in m.errors[0][0])46 def test_simple_not_assigned(self) -> None:47 test_exec = """48import __strict__49abc + 150"""51 self.assertError(test_exec, "NameError")52 def test_simple_del_not_assigned(self) -> None:53 test_exec = """54import __strict__55del abc56"""57 self.assertNoError(test_exec)58 def test_simple_assign_del_ok(self) -> None:59 test_exec = """60import __strict__61abc = 162del abc63"""64 self.assertNoError(test_exec)65 def test_simple_assign_double_del(self) -> None:66 test_exec = """67import __strict__68abc = 169del abc70del abc71"""72 self.assertNoError(test_exec)73 def test_simple_if(self) -> None:74 test_exec = """75import __strict__76if False:77 abc = 178abc + 179"""80 self.assertError(test_exec, "NameError")81 def test_simple_if_del(self) -> None:82 test_exec = """83import __strict__84abc = 185if True:86 del abc87abc + 188"""89 self.assertError(test_exec, "NameError")90 def test_simple_if_else(self) -> None:91 test_exec = """92import __strict__93if str:94 foo = 195else:96 abc = 297abc + 198"""99 self.assertError(test_exec, "NameError")100 def test_simple_if_else_del(self) -> None:101 test_exec = """102import __strict__103abc = 1104if str:105 pass106else:107 del abc108abc + 1109"""110 self.assertNoError(test_exec)111 def test_simple_if_ok(self) -> None:112 test_exec = """113import __strict__114if str:115 abc = 1116else:117 abc = 2118abc119"""120 self.assertNoError(test_exec)121 def test_func_dec(self) -> None:122 test_exec = """123import __strict__124@abc125def f(x): pass126"""127 self.assertError(test_exec, "NameError")128 def test_func_self_default(self) -> None:129 test_exec = """130import __strict__131def f(x = f()): pass132"""133 self.assertError(test_exec, "NameError")134 def test_async_func_dec(self) -> None:135 test_exec = """136import __strict__137@abc138async def f(x): pass139"""140 self.assertError(test_exec, "NameError")141 def test_async_func_self_default(self) -> None:142 test_exec = """143import __strict__144async def f(x = f()): pass145"""146 self.assertError(test_exec, "NameError")147 def test_while(self) -> None:148 test_exec = """149import __strict__150while False:151 abc = 1152abc + 1153"""154 self.assertError(test_exec, "NameError")155 def test_while_else(self) -> None:156 test_exec = """157import __strict__158while False:159 abc = 1160else:161 abc = 1162abc163"""164 self.assertNoError(test_exec)165 def test_while_del(self) -> None:166 test_exec = """167import __strict__168abc = 1169while str:170 del abc171 break172abc + 1173"""174 self.assertError(test_exec, "NameError")175 def test_while_else_del(self) -> None:176 test_exec = """177import __strict__178abc = 1179while False:180 pass181else:182 del abc183x = abc + 1184"""185 self.assertError(test_exec, "NameError")186 def test_while_del_else(self) -> None:187 test_exec = """188import __strict__189abc = 1190x = 1191while x > 0:192 del abc193 x = x - 1194else:195 abc + 1196"""197 self.assertError(test_exec, "NameError")198 def test_class_defined(self) -> None:199 test_exec = """200import __strict__201class C:202 pass203C204"""205 self.assertNoError(test_exec)206 def test_class_defined_with_func(self) -> None:207 test_exec = """208import __strict__209class C:210 def __init__(self):211 pass212C213"""214 self.assertNoError(test_exec)215 def test_class_scoping(self) -> None:216 test_exec = """217import __strict__218class C:219 abc = 42220x = abc + 1221"""222 self.assertError(test_exec, "NameError")223 def test_class_uninit_global_read(self) -> None:224 test_exec = """225import __strict__226class C:227 x = abc + 1228"""229 self.assertError(test_exec, "NameError")230 def test_class_uninit_class_read(self) -> None:231 test_exec = """232import __strict__233class C:234 if str:235 abc = 42236 abc + 1237"""238 self.assertNoError(test_exec)239 def test_nested_class_uninit_read(self) -> None:240 test_exec = """241import __strict__242class C:243 abc = 42244 class D:245 x = abc + 1246"""247 self.assertError(test_exec, "NameError")248 def test_class_undef_dec(self) -> None:249 test_exec = """250import __strict__251@abc252class C:253 pass254"""255 self.assertError(test_exec, "NameError")256 def test_uninit_aug_assign(self) -> None:257 test_exec = """258import __strict__259abc += 1260"""261 self.assertError(test_exec, "NameError")262 def test_aug_assign(self) -> None:263 test_exec = """264import __strict__265abc = 0266abc += 1267 """268 self.assertNoError(test_exec)269 def test_with_no_assign(self) -> None:270 test_exec = """271import __strict__272class A:273 def __enter__(self):274 pass275 def __exit__(self, exc_tp, exc, tb):276 pass277with A():278 abc = 1279abc + 1280"""281 self.assertNoError(test_exec)282 def test_with_var(self) -> None:283 test_exec = """284import __strict__285class A:286 def __enter__(self):287 pass288 def __exit__(self, exc_tp, exc, tb):289 pass290with A() as abc:291 pass292abc293"""294 self.assertNoError(test_exec)295 def test_with_var_destructured(self) -> None:296 test_exec = """297import __strict__298class A:299 def __enter__(self):300 return 1, 3301 def __exit__(self, exc_tp, exc, tb):302 pass303with A() as (abc, foo):304 pass305abc306foo307"""308 self.assertNoError(test_exec)309 def test_import(self) -> None:310 test_exec = """311import __strict__312import abc313abc314"""315 self.assertNoError(test_exec)316 def test_import_as(self) -> None:317 test_exec = """318import __strict__319import foo as abc320abc321"""322 self.assertNoError(test_exec)323 def test_import_from(self) -> None:324 test_exec = """325import __strict__326from foo import abc327abc328"""329 self.assertNoError(test_exec)330 def test_import_from_as(self) -> None:331 test_exec = """332import __strict__333from foo import bar as abc334abc335"""336 self.assertNoError(test_exec)337 def test_del_in_finally(self) -> None:338 test_exec = """339import __strict__340try:341 abc = 1342finally:343 del abc344"""345 self.assertNoError(test_exec)346 def test_del_in_finally_2(self) -> None:347 test_exec = """348import __strict__349abc = 1350try:351 pass352finally:353 del abc354abc + 1355"""356 self.assertError(test_exec, "NameError")357 def test_finally_no_del(self) -> None:358 test_exec = """359import __strict__360try:361 abc = 1362finally:363 pass364abc365 """366 self.assertNoError(test_exec)367 def test_finally_not_defined(self) -> None:368 test_exec = """369import __strict__370try:371 abc = 1372finally:373 abc + 1374"""375 self.assertNoError(test_exec)376 def test_try_finally_deletes_apply(self) -> None:377 test_exec = """378import __strict__379abc = 1380try:381 del abc382finally:383 pass384abc + 1385"""386 self.assertError(test_exec, "NameError")387 def test_try_except_var_defined(self) -> None:388 test_exec = """389import __strict__390try:391 pass392except Exception as abc:393 abc394"""395 self.assertNoError(test_exec)396 def test_try_except_var_not_defined_after(self) -> None:397 test_exec = """398import __strict__399try:400 pass401except Exception as abc:402 pass403abc + 1404"""405 self.assertError(test_exec, "NameError")406 def test_try_except_no_try_define(self) -> None:407 test_exec = """408import __strict__409try:410 abc = 1411except Exception:412 pass413abc + 1414"""415 self.assertNoError(test_exec)416 def test_try_except_no_except_define(self) -> None:417 test_exec = """418import __strict__419try:420 pass421except Exception:422 abc = 1423abc + 1424"""425 self.assertError(test_exec, "NameError")426 def test_try_except_dels_assumed(self) -> None:427 test_exec = """428import __strict__429abc = 1430try:431 del abc432except Exception:433 pass434abc + 1435"""436 self.assertError(test_exec, "NameError")437 def test_try_except_dels_assumed_in_except(self) -> None:438 test_exec = """439import __strict__440abc = 1441try:442 del abc443except Exception:444 abc + 1445"""446 self.assertNoError(test_exec)447 def test_try_except_except_dels_assumed(self) -> None:448 test_exec = """449import __strict__450abc = 1451try:452 pass453except Exception:454 del abc455abc + 1456"""457 self.assertNoError(test_exec)458 def test_try_except_finally(self) -> None:459 test_exec = """460import __strict__461try:462 pass463except Exception:464 pass465finally:466 abc = 1467abc468"""469 self.assertNoError(test_exec)470 def test_try_except_finally_try_not_assumed(self) -> None:471 test_exec = """472import __strict__473try:474 abc = 1475except Exception:476 pass477finally:478 abc + 1479"""480 self.assertNoError(test_exec)481 def test_try_except_finally_except_not_assumed(self) -> None:482 test_exec = """483import __strict__484try:485 pass486except Exception:487 abc = 1488finally:489 abc + 1490"""491 self.assertError(test_exec, "NameError")492 def test_try_except_else_try_assumed(self) -> None:493 test_exec = """494import __strict__495try:496 abc = 1497except Exception:498 pass499else:500 abc501"""502 self.assertNoError(test_exec)503 def test_try_except_else_try_assumed_del(self) -> None:504 test_exec = """505import __strict__506try:507 abc = 1508except Exception:509 pass510else:511 del abc512"""513 self.assertNoError(test_exec)514 def test_try_except_else_except_not_assumed(self) -> None:515 test_exec = """516import __strict__517try:518 pass519except Exception:520 abc = 1521else:522 x = abc + 1523"""524 self.assertError(test_exec, "NameError")525 def test_try_except_else_except_del_not_assumed(self) -> None:526 test_exec = """527import __strict__528abc = 1529try:530 pass531except Exception:532 del abc533else:534 x = abc + 1535"""536 self.assertNoError(test_exec)537 def test_try_except_else_assign_not_assumed_for_finally(self) -> None:538 test_exec = """539import __strict__540try:541 pass542except Exception:543 pass544else:545 abc = 1546finally:547 x = abc + 1548"""549 self.assertNoError(test_exec)550 def test_try_except_finally_del_assumed(self) -> None:551 test_exec = """552import __strict__553abc = 1554try:555 pass556except Exception:557 del abc558finally:559 x = abc + 1560"""561 self.assertNoError(test_exec)562 def test_lambda_not_assigned(self) -> None:563 test_exec = """564import __strict__565x = (lambda x=abc + 1: 42)566"""567 self.assertError(test_exec, "NameError")568 def test_lambda_ok(self) -> None:569 test_exec = """570import __strict__571x = lambda x: abc572"""573 self.assertNoError(test_exec)574 def test_list_comp(self) -> None:575 test_exec = """576import __strict__577foo = [1, 2, 3]578bar = [x for x in foo]579"""580 self.assertNoError(test_exec)581 def test_list_comp_undef(self) -> None:582 test_exec = """583import __strict__584bar = [x for x in abc]585"""586 self.assertError(test_exec, "NameError")587 def test_list_comp_if(self) -> None:588 test_exec = """589import __strict__590foo = [1, 2, 3]591bar = [x for x in foo if x]592"""593 self.assertNoError(test_exec)594 def test_set_comp(self) -> None:595 test_exec = """596import __strict__597foo = [1, 2, 3]598bar = {x for x in foo}599"""600 self.assertNoError(test_exec)601 def test_set_comp_undef(self) -> None:602 test_exec = """603import __strict__604bar = {x for x in abc}605"""606 self.assertError(test_exec, "NameError")607 def test_set_comp_undef_value(self) -> None:608 test_exec = """609import __strict__610foo = [1, 2, 3]611bar = {(x, abc) for x in foo}612"""613 self.assertError(test_exec, "NameError")614 def test_set_comp_if(self) -> None:615 test_exec = """616import __strict__617foo = [1, 2, 3]618bar = {x for x in foo if x}619"""620 self.assertNoError(test_exec)621 def test_gen_comp(self) -> None:622 test_exec = """623import __strict__624foo = [1, 2, 3]625bar = (x for x in foo)626"""627 self.assertNoError(test_exec)628 def test_gen_comp_undef(self) -> None:629 test_exec = """630import __strict__631bar = (x for x in abc)632"""633 self.assertError(test_exec, "NameError")634 def test_gen_comp_undef_value(self) -> None:635 test_exec = """636import __strict__637foo = [1, 2, 3]638bar = ((x, abc) for x in foo)639"""640 self.assertError(test_exec, "NameError")641 def test_gen_comp_if(self) -> None:642 test_exec = """643import __strict__644foo = [1, 2, 3]645bar = (x for x in foo if x)646"""647 self.assertNoError(test_exec)648 def test_dict_comp(self) -> None:649 test_exec = """650import __strict__651foo = [1, 2, 3]652bar = {x:x for x in foo}653"""654 self.assertNoError(test_exec)655 def test_dict_comp_undef(self) -> None:656 test_exec = """657import __strict__658bar = {x:x for x in abc}659"""660 self.assertError(test_exec, "NameError")661 def test_dict_comp_if(self) -> None:662 test_exec = """663import __strict__664foo = [1, 2, 3]665bar = {x:x for x in foo if x}666"""667 self.assertNoError(test_exec)668 def test_self_assign(self) -> None:669 test_exec = """670import __strict__671abc = abc + 1672"""673 self.assertError(test_exec, "NameError")674 def test_ann_assign_not_defined(self) -> None:675 test_exec = """676import __strict__677abc: int678abc + 1679"""680 self.assertError(test_exec, "NameError")681 def test_expected_globals_name(self) -> None:682 test_exec = """683import __strict__684x = __name__685"""686 self.assertNoError(test_exec)687 def test_raise_unreachable(self) -> None:688 test_exec = """689import __strict__690x = 0691if x:692 raise Exception693 abc = 2694else:695 abc = 1696abc + 1697"""...
test_graph_builder.py
Source:test_graph_builder.py
...15 """16 A simple one-step pipeline 17 """18 def testpipe(x,y):19 u,v=test_exec(a=x, b=y)20 return u,v21 22 graph=build_graph(testpipe)23 self.assertTrue(isinstance(graph,Graph))24 25 # there is just a single task - hence just a single node / tick26 self.assertEqual(1, len(graph.get_all_ticks()))27 task=graph.get_task(graph.get_all_ticks()[0])28 self.assertTrue(isinstance(task,ExecTask))29 self.assertEqual('test_exec',task.command)30 self.assertEqual(set(['a','b']),set(task.inputnames))31 self.assertEqual(set(['c','d']),set(task.outputnames))32 expected = G(33 C(START_TICK, 'a', 1,'a'),34 C(START_TICK, 'b', 1,'b'),35 C(START_TICK, 'context', 1,'context'),36 T(1, task, {'name': 'test_exec', 'path':'test_exec'}),37 C(1, 'c', FINAL_TICK,'test_exec.c'),38 C(1, 'd', FINAL_TICK,'test_exec.d')39 )40 expected.set_task_property(FINAL_TICK,'aliases', 41 {'test_exec.c':'test_exec.c', 'test_exec.d':'test_exec.d'})42 43 utils.assert_graph_equal(expected, graph)44 def test_two_steps(self):45 46 def testpipe(x,y):47 u,v=test_exec(a=x, b=y)48 w,z=test_exec(name="test_exec2", a=u, b=v)49 return u,w,z50 51 graph=build_graph(testpipe)52 self.assertTrue(isinstance(graph,Graph)) 53 54 # there are two tasks55 self.assertEqual(2, len(graph.get_all_ticks()))56 ticks=sorted(graph.get_all_ticks()) 57 task1=graph.get_task(ticks[0])58 task2=graph.get_task(ticks[1])59 self.assertTrue(isinstance(task1,ExecTask))60 self.assertEqual('test_exec',task1.command)61 self.assertEqual(set(['a','b']),set(task1.inputnames))62 self.assertEqual(set(['c','d']),set(task1.outputnames))63 self.assertTrue(isinstance(task2,ExecTask))64 self.assertEqual('test_exec',task2.command)65 self.assertEqual(set(['a','b']),set(task2.inputnames))66 self.assertEqual(set(['c','d']),set(task2.outputnames))67 expected = G(68 C(START_TICK, 'a', 1,'a'),69 C(START_TICK, 'b', 1,'b'),70 C(START_TICK, 'context', 1,'context'),71 C(START_TICK, 'context', 2,'context'),72 T(1, task1, {'name': 'test_exec', 'path':'test_exec'}),73 C(1, 'c', 2, 'a'),74 C(1, 'd', 2, 'b'),75 T(2, task2, {'name': 'test_exec2', 'path':'test_exec2'}),76 C(1, 'c', FINAL_TICK,'test_exec.c'),77 C(2, 'c', FINAL_TICK,'test_exec2.c'),78 C(2, 'd', FINAL_TICK,'test_exec2.d')79 )80 expected.set_task_property(FINAL_TICK, 'aliases', 81 {'test_exec.c':'test_exec.c', 'test_exec2.c':'test_exec2.c', 'test_exec2.d':'test_exec2.d'})82 utils.assert_graph_equal(expected, graph)83 def test_inline(self):84 def testpipe(x,y):85 return test_task(x=x,y=y)86 def test_task(x,y):87 u,v=test_exec(a=x, b=y)88 w,z=test_exec(name="test_exec2", a=u, b=v)89 return u,w,z90 91 graph = build_graph(testpipe)92 # there are two tasks93 self.assertEqual(2, len(graph.get_all_ticks()))94 ticks=sorted(graph.get_all_ticks()) 95 task1=graph.get_task(ticks[0])96 task2=graph.get_task(ticks[1])97 self.assertTrue(isinstance(task1,ExecTask))98 self.assertEqual('test_exec',task1.command)99 self.assertEqual(set(['a','b']),set(task1.inputnames))100 self.assertEqual(set(['c','d']),set(task1.outputnames))101 self.assertTrue(isinstance(task2,ExecTask))102 self.assertEqual('test_exec',task2.command)103 self.assertEqual(set(['a','b']),set(task2.inputnames))104 self.assertEqual(set(['c','d']),set(task2.outputnames))105 expected = G(106 C(START_TICK, 'a', 1,'a'),107 C(START_TICK, 'b', 1,'b'),108 C(START_TICK, 'context', 1,'context'),109 C(START_TICK, 'context', 2,'context'),110 T(1, task1, {'name': 'test_exec', 'path':'test_exec'}),111 C(1, 'c', 2, 'a'),112 C(1, 'd', 2, 'b'),113 T(2, task2, {'name': 'test_exec2', 'path':'test_exec2'}),114 C(1, 'c', FINAL_TICK,'test_exec.c'),115 C(2, 'c', FINAL_TICK,'test_exec2.c'),116 C(2, 'd', FINAL_TICK,'test_exec2.d')117 )118 expected.set_task_property(FINAL_TICK,'aliases', 119 {'test_exec.c':'test_exec.c', 'test_exec2.c':'test_exec2.c', 'test_exec2.d':'test_exec2.d'})120 121 utils.assert_graph_equal(expected, graph)122 def test_nested(self):123 124 graph = build_graph(testpipe_nested)125 # there are two tasks126 self.assertEqual(1, len(graph.get_all_ticks()))127 nested=graph.get_task(graph.get_all_ticks()[0])128 self.assertTrue(isinstance(nested,NestedGraphTask))129 self.assertEqual('test_task_nested',nested.name)130 task1=ExecTask('test_exec', Package('test'), ['a','b'], ['c','d'])131 task2=ExecTask('test_exec', Package('test'), ['a','b'], ['c','d'])132 subgraph = G(133 C(START_TICK, 'a', 1,'a'),134 C(START_TICK, 'b', 1,'b'),135 C(START_TICK, 'context', 1,'context'),136 C(START_TICK, 'context', 2,'context'),137 T(1, task1, {'name': 'test_exec', 'path':'test_exec'}),138 C(1, 'c', 2, 'a'),139 C(1, 'd', 2, 'b'),140 T(2, task2, {'name': 'test_exec2', 'path':'test_exec2'}),141 C(1, 'c', FINAL_TICK,'test_exec.c'),142 C(2, 'c', FINAL_TICK,'test_exec2.c'),143 C(2, 'd', FINAL_TICK,'test_exec2.d')144 )145 subgraph.set_task_property(FINAL_TICK,'aliases', 146 {'test_exec.c':'test_exec.c', 'test_exec2.c':'test_exec2.c', 'test_exec2.d':'test_exec2.d'})147 utils.assert_graph_equal(subgraph, nested.body_graph)148 149 expected = G(150 C(START_TICK, 'x', 1,'x'),151 C(START_TICK, 'y', 1,'y'),152 C(START_TICK, 'context', 1,'context'),153 T(1, nested, {'name': 'test_task_nested', 'path':'test_task_nested'}),154 C(1, 'test_exec.c', FINAL_TICK,'test_task_nested.test_exec.c'),155 C(1, 'test_exec2.c', FINAL_TICK,'test_task_nested.test_exec2.c'),156 C(1, 'test_exec2.d', FINAL_TICK,'test_task_nested.test_exec2.d')157 )158 expected.set_task_property(FINAL_TICK,'aliases',159 {'test_task_nested.test_exec.c':'test_task_nested.test_exec.c', 160 'test_task_nested.test_exec2.c':'test_task_nested.test_exec2.c',161 'test_task_nested.test_exec2.d':'test_task_nested.test_exec2.d'})162 utils.assert_graph_equal(expected, graph)163 164 def test_parallel(self):165 166 graph = build_graph(testpipe_parallel)167 # there are two tasks168 self.assertEqual(1, len(graph.get_all_ticks()))169 parallel=graph.get_task(graph.get_all_ticks()[0])170 self.assertTrue(isinstance(parallel,ParallelSplitTask))171 self.assertEqual('test_task_parallel',parallel.name)172 task1=ExecTask('test_exec', Package('test'), ['a','b'], ['c','d'])173 task2=ExecTask('test_exec', Package('test'), ['a','b'], ['c','d'])174 subgraph = G(175 C(START_TICK, 'a', 1,'a'),176 C(START_TICK, 'b', 1,'b'),177 C(START_TICK, 'context', 1,'context'),178 C(START_TICK, 'context', 2,'context'),179 T(1, task1, {'name': 'test_exec', 'path':'test_exec'}),180 C(1, 'c', 2, 'a'),181 C(1, 'd', 2, 'b'),182 T(2, task2, {'name': 'test_exec2', 'path':'test_exec2'}),183 C(1, 'c', FINAL_TICK,'test_exec.c'),184 C(2, 'c', FINAL_TICK,'test_exec2.c'),185 C(2, 'd', FINAL_TICK,'test_exec2.d')186 )187 subgraph.set_task_property(FINAL_TICK,'aliases',188 {'test_exec.c':'test_exec.c', 'test_exec2.c':'test_exec2.c', 'test_exec2.d':'test_exec2.d'})189 utils.assert_graph_equal(subgraph, parallel.body_graph)190 191 expected = G(192 C(START_TICK, 'x', 1,'x'),193 C(START_TICK, 'y', 1,'y'),194 C(START_TICK, 'context', 1,'context'),195 T(1, parallel, {'name': 'test_task_parallel', 'path':'test_task_parallel'}),196 C(1, 'tuplelist', FINAL_TICK,'test_task_parallel.tuplelist'),197 )198 expected.set_task_property(FINAL_TICK,'aliases',{'test_task_parallel.tuplelist':'test_task_parallel.tuplelist'})199 utils.assert_graph_equal(expected, graph)200 def test_unsupported_invocation_type(self):201 def testpipe(x,y):202 u,v=test_exec(a=x, b=y)203 return u,v204 205 class FakeInvocation(Invocation):206 def __init__(self, name):207 Invocation.__init__(self, name)208 209 builder=PydronGraphBuilder(testpipe)210 builder.build()211 ticks=sorted(builder.graph.get_all_ticks())212 last_tick=ticks[-1]213 try:214 builder.add_task(last_tick, FakeInvocation('fake'))215 self.fail("Unsupported Invocations should be caught.")216 except PipelineGraphError:217 pass218testpkg=Package(pkgname='test')219def test_exec(**kwargs):220 inputnames=("a","b")221 outputnames=("c","d")222 taskprops=TaskProperties("test_exec", testpkg)223 return invoke_task(taskprops, inputnames, outputnames, **kwargs)224 225@nested()226def test_task_nested(x,y):227 u,v=test_exec(a=x, b=y)228 w,z=test_exec(name="test_exec2", a=u, b=v)229 return u,w,z230def testpipe_nested(x,y):231 return test_task_nested(x=x,y=y)232@parallel(iterable='x')233def test_task_parallel(x,y):234 u,v=test_exec(a=x, b=y)235 w,z=test_exec(name="test_exec2", a=u, b=v)236 return u,w,z237def testpipe_parallel(x,y):...
fuse_test.py
Source:fuse_test.py
...10from utils import Exp, dwarn11from config import Config12VERSION_2_9=213VERSION_3_0=314def test_exec(cmd):15 p = subprocess.Popen(cmd, shell=True)16 try:17 ret = p.wait()18 stdout, stderr = p.communicate()19 ret = p.returncode20 if (ret == 0):21 return22 else:23 msg = ""24 msg = msg + "cmd: " + cmd25 msg = msg + "\nstdout: " + str(stdout)26 msg = msg + "\nstderr: " + str(stderr)27 raise Exp(ret, msg)28 except KeyboardInterrupt as err:29 dwarn("interupted")30 p.kill()31 exit(errno.EINTR)32def yfuse_test(version, home):33 config = Config(home)34 slp = 035 cmd_line="fusermount -u /tmp/fuse/"36 try:37 test_exec(cmd_line)38 except Exception as e:39 pass40 test_exec('mkdir -p /tmp/fuse')41 test_exec('sdfs.mkdir /yfuse')42 if version == VERSION_2_9:43 cmd_line="%s --dir=/yfuse --service=1 /tmp/fuse" % (config.uss_fuse)44 elif version == VERSION_3_0:45 cmd_line="%s --dir=/yfuse --service=1 /tmp/fuse" % (config.uss_fuse3)46 else:47 print "unsupported fuse version"48 print cmd_line49 test_exec(cmd_line)50 time.sleep(slp)51 cmd_line="touch /tmp/fuse/hello"52 print cmd_line53 test_exec(cmd_line)54 time.sleep(slp)55 cmd_line="echo helloworld >> /tmp/fuse/hello"56 print cmd_line57 test_exec(cmd_line)58 time.sleep(slp)59 cmd_line="cat /tmp/fuse/hello"60 print cmd_line61 test_exec(cmd_line)62 time.sleep(slp)63 cmd_line="mkdir /tmp/fuse/testdir"64 print cmd_line65 test_exec(cmd_line)66 time.sleep(slp)67 cmd_line="cp /tmp/fuse/hello /tmp/fuse/testdir"68 print cmd_line69 test_exec(cmd_line)70 time.sleep(slp)71 cmd_line="rm -rf /tmp/fuse/testdir/hello"72 print cmd_line73 test_exec(cmd_line)74 time.sleep(slp)75 cmd_line="rmdir /tmp/fuse/testdir"76 print cmd_line77 test_exec(cmd_line)78 time.sleep(slp)79 cmd_line="mv /tmp/fuse/hello /tmp/fuse/hello_back"80 print cmd_line81 test_exec(cmd_line)82 time.sleep(slp)83 cmd_line="ln /tmp/fuse/hello_back /tmp/fuse/hard_hello"84 print cmd_line85 test_exec(cmd_line)86 time.sleep(slp)87 cmd_line="ln -s /tmp/fuse/hello_back /tmp/fuse/soft_hello"88 print cmd_line89 test_exec(cmd_line)90 time.sleep(slp)91 cmd_line="chmod +x /tmp/fuse/hello_back"92 print cmd_line93 test_exec(cmd_line)94 time.sleep(slp)95 cmd_line="rm -rf /tmp/fuse/*"96 print cmd_line97 test_exec(cmd_line)98 time.sleep(slp)99 if version == VERSION_2_9:100 cmd_line="fusermount -u /tmp/fuse/"101 elif version == VERSION_3_0:102 cmd_line="fusermount3 -u /tmp/fuse/"103 else:104 print "unsupported fuse version"105 print cmd_line106 retry = 3107 while (retry < 3):108 try:109 test_exec(cmd_line)110 break111 except Exp, e:112 #todo umount ä¼åºé113 dwarn("retry %s %s" % (cmd_line, e))114 retry = retry - 1115 time.sleep(slp)116 test_exec('rm -rf /tmp/fuse')117 test_exec('sdfs.rmdir /yfuse')118 print "!!!fuse_test success!!!"119if __name__ == "__main__":120 parser = argparse.ArgumentParser(formatter_class=RawTextHelpFormatter)121 parser.add_argument("--home", required=True, help="")122 args = parser.parse_args()123 home = args.home124 print "start testing fuse_2_9"125 yfuse_test(VERSION_2_9, home)126 print "start testing fuse_3_0"...
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 pytest-play 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?
