How to use test_job method in autotest

Best Python code snippet using autotest_python

model_jobs.py

Source:model_jobs.py Github

copy

Full Screen

1# Copyright 2014 Google Inc. All Rights Reserved.2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7# http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS-IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14"""Functional tests for models/jobs.py."""15__author__ = [16 'mgainer@google.com (Mike Gainer)',17]18from common.utils import Namespace19from models import jobs20from models import transforms21from tests.functional import actions22from google.appengine.ext import db23TEST_NAMESPACE = 'test'24TEST_DATA = {'bunny_names': ['flopsy', 'mopsy', 'cottontail']}25TEST_DURATION = 12326TEST_BAD_DATA = {'wolf_actions': ['huff', 'puff', 'blow your house down']}27TEST_BAD_DURATION = 428class MockAppContext(object):29 def __init__(self, namespace):30 self._namespace = namespace31 def get_namespace_name(self):32 return self._namespace33class TestJob(jobs.DurableJobBase):34 def __init__(self, namespace):35 super(TestJob, self).__init__(MockAppContext(namespace))36 def force_start_job(self, sequence_num):37 with Namespace(self._namespace):38 db.run_in_transaction(39 jobs.DurableJobEntity._start_job, self._job_name,40 sequence_num)41 def force_complete_job(self, sequence_num, data, duration):42 data = transforms.dumps(data)43 with Namespace(self._namespace):44 db.run_in_transaction(45 jobs.DurableJobEntity._complete_job, self._job_name,46 sequence_num, data, duration)47 def force_fail_job(self, sequence_num, data, duration):48 data = transforms.dumps(data)49 with Namespace(self._namespace):50 db.run_in_transaction(51 jobs.DurableJobEntity._fail_job, self._job_name,52 sequence_num, data, duration)53 def get_output(self):54 return transforms.loads(self.load().output)55class JobOperationsTest(actions.TestBase):56 """Validate operation of job behaviors."""57 def setUp(self):58 super(JobOperationsTest, self).setUp()59 self.test_job = TestJob(TEST_NAMESPACE)60 # ---------------------------------------------------------------------61 # Tests with no item in datastore62 def test_load_finds_none(self):63 self.assertIsNone(self.test_job.load())64 def test_cancel_finds_none(self):65 self.assertIsNone(self.test_job.cancel())66 def test_not_active(self):67 self.assertFalse(self.test_job.is_active())68 # ---------------------------------------------------------------------69 # Normal operation w/ no admin intervention70 def test_submit_enqueues_job(self):71 self.assertFalse(self.test_job.is_active())72 self.test_job.submit()73 self.assertTrue(self.test_job.is_active())74 self.assertEquals(jobs.STATUS_CODE_QUEUED,75 self.test_job.load().status_code)76 def test_start_starts_job(self):77 self.assertFalse(self.test_job.is_active())78 sequence_num = self.test_job.submit()79 self.test_job.force_start_job(sequence_num)80 self.assertTrue(self.test_job.is_active())81 self.assertEquals(jobs.STATUS_CODE_STARTED,82 self.test_job.load().status_code)83 def test_complete_job_saves_result(self):84 self._test_saves_result(self.test_job.force_complete_job,85 jobs.STATUS_CODE_COMPLETED)86 def test_fail_job_saves_result(self):87 self._test_saves_result(self.test_job.force_fail_job,88 jobs.STATUS_CODE_FAILED)89 def _test_saves_result(self, func, expected_status):90 self.assertFalse(self.test_job.is_active())91 sequence_num = self.test_job.submit()92 self.test_job.force_start_job(sequence_num)93 self.assertIsNone(self.test_job.load().output)94 func(sequence_num, TEST_DATA, TEST_DURATION)95 self.assertFalse(self.test_job.is_active())96 self.assertEquals(expected_status, self.test_job.load().status_code)97 self.assertEquals(TEST_DATA, self.test_job.get_output())98 self.assertEquals(TEST_DURATION,99 self.test_job.load().execution_time_sec)100 def test_submit_does_not_restart_running_job(self):101 sequence_num = self.test_job.submit()102 self.test_job.force_start_job(sequence_num)103 next_seq = self.test_job.submit()104 self.assertEquals(next_seq, -1)105 # Check status is still STARTED, not QUEUED, which it would be106 # if we'd started the job anew.107 self.assertEquals(jobs.STATUS_CODE_STARTED,108 self.test_job.load().status_code)109 # --------------------------------------------------------------------110 # Cancelling jobs111 def test_cancel_kills_queued_job(self):112 self.assertFalse(self.test_job.is_active())113 self.test_job.submit()114 self.test_job.cancel()115 self.assertFalse(self.test_job.is_active())116 self.assertIn('Canceled by default', self.test_job.load().output)117 self.assertEquals(jobs.STATUS_CODE_FAILED,118 self.test_job.load().status_code)119 def test_cancel_kills_started_job(self):120 self.assertFalse(self.test_job.is_active())121 sequence_num = self.test_job.submit()122 self.test_job.force_start_job(sequence_num)123 self.test_job.cancel()124 self.assertFalse(self.test_job.is_active())125 self.assertEquals(jobs.STATUS_CODE_FAILED,126 self.test_job.load().status_code)127 def test_cancel_does_not_kill_completed_job(self):128 sequence_num = self.test_job.submit()129 self.test_job.force_start_job(sequence_num)130 self.test_job.force_complete_job(sequence_num, TEST_DATA,131 TEST_DURATION)132 self.assertFalse(self.test_job.is_active())133 self.test_job.cancel()134 self.assertEquals(jobs.STATUS_CODE_COMPLETED,135 self.test_job.load().status_code)136 def test_killed_job_can_still_complete(self):137 self._killed_job_can_still_record_results(138 self.test_job.force_complete_job, jobs.STATUS_CODE_COMPLETED)139 def test_killed_job_can_still_fail(self):140 self._killed_job_can_still_record_results(141 self.test_job.force_fail_job, jobs.STATUS_CODE_FAILED)142 def _killed_job_can_still_record_results(self, func, expected_status):143 sequence_num = self.test_job.submit()144 self.test_job.force_start_job(sequence_num)145 self.test_job.cancel()146 self.assertIn('Canceled by default', self.test_job.load().output)147 self.assertEquals(jobs.STATUS_CODE_FAILED,148 self.test_job.load().status_code)149 func(sequence_num, TEST_DATA, TEST_DURATION)150 self.assertEquals(expected_status, self.test_job.load().status_code)151 self.assertEquals(TEST_DATA, self.test_job.get_output())152 self.assertEquals(TEST_DURATION,153 self.test_job.load().execution_time_sec)154 # --------------------------------------------------------------------155 # Results from older runs are ignored, even if a seemingly-hung job156 # later completes or fails.157 def test_kill_and_restart_job_old_job_completes(self):158 self._test_kill_and_restart(self.test_job.force_complete_job)159 def test_kill_and_restart_job_old_job_fails(self):160 self._test_kill_and_restart(self.test_job.force_fail_job)161 def _test_kill_and_restart(self, func):162 sequence_num = self.test_job.submit()163 self.test_job.force_start_job(sequence_num)164 self.test_job.cancel()165 sequence_num_2 = self.test_job.submit()166 self.assertEquals(sequence_num_2, sequence_num + 1)167 self.test_job.force_start_job(sequence_num_2)168 self.test_job.force_complete_job(sequence_num_2, TEST_DATA,169 TEST_DURATION)170 # Now try to complete the (long-running) first try.171 # Results from previous run should not overwrite more-recent.172 func(sequence_num, TEST_BAD_DATA, TEST_BAD_DURATION)173 self.assertEquals(TEST_DATA, self.test_job.get_output())174 self.assertEquals(TEST_DURATION,...

Full Screen

Full Screen

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:

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!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful