How to use init_test method in autotest
Best Python code snippet using autotest_python
test_spark.py
Source:test_spark.py 
...7class SparkTest(unittest.TestCase):8 neo4j_session = None9 neo4j_container = None10 spark = None11 def init_test(self, query, parameters=None):12 self.neo4_session.run("MATCH (n) DETACH DELETE n;")13 self.neo4_session.run(query, parameters)14 return self.spark.read.format("org.neo4j.spark.DataSource") \15 .option("url", self.neo4j_container.get_connection_url()) \16 .option("authentication.type", "basic") \17 .option("authentication.basic.username", "neo4j") \18 .option("authentication.basic.password", "password") \19 .option("labels", "Person") \20 .load()21 def test_string(self):22 name = "Foobar"23 df = self.init_test(24 "CREATE (p:Person {name: '" + name + "'})")25 assert name == df.select("name").collect()[0].name26 def test_int(self):27 age = 3228 df = self.init_test(29 "CREATE (p:Person {age: " + str(age) + "})")30 assert age == df.select("age").collect()[0].age31 def test_double(self):32 score = 32.333 df = self.init_test(34 "CREATE (p:Person {score: " + str(score) + "})")35 assert score == df.select("score").collect()[0].score36 def test_boolean(self):37 df = self.init_test("CREATE (p:Person {boolean: true})")38 assert True == df.select("boolean").collect()[0].boolean39 def test_time(self):40 time = datetime.time(12, 23, 0, 0, get_localzone())41 df = self.init_test(42 "CREATE (p:Person {myTime: time({hour:12, minute: 23, second: 0})})"43 )44 timeResult = df.select("myTime").collect()[0].myTime45 assert "offset-time" == timeResult.type46 # .replace used in case of UTC timezone because of https://stackoverflow.com/a/42777551/140977247 assert str(time).replace("+00:00", "Z") \48 == timeResult.value.split("+")[0]49 def test_datetime(self):50 dtString = "2015-06-24T12:50:35"51 df = self.init_test(52 "CREATE (p:Person {datetime: datetime('"+dtString+"')})")53 dt = datetime.datetime(2015, 6, 24, 12, 50, 35, 0)54 dtResult = df.select("datetime").collect()[0].datetime55 assert dt == dtResult56 def test_date(self):57 df = self.init_test(58 "CREATE (p:Person {born: date('2009-10-10')})")59 dt = datetime.date(2009, 10, 10)60 dtResult = df.select("born").collect()[0].born61 assert dt == dtResult62 def test_point(self):63 df = self.init_test(64 "CREATE (p:Person {location: point({x: 12.12, y: 13.13})})"65 )66 pointResult = df.select("location").collect()[0].location67 assert "point-2d" == pointResult[0]68 assert 7203 == pointResult[1]69 assert 12.12 == pointResult[2]70 assert 13.13 == pointResult[3]71 def test_point3d(self):72 df = self.init_test(73 "CREATE (p:Person {location: point({x: 12.12, y: 13.13, z: 1})})"74 )75 pointResult = df.select("location").collect()[0].location76 assert "point-3d" == pointResult[0]77 assert 9157 == pointResult[1]78 assert 12.12 == pointResult[2]79 assert 13.13 == pointResult[3]80 assert 1.0 == pointResult[4]81 def test_geopoint(self):82 df = self.init_test(83 "CREATE (p:Person {location: point({longitude: 12.12, latitude: 13.13})})"84 )85 pointResult = df.select("location").collect()[0].location86 assert "point-2d" == pointResult[0]87 assert 4326 == pointResult[1]88 assert 12.12 == pointResult[2]89 assert 13.13 == pointResult[3]90 def test_duration(self):91 df = self.init_test(92 "CREATE (p:Person {range: duration({days: 14, hours:16, minutes: 12})})"93 )94 durationResult = df.select("range").collect()[0].range95 assert "duration" == durationResult[0]96 assert 0 == durationResult[1]97 assert 14 == durationResult[2]98 assert 58320 == durationResult[3]99 assert 0 == durationResult[4]100 def test_string_array(self):101 df = self.init_test(102 "CREATE (p:Person {names: ['John', 'Doe']})")103 result = df.select("names").collect()[0].names104 assert "John" == result[0]105 assert "Doe" == result[1]106 def test_int_array(self):107 df = self.init_test("CREATE (p:Person {ages: [24, 56]})")108 result = df.select("ages").collect()[0].ages109 assert 24 == result[0]110 assert 56 == result[1]111 def test_double_array(self):112 df = self.init_test(113 "CREATE (p:Person {scores: [24.11, 56.11]})")114 result = df.select("scores").collect()[0].scores115 assert 24.11 == result[0]116 assert 56.11 == result[1]117 def test_boolean_array(self):118 df = self.init_test(119 "CREATE (p:Person {field: [true, false]})")120 result = df.select("field").collect()[0].field121 assert True == result[0]122 assert False == result[1]123 def test_time_array(self):124 df = self.init_test(125 "CREATE (p:Person {result: [time({hour:11, minute: 23, second: 0}), time({hour:12, minute: 23, second: 0})]})"126 )127 timeResult = df.select("result").collect()[0].result128 # .replace used in case of UTC timezone because of https://stackoverflow.com/a/42777551/1409772129 assert "offset-time" == timeResult[0].type130 assert str(datetime.time(11, 23, 0, 0, get_localzone())).replace("+00:00", "Z") \131 == timeResult[0].value.split("+")[0]132 # .replace used in case of UTC timezone because of https://stackoverflow.com/a/42777551/1409772133 assert "offset-time" == timeResult[1].type134 assert str(datetime.time(12, 23, 0, 0, get_localzone())).replace("+00:00", "Z") \135 == timeResult[1].value.split("+")[0]136 def test_datetime_array(self):137 df = self.init_test(138 "CREATE (p:Person {result: [datetime('2007-12-03T10:15:30'), datetime('2008-12-03T10:15:30')]})"139 )140 dt1 = datetime.datetime(141 2007, 12, 3, 10, 15, 30, 0)142 dt2 = datetime.datetime(143 2008, 12, 3, 10, 15, 30, 0)144 dtResult = df.select("result").collect()[0].result145 assert dt1 == dtResult[0]146 assert dt2 == dtResult[1]147 def test_date_array(self):148 df = self.init_test(149 "CREATE (p:Person {result: [date('2009-10-10'), date('2008-10-10')]})"150 )151 dt1 = datetime.date(2009, 10, 10)152 dt2 = datetime.date(2008, 10, 10)153 dtResult = df.select("result").collect()[0].result154 assert dt1 == dtResult[0]155 assert dt2 == dtResult[1]156 def test_point_array(self):157 df = self.init_test(158 "CREATE (p:Person {location: [point({x: 12.12, y: 13.13}), point({x: 13.13, y: 14.14})]})"159 )160 pointResult = df.select("location").collect()[0].location161 assert "point-2d" == pointResult[0][0]162 assert 7203 == pointResult[0][1]163 assert 12.12 == pointResult[0][2]164 assert 13.13 == pointResult[0][3]165 assert "point-2d" == pointResult[1][0]166 assert 7203 == pointResult[1][1]167 assert 13.13 == pointResult[1][2]168 assert 14.14 == pointResult[1][3]169 def test_point3d_array(self):170 df = self.init_test(171 "CREATE (p:Person {location: [point({x: 12.12, y: 13.13, z: 1}), point({x: 14.14, y: 15.15, z: 1})]})"172 )173 pointResult = df.select("location").collect()[0].location174 assert "point-3d" == pointResult[0][0]175 assert 9157 == pointResult[0][1]176 assert 12.12 == pointResult[0][2]177 assert 13.13 == pointResult[0][3]178 assert 1.0 == pointResult[0][4]179 assert "point-3d" == pointResult[1][0]180 assert 9157 == pointResult[1][1]181 assert 14.14 == pointResult[1][2]182 assert 15.15 == pointResult[1][3]183 assert 1.0 == pointResult[1][4]184 def test_geopoint_array(self):185 df = self.init_test(186 "CREATE (p:Person {location: [point({longitude: 12.12, latitude: 13.13}), point({longitude: 14.14, latitude: 15.15})]})"187 )188 pointResult = df.select("location").collect()[0].location189 assert "point-2d" == pointResult[0][0]190 assert 4326 == pointResult[0][1]191 assert 12.12 == pointResult[0][2]192 assert 13.13 == pointResult[0][3]193 assert "point-2d" == pointResult[1][0]194 assert 4326 == pointResult[1][1]195 assert 14.14 == pointResult[1][2]196 assert 15.15 == pointResult[1][3]197 def test_duration_array(self):198 df = self.init_test(199 "CREATE (p:Person {range: [duration({days: 14, hours:16, minutes: 12}), duration({days: 15, hours:16, minutes: 12})]})"200 )201 durationResult = df.select("range").collect()[0].range202 assert "duration" == durationResult[0][0]203 assert 0 == durationResult[0][1]204 assert 14 == durationResult[0][2]205 assert 58320 == durationResult[0][3]206 assert 0 == durationResult[0][4]207 assert "duration" == durationResult[1][0]208 assert 0 == durationResult[1][1]209 assert 15 == durationResult[1][2]210 assert 58320 == durationResult[1][3]211 assert 0 == durationResult[1][4]212 def test_unexisting_property(self):...
test_updater.py
Source:test_updater.py
...11 def __create_downloader_instance(self):12 downloader = combivep_updater.Downloader()13 return downloader14 def test_download(self):15 self.init_test('test_download')16 downloader = self.__create_downloader_instance()17 target_url = 'http://dbnsfp.houstonbioinformatics.org/dbNSFPzip/dbNSFP_light_v1.3.readme.txt'18 target_file = os.path.join(self.working_dir,19 os.path.basename(target_url))20 downloader.download(target_url, self.working_dir)21 self.assertGreater(os.stat(target_file).st_size,22 3397,23 msg='Download does not functional properly')24 def tearDown(self):25 self.remove_working_dir()26class TestUpdater(test_template.SafeRefDBTester):27 def __init__(self, test_name):28 test_template.SafeRefDBTester.__init__(self, test_name)29 def setUp(self):30 self.test_class = 'updater'31 def __create_updater_instance(self):32 updater = combivep_updater.Updater()33 updater.working_dir = self.working_dir34 return updater35 def test_ready1(self):36 self.init_test('test_ready1')37 updater = self.__create_updater_instance()38 self.assertEqual(updater.check_new_file('135'),39 None,40 msg='error in checking Updater ready state')41 def test_ready2(self):42 self.init_test('test_ready2')43 updater = self.__create_updater_instance()44 updater.folder_url = 'http://dummy_url'45 updater.version_pattern = 'dummy_version_pattern'46 self.assertFalse(updater.check_new_file('135'),47 msg='error in checking Updater ready state')48 def test_ready3(self):49 self.init_test('test_ready3')50 updater = self.__create_updater_instance()51 updater.files_pattern = r"""href="(?P<file_name>snp\d{3}.sql)">.*>.*(?P<date>\d{2}-[a-zA-Z]{3}-\d{4})"""52 updater.version_pattern = 'dummy_version_pattern'53 self.assertFalse(updater.check_new_file('135'),54 msg='error in checking Updater ready state')55 def tearDown(self):56 self.remove_working_dir()57#@unittest.skip("temporary disable due to high bandwidth usage")58class TestUcscUpdater(test_template.SafeRefDBTester):59 def __init__(self, test_name):60 test_template.SafeRefDBTester.__init__(self, test_name)61 def setUp(self):62 self.test_class = 'ucsc_updater'63 def __create_ucsc_updater_instance(self):64 ucsc_updater = combivep_updater.UcscUpdater()65 ucsc_updater.working_dir = self.working_dir66 ucsc_updater.files_pattern = r"""href="(?P<file_name>snp\d{3}.sql)">.*>.*(?P<date>\d{2}-[a-zA-Z]{3}-\d{4})"""67 ucsc_updater.tmp_file = cbv_const.UCSC_LIST_FILE_NAME68 ucsc_updater.local_ref_db_dir = self.working_dir69 return ucsc_updater70 @unittest.skip("temporary disable due to high bandwidth usage")71 def test_update1(self):72# self.individual_debug = True73 self.init_test('test_update1')74 ucsc_updater = self.__create_ucsc_updater_instance()75 new_file, new_version = ucsc_updater.check_new_file('135')76 self.assertTrue(new_file.endswith('.sql'),77 msg='incorrectly identify updating result')78 @unittest.skip("temporary disable due to high bandwidth usage")79 def test_update2(self):80 self.init_test('test_update2')81 ucsc_updater = self.__create_ucsc_updater_instance()82 new_file, new_version = ucsc_updater.check_new_file('136')83 self.assertTrue(new_file.endswith('.sql'),84 msg='incorrectly identify updating result')85 @unittest.skip("temporary disable due to high bandwidth usage")86 def test_not_update1(self):87 self.init_test('test_not_update1')88 ucsc_updater = self.__create_ucsc_updater_instance()89 new_file, new_version = ucsc_updater.check_new_file('137')90 self.assertEqual(new_version,91 None,92 msg='incorrectly identify updating result')93 @unittest.skip("temporary disable due to high bandwidth usage")94 def test_not_update2(self):95 self.init_test('test_not_update2')96 ucsc_updater = self.__create_ucsc_updater_instance()97 new_file, new_version = ucsc_updater.check_new_file('138')98 self.assertequal(new_version,99 none,100 msg='incorrectly identify updating result')101 @unittest.skip("temporary disable due to high bandwidth usage")102 def test_full_update1(self):103 #init104 self.init_test('test_full_update1')105 ucsc_updater = self.__create_ucsc_updater_instance()106 #run test107 new_file, new_version = ucsc_updater.check_new_file('135')108 self.asserttrue(new_file.endswith('.sql'),109 msg='UCSC updater got an invalid new file')110 self.assertEqual(new_version,111 '137',112 msg='UCSC updater got an invalid new version number')113 downloaded_file = ucsc_updater.download_new_file()114 self.assertTrue(os.path.exists(downloaded_file),115 msg="UCSC's downloaded file (%s) doesn't exist" % (downloaded_file))116 def test_parse(self):117 self.init_test('test_parse')118 ucsc_updater = self.__create_ucsc_updater_instance()119 test_file = os.path.join(self.data_dir, 'dummy_ucsc_list_file')120 out = ucsc_updater.parse(test_file)121 self.assertEqual(out['135'],122 'snp135.sql',123 msg='UCSC parser cannot find UCSC files')124 def tearDown(self):125 self.remove_working_dir()126class TestLJBUpdater(test_template.SafeRefDBTester):127 def __init__(self, test_name):128 test_template.SafeRefDBTester.__init__(self, test_name)129 def setUp(self):130 self.test_class = 'ljb_updater'131 def __create_ljb_updater_instance(self):132 ljb_updater = combivep_updater.LjbUpdater()133 ljb_updater.working_dir = self.working_dir134 ljb_updater.files_pattern = r"""href="(?P<file_name>dbNSFPv[\d.]*.readme.txt)">"""135 ljb_updater.tmp_file = cbv_const.LJB_LIST_FILE_NAME136 ljb_updater.local_ref_db_dir = self.working_dir137 return ljb_updater138 def test_update1(self):139 self.init_test('test_update1')140 ljb_updater = self.__create_ljb_updater_instance()141 new_file, new_version = ljb_updater.check_new_file('1.2')142 self.assertNotEqual(new_version,143 None,144 msg='incorrectly identify updating result')145 def test_not_update1(self):146 self.init_test('test_not_update1')147 ljb_updater = self.__create_ljb_updater_instance()148 new_file, new_version = ljb_updater.check_new_file('1.3')149 self.assertEqual(new_version,150 None,151 msg='incorrectly identify updating result')152 def test_not_update2(self):153 self.init_test('test_not_update2')154 ljb_updater = self.__create_ljb_updater_instance()155 new_file, new_version = ljb_updater.check_new_file('1.4')156 self.assertEqual(new_version,157 None,158 msg='incorrectly identify updating result')159 def test_full_update1(self):160 #init161 self.init_test('test_full_update1')162 ljb_updater = self.__create_ljb_updater_instance()163 #run test164 new_file, new_version = ljb_updater.check_new_file('1.2')165 self.assertTrue(new_file.endswith('.readme.txt'),166 msg='LJB updater got an invalid new file')167 self.assertEqual(new_version,168 '1.3',169 msg='LJB updater got an invalid new version number')170 downloaded_file = ljb_updater.download_new_file()171 self.assertTrue(os.path.exists(downloaded_file),172 msg="LJB's downloaded file (%s) doesn't exist" % (downloaded_file))173 def test_parse(self):174 self.init_test('test_parse')175 ljb_updater = self.__create_ljb_updater_instance()176 test_file = os.path.join(self.data_dir, 'dummy_ljb_list_file')177 out = ljb_updater.parse(test_file)178 self.assertEqual(out['1.3'],179 'dbNSFPv1.3.readme.txt',180 msg='LJB parser cannot find LJB files')181 def tearDown(self):182 self.remove_working_dir()183class TestMisc(test_template.SafeRefDBTester):184 """ test (a few) miscellaneous function(s) """185 def __init__(self, test_name):186 test_template.SafeRefDBTester.__init__(self, test_name)187 def setUp(self):188 self.test_class = 'misc'189 def test_unzip(self):190 self.init_test('test_unzip')191 test_file = os.path.join(self.data_dir, 'dummy_ljb.zip')192 out = combivep_updater.unzip(test_file, self.working_dir)193 self.assertEqual(out[0],194 os.path.join(self.working_dir,195 'try.in'),196 msg='some files are missing')197 self.assertEqual(out[1],198 os.path.join(self.working_dir,199 'search_dbNSFP_light_v1.3.readme.pdf'),200 msg='some files are missing')201 self.assertEqual(out[2],202 os.path.join(self.working_dir,203 'search_dbNSFP_light_v1.3.readme.doc'),204 msg='some files are missing')205 self.assertEqual(out[3],206 os.path.join(self.working_dir,207 'search_dbNSFP_light13.class'),208 msg='some files are missing')209 self.assertEqual(out[4],210 os.path.join(self.working_dir,211 'dbNSFP_light_v1.3.readme.txt'),212 msg='some files are missing')213 self.assertEqual(out[5],214 os.path.join(self.working_dir,215 'abc/tryhg19.in'),216 msg='some files are missing')217 for out_file in out:218 self.assertTrue(os.path.exists(out_file),219 msg='"%s" file is missing' % (out_file))220 def test_gz(self):221 self.init_test('test_gz')222 test_file = os.path.join(self.data_dir, 'test_gz.txt.gz')223 working_file = os.path.join(self.working_dir, 'test_gz.txt.gz')224 self.copy_file(test_file, working_file)225 out = combivep_updater.ungz(working_file)226 self.assertTrue(out,227 'no output file from "ungz" function')228 self.assertTrue(os.path.exists(out),229 'ungz function does not work properly')230 def tearDown(self):...
test_Microprocess.py
Source:test_Microprocess.py
...32 """A full set of tests for the Microprocess class."""33 classtotest = microprocess34 def test_SmokeTest_NoArguments(self):35 "__init__ - Called with no arguments. Creates multiple microprocess objects with no arguments and checks they do not have matching ids. "36 self.init_test(microprocess)37 class DummySched:38 run = None39 def __init__(self):40 self.threads = {}41 if (self.__class__.run is None):42 self.__class__.run = self43 def _addThread(self, mprocess):44 self.threads[mprocess] = 145 def wakeThread(self, mprocess, canActivate=False):46 if mprocess in self.threads:47 self.threads[mprocess] = 148 def pauseThread(self, mprocess):49 if mprocess in self.threads:50 self.threads[mprocess] = 051 def isThreadPaused(self, mprocess):52 return 0==self.threads[mprocess]53##name54# def test_InitWithArgs(self,name = "fred"):55# "__init__ - With test argument to check that the name gets set"56# m = self.init_test(microprocess, (name,))57# self.failIf(name != m.name, "Name not set at __init__")58#init flag59#id60 def init_test(self,mpsubclass=microprocess, args=()):61 "This is an internal method that can be used to create a microprocess object with the arguments that you want. It also runs the duplicate ID check by creating a couple of objects. Although there is only one optional argument to init a tuple is used here in case of future extension."62# m=apply(mpsubclass,args)63 m = mpsubclass(*args)64 self.failUnless(m.init," Microprocess initialization failed with no arguments")65 # This is weak test for duplicate IDs but might catch some silly errors66# n=apply(mpsubclass,(args))67 n=mpsubclass(*args)68# o=apply(mpsubclass,(args))69 o=mpsubclass(*args)70 self.failUnless(m.id and n.id and o.id,"All microprocesses should have id values.")71 self.failIf(m.id == n.id or n.id == o.id , "Non-unique IDs")72 return m73#activate test74 def test_activate_basicActivatation(self):75 "Tests the activation method operates as expected with a chosen scheduler"76 testscheduler = self.DummySched()77 mp = self.init_test()78 temp = len(testscheduler.threads)79 mp.activate(testscheduler)80 self.failIf(len(testscheduler.threads) != temp+1, "activate doesn't call _addthread on the requested scheduler")81 self.failIf(mp not in testscheduler.threads, "Microprocess not added to scheduler properly.")82#setSchedulerClass Test83 def test_activate_afterCallingSetSchedulerClass(self):84 "Tests the setting of the scheduler class to be used for a subclass of microprocess is actually reflected at activation"85 # This creates a DummySched.run instance.86 testscheduler = self.DummySched()87 temp = len(testscheduler.threads)88 basicmp = self.init_test()89 class SpecialSchedulerMicroprocess(microprocess):90 pass91 SpecialSchedulerMicroprocess.setSchedulerClass(self.DummySched)92 specialmp = self.init_test(SpecialSchedulerMicroprocess)93 basicmp.activate()94 specialmp.activate()95 self.failIf(len(testscheduler.threads) == temp+1, "activate doesn't call _addthread on the requested scheduler")96 self.failIf(specialmp in testscheduler.threads, "Microprocess not added to scheduler properly.")97 def test_setSchedulerClass(self):98 "Tests setting scheduler class and that the default scheduler is Scheduler.scheduler"99 class DummySched:100 pass101 premp = self.init_test()102 before = premp.__class__.schedulerClass103 microprocess.setSchedulerClass(DummySched)104 postmp = self.init_test()105 after = postmp.__class__.schedulerClass106 self.failUnless(before == Scheduler.scheduler, "Default scheduler not as expected!")107 self.failUnless(after == DummySched, "Set scheduler failed to set the scheduler properly")108 self.failUnless(before != after, "Setting scheduler did not change class!")109 microprocess.setSchedulerClass(before)110 resetmp=self.init_test()111 self.failUnless(resetmp.__class__.schedulerClass == Scheduler.scheduler)112 def test_overriddenMainWorksWithNext(self):113 "Tests that an overridden main is run correctly by repeatedly calling next() and that termination occurs at the proper time with the proper StopIteration exception."114 class testthread(microprocess):115 def __init__(self):116 self.i = 0117 self.__super.__init__()118 def main(self):119 while self.i < 100:120 self.i = self.i + 1121 yield 1122 123 thr = self.init_test(testthread)124 thr.activate(self.DummySched())125 for i in vrange(1,101):126 self.failUnless(thr.next(), "Unexpected false return value")127 self.failUnless(thr.i == i, "Iteration of main not performed!")128 #self.failIf(thr.next(), "Should return false as has returned at this point.")129 self.failUnlessRaises(StopIteration, thr.next)130 self.failUnless(thr.i == 100, "Unexpected final result of iteration")131 def test_Stop(self):132 "After being stopped a microprocess returns true to _isStopped and false to _isRunnable."133 thr = self.init_test()134 testsched = self.DummySched()135 thr.activate(testsched)136 self.failIf(thr._isStopped(), "Thread reports itself stopped before stop is called!")137 self.failUnless(thr._isRunnable(), "Thread doesn't report itself runnable before stop is called.")138 thr.stop()139 self.failUnless(thr._isStopped, "Thread doesn't report itself stopped after stop is called.")140 self.assert_(thr._isRunnable(),"Scheduler reports thread reports as runnable, but actually thread is stopped at this point.")141 thr.unpause()142 self.failUnless(thr._isStopped, "Thread doesn't report itself stopped after unpause attempt.")143 self.assert_(thr._isRunnable(),"Scheduler reports thread reports as runnable, but actually thread is stopped at this point")144 def test_pause(self):145 "After being paused a microprocess returns false to _isRunnable. Also tests _isRunnable and _unpause."146 thr = self.init_test()147 thr.activate(self.DummySched())148 self.failUnless(thr._isRunnable(), "Thread doesn't report itself runnable before pause is called.")149 thr.pause()150 self.failIf(thr._isRunnable(),"Thread reports as runnable. Should be paused at this point.")151 thr.unpause()152 self.failUnless(thr._isRunnable(), "Thread doesn't report itself runnable after _unpause is called.")153 def test_Next(self):154 "Additional checks over the main overridden main checks to test pausing and stopping behaviour."155 class testthread(microprocess):156 def __init__(self):157 self.i = 0158 self.__super.__init__()159 def main(self):160 while 1:161 self.i = self.i + 1162 yield 1163 thr = self.init_test(testthread)164 sched = self.DummySched()165 thr.activate(sched)166 for x in vrange(1,5):167 self.failUnless(thr.next())168 self.failUnless(thr.i == x)169 thr.pause()170 for x in vrange(5,10):171 self.failUnless(thr.next())172 self.assert_(thr.i == x, "Thread does not pause itself, that is down to the scheduler")173 thr.unpause()174 for x in vrange(10,15):175 self.failUnless(thr.next())176 self.failUnless(thr.i == x)177 thr.stop()178 self.failIf(thr.next())179 self.failUnlessRaises(StopIteration, thr.next)180 def test__closeDownMicroprocess(self):181 "Stub _closeDownMicroprocess should always return 0"182 mp = self.init_test()183 self.failUnless(0 == mp._closeDownMicroprocess())184if __name__=='__main__':...
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 autotest 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?
