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

How to use test_job method in autotest

Best Python code snippet using autotest_python

model_jobs.py

Source:model_jobs.py

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,...

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.