Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

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.