KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use make_job method in autotest

Best Python code snippet using autotest_python

test_depmgr.py

Source:test_depmgr.py

...27from plainbox.impl.depmgr import DependencySolver28from plainbox.impl.testing_utils import make_job29class DependencyCycleErrorTests(TestCase):30    def setUp(self):31        self.A = make_job("A", depends="B")32        self.B = make_job("B", depends="A")33        self.exc = DependencyCycleError([self.A, self.B, self.A])34    def test_job_list(self):35        self.assertEqual(self.exc.job_list, [self.A, self.B, self.A])36    def test_affected_job(self):37        self.assertIs(self.exc.affected_job, self.A)38    def test_affecting_job(self):39        # This is the same as affected_job as this is a cycle40        self.assertIs(self.exc.affecting_job, self.A)41    def test_str(self):42        expected = "dependency cycle detected: A -> B -> A"43        observed = str(self.exc)44        self.assertEqual(expected, observed)45    def test_repr(self):46        expected = ("<DependencyCycleError job_list:["47                    "<JobDefinition id:'A' plugin:'dummy'>, "48                    "<JobDefinition id:'B' plugin:'dummy'>, "49                    "<JobDefinition id:'A' plugin:'dummy'>]>")50        observed = repr(self.exc)51        self.assertEqual(expected, observed)52class DependencyMissingErrorTests(TestCase):53    def setUp(self):54        self.A = make_job("A")55        self.exc_direct = DependencyMissingError(56            self.A, 'B', DependencyMissingError.DEP_TYPE_DIRECT)57        self.exc_resource = DependencyMissingError(58            self.A, 'B', DependencyMissingError.DEP_TYPE_RESOURCE)59    def test_job(self):60        self.assertIs(self.exc_direct.job, self.A)61        self.assertIs(self.exc_resource.job, self.A)62    def test_affected_job(self):63        self.assertIs(self.exc_direct.affected_job, self.A)64        self.assertIs(self.exc_resource.affected_job, self.A)65    def test_affecting_job(self):66        self.assertIs(self.exc_direct.affecting_job, None)67        self.assertIs(self.exc_resource.affecting_job, None)68    def test_missing_job_id(self):69        self.assertEqual(self.exc_direct.missing_job_id, 'B')70        self.assertEqual(self.exc_resource.missing_job_id, 'B')71    def test_str_direct(self):72        expected = "missing dependency: 'B' (direct)"73        observed = str(self.exc_direct)74        self.assertEqual(expected, observed)75    def test_str_resoucee(self):76        expected = "missing dependency: 'B' (resource)"77        observed = str(self.exc_resource)78        self.assertEqual(expected, observed)79    def test_repr_direct(self):80        expected = ("<DependencyMissingError "81                    "job:<JobDefinition id:'A' plugin:'dummy'> "82                    "missing_job_id:'B' "83                    "dep_type:'direct'>")84        observed = repr(self.exc_direct)85        self.assertEqual(expected, observed)86    def test_repr_resource(self):87        expected = ("<DependencyMissingError "88                    "job:<JobDefinition id:'A' plugin:'dummy'> "89                    "missing_job_id:'B' "90                    "dep_type:'resource'>")91        observed = repr(self.exc_resource)92        self.assertEqual(expected, observed)93class DependencyDuplicateErrorTests(TestCase):94    def setUp(self):95        self.A = make_job("A")96        self.another_A = make_job("A")97        self.exc = DependencyDuplicateError(self.A, self.another_A)98    def test_job(self):99        self.assertIs(self.exc.job, self.A)100    def test_duplicate_job(self):101        self.assertIs(self.exc.duplicate_job, self.another_A)102    def test_affected_job(self):103        self.assertIs(self.exc.affected_job, self.A)104    def test_affecting_job(self):105        self.assertIs(self.exc.affecting_job, self.another_A)106    def test_str(self):107        expected = "duplicate job id: 'A'"108        observed = str(self.exc)109        self.assertEqual(expected, observed)110    def test_repr(self):111        expected = ("<DependencyDuplicateError "112                    "job:<JobDefinition id:'A' plugin:'dummy'> "113                    "duplicate_job:<JobDefinition id:'A' plugin:'dummy'>>")114        observed = repr(self.exc)115        self.assertEqual(expected, observed)116class DependencySolverInternalsTests(TestCase):117    def test_get_job_map_produces_map(self):118        A = make_job('A')119        B = make_job('B')120        expected = {'A': A, 'B': B}121        observed = DependencySolver._get_job_map([A, B])122        self.assertEqual(expected, observed)123    def test_get_job_map_find_duplicates(self):124        A = make_job('A')125        another_A = make_job('A')126        with self.assertRaises(DependencyDuplicateError) as call:127            DependencySolver._get_job_map([A, another_A])128        self.assertIs(call.exception.job, A)129        self.assertIs(call.exception.duplicate_job, another_A)130class TestDependencySolver(TestCase):131    def test_empty(self):132        observed = DependencySolver.resolve_dependencies([])133        expected = []134        self.assertEqual(expected, observed)135    def test_direct_deps(self):136        # This tests the following simple job chain137        # A -> B -> C138        A = make_job(id='A', depends='B')139        B = make_job(id='B', depends='C')140        C = make_job(id='C')141        job_list = [A, B, C]142        expected = [C, B, A]143        observed = DependencySolver.resolve_dependencies(job_list)144        self.assertEqual(expected, observed)145    def test_independent_groups_deps(self):146        # This tests two independent job chains147        # A1 -> B1148        # A2 -> B2149        A1 = make_job(id='A1', depends='B1')150        B1 = make_job(id='B1',)151        A2 = make_job(id='A2', depends='B2')152        B2 = make_job(id='B2')153        job_list = [A1, B1, A2, B2]154        expected = [B1, A1, B2, A2]155        observed = DependencySolver.resolve_dependencies(job_list)156        self.assertEqual(expected, observed)157    def test_visiting_blackend_node(self):158        # This tests a visit to already visited job159        # A160        # B -> A161        # A will be visited twice162        A = make_job(id='A')163        B = make_job(id='B', depends='A')164        job_list = [A, B]165        expected = [A, B]166        observed = DependencySolver.resolve_dependencies(job_list)167        self.assertEqual(expected, observed)168    def test_resource_deps(self):169        # This tests resource deps170        # A ~> R171        A = make_job(id='A', requires='R.foo == "bar"')172        R = make_job(id='R', plugin='resource')173        job_list = [A, R]174        expected = [R, A]175        observed = DependencySolver.resolve_dependencies(job_list)176        self.assertEqual(expected, observed)177    def test_duplicate_error(self):178        A = make_job('A')179        another_A = make_job('A')180        job_list = [A, another_A]181        with self.assertRaises(DependencyDuplicateError) as call:182            DependencySolver.resolve_dependencies(job_list)183        self.assertIs(call.exception.job, A)184        self.assertIs(call.exception.duplicate_job, another_A)185    def test_missing_direct_dependency(self):186        # This tests missing dependencies187        # A -> (inexisting B)188        A = make_job(id='A', depends='B')189        job_list = [A]190        with self.assertRaises(DependencyMissingError) as call:191            DependencySolver.resolve_dependencies(job_list)192        self.assertIs(call.exception.job, A)193        self.assertEqual(call.exception.missing_job_id, 'B')194        self.assertEqual(call.exception.dep_type,195                         call.exception.DEP_TYPE_DIRECT)196    def test_missing_resource_dependency(self):197        # This tests missing resource dependencies198        # A ~> (inexisting R)199        A = make_job(id='A', requires='R.attr == "value"')200        job_list = [A]201        with self.assertRaises(DependencyMissingError) as call:202            DependencySolver.resolve_dependencies(job_list)203        self.assertIs(call.exception.job, A)204        self.assertEqual(call.exception.missing_job_id, 'R')205        self.assertEqual(call.exception.dep_type,206                         call.exception.DEP_TYPE_RESOURCE)207    def test_dependency_cycle_self(self):208        # This tests dependency loops209        # A -> A210        A = make_job(id='A', depends='A')211        job_list = [A]212        with self.assertRaises(DependencyCycleError) as call:213            DependencySolver.resolve_dependencies(job_list)214        self.assertEqual(call.exception.job_list, [A, A])215    def test_dependency_cycle_simple(self):216        # This tests dependency loops217        # A -> B -> A218        A = make_job(id='A', depends='B')219        B = make_job(id='B', depends='A')220        job_list = [A, B]221        with self.assertRaises(DependencyCycleError) as call:222            DependencySolver.resolve_dependencies(job_list)223        self.assertEqual(call.exception.job_list, [A, B, A])224    def test_dependency_cycle_longer(self):225        # This tests dependency loops226        # A -> B -> C -> D -> B227        A = make_job(id='A', depends='B')228        B = make_job(id='B', depends='C')229        C = make_job(id='C', depends='D')230        D = make_job(id='D', depends='B')231        job_list = [A, B, C, D]232        with self.assertRaises(DependencyCycleError) as call:233            DependencySolver.resolve_dependencies(job_list)234        self.assertEqual(call.exception.job_list, [B, C, D, B])235    def test_dependency_cycle_via_resource(self):236        # This tests dependency loops237        # A -> R -> A238        A = make_job(id='A', requires='R.key == "value"')239        R = make_job(id='R', depends='A', plugin="resource")240        job_list = [A, R]241        with self.assertRaises(DependencyCycleError) as call:242            DependencySolver.resolve_dependencies(job_list)...

run.py

Source:run.py

...19    from crontab import CronTab20    cron = CronTab(user=True)21    # Clear out existing jobs22    cron.remove_all(comment='DOOBSAUTO')23    def make_job(job):24        p = path.realpath(__file__)25        c = cron.new(command='{}/venv/bin/python {} {}'.format(path.split(p)[0],\26                p, job), comment='DOOBSAUTO')27        return c28    # Create the jobs29    winrate = make_job('calc_winrates')30    ts3_move_afk = make_job('ts3_move_afk')31    ts3_snapshot = make_job('ts3_snapshot')32    ts3_award_points = make_job('ts3_award_points')33    ts3_process_events = make_job('ts3_process_events')34    # Set their frequency to run35    winrate.every(1).day()36    ts3_move_afk.every(app.config['MOVE_AFK_FREQUENCY']).minute()37    ts3_snapshot.every(app.config['SNAPSHOT_FREQUENCY']).hour()38    ts3_award_points.every(app.config['AWARD_POINTS_FREQUENCY']).minute()39    ts3_process_events.every(app.config['PROCESS_EVENTS_FREQUENCY']).hour()40    try:41        assert True == winrate.is_valid()42        assert True == ts3_move_afk.is_valid()43        assert True == ts3_snapshot.is_valid()44        assert True == ts3_award_points.is_valid()45        assert True == ts3_process_events.is_valid()46    except AssertionError as e:47        print "Problem installing cronjobs: {}".format(e)...

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.