Best Python code snippet using autotest_python
tests_model.py
Source:tests_model.py  
1import datetime2from django.test import TestCase3from django.utils.timezone import make_aware4from psycopg2.extras import DateTimeTZRange5from job.models import Job6from utilities import samples7from utilities.Exceptions.job import *8from utilities.Exceptions.timeslot import *9class JobModelTests(TestCase):10    # Job Tests11    def test_executor_with_job_in_range_cannot_create(self):12        """ To check that an executor that has a job already in the13        new job range cannot create a new job in that range14        e.g ==> Existing job: 12-1615                Recess time: 16-1516                New job request: 12-1317        """18        creator = samples.sample_user()19        executor = samples.sample_user(email='tito123@pluto.com')20        # create time slot and  sample job at21        #  ==> 2019, 12, 27, 12, 022        samples.sample_job(creator, executor)23        job_start = make_aware(datetime.datetime(2019, 12, 27, 12, 0))24        job_end = job_start + datetime.timedelta(hours=2)  # 1425        job_duration = DateTimeTZRange(job_start, job_end)26        # create a job within the duration27        job_kwargs = dict(28            creator=creator,29            executor=executor,30            duration=job_duration,31            type='turnover',32            price=30.05,33        )34        with self.assertRaises(JobOverlapError):35            Job.objects.create(**job_kwargs)36    def test_executor_cannot_create_job_in_recess(self):37        """ To check that an executor cannot create a new job in38        a recess time period39        e.g ==> Existing job: 12-1640                Recess time: 16-1541                New job request: 16-1842        """43        creator = samples.sample_user()44        executor = samples.sample_user(email='tito123@pluto.com')45        # create time slot and  sample job at46        #  ==> 2019, 12, 27, 12, 047        time_start= datetime.datetime(2019, 12, 27, 12, 0)48        time_end = time_start + datetime.timedelta(hours=6)  # 1849        t = samples.sample_timeslot(executor, time_start, time_end)50        # create job on timeslot51        job_start = make_aware(datetime.datetime(2019, 12, 27, 12, 0))52        job_end = job_start + datetime.timedelta(hours=4)  # 1653        job_duration = DateTimeTZRange(job_start, job_end)54        # create a job within the duration55        job_kwargs = dict(56            creator=creator,57            executor=executor,58            duration=job_duration,59            type='turnover',60            price=30.05,61        )62        Job.objects.create(**job_kwargs)63        #create new job on a recess time64        job_start = make_aware(datetime.datetime(2019, 12, 27, 16, 0))65        job_end = job_start + datetime.timedelta(hours=2)  # 1766        job_duration = DateTimeTZRange(job_start, job_end)67        # create a job within the duration68        job_kwargs = dict(69            creator=creator,70            executor=executor,71            duration=job_duration,72            type='test',73            price=30.05,74        )75        with self.assertRaises(JobOnRecessError):76            Job.objects.create(**job_kwargs)77    # Timeslot Tests78    def test_executor_with_no_timeslot_cannot_create_job(self):79        """ To check that the executor of job has a80        timeslot valid before job is created81        """82        creator = samples.sample_user()83        executor = samples.sample_user(email='tito123@pluto.com')84        job_start = make_aware(datetime.datetime(2019, 12, 27, 12, 0))85        job_end = job_start + datetime.timedelta(hours=3)  # 1586        job_duration = DateTimeTZRange(job_start, job_end)87        # create a job within the duration88        job_kwargs = dict(89            creator=creator,90            executor=executor,91            duration=job_duration,92            type='turnover',93            price=30.05,94        )95        # expect an error96        with self.assertRaises(TimeSlotsNotFound):97            Job.objects.create(**job_kwargs)98    def test_create_job_user_with_invalid_timeslot_not_successful(self):99        """ Test to create a new job  in which  user has100         DOES NOT have a valid timeslot that matches the101        duration of the job102        """103        creator = samples.sample_user()104        executor = samples.sample_user(email='tito123@pluto.com')105        timeslot_start = make_aware(datetime.datetime(2019, 12, 27, 12, 0))106        timeslot_end = timeslot_start + datetime.timedelta(hours=4)  # 16107        timeslot_period = DateTimeTZRange(timeslot_start, timeslot_end)108        job_start = make_aware(datetime.datetime(2019, 12, 27, 10, 0))109        job_end = job_start + datetime.timedelta(hours=2)  # 12110        job_duration = DateTimeTZRange(job_start, job_end)111        # create a time slot for the executor112        samples.sample_timeslot(executor, period=timeslot_period)113        # create a job within the duration114        job_kwargs = dict(115            creator=creator,116            executor=executor,117            duration=job_duration,118            type='turnover',119            price=30.05,120        )121        with self.assertRaises(TimeSlotsJobMatchError):122            Job.objects.create(**job_kwargs)123    def test_job_creator_executor_not_equal(self):124        """ Test to check that the creator and executor125        of a job entry is not equal to the same user126        """127        creator = samples.sample_user()128        job_start = make_aware(datetime.datetime(2019, 12, 27, 12, 0))129        job_end = job_start + datetime.timedelta(hours=3)  # 15130        job_duration = DateTimeTZRange(job_start, job_end)131        job_kwargs = dict(132            creator=creator,133            executor=creator,134            duration=job_duration,135            type='turnover',136            price=30.05,137        )138        with self.assertRaises(ValueError):139            Job.objects.create(**job_kwargs)140    def test_job_duration_check(self):141        """ To check that a job start time is greater than142        its end time143        """144        creator = samples.sample_user()145        executor = samples.sample_user(email='tito123@pluto.com')146        start = make_aware(datetime.datetime(2019, 12, 27, 12, 0))147        end = make_aware(datetime.datetime(2019, 12, 27, 11, 0))  # -1148        duration = DateTimeTZRange(start, end)149        job_kwargs = dict(150            creator=creator,151            executor=executor,152            duration=duration,153            type='turnover',154            price=30.05,155        )156        with self.assertRaises(ValueError):157            Job.objects.create(**job_kwargs)158    def test_create_new_job_successful(self):159        """ Test to create a new job  in which  user has160         a valid timeslot that matches the161        duration of the job162        """163        creator = samples.sample_user()164        executor = samples.sample_user(email='tito123@pluto.com')165        timeslot_start = make_aware(datetime.datetime(2019, 12, 27, 12, 0))166        timeslot_end = timeslot_start + datetime.timedelta(hours=4)167        timeslot_period = DateTimeTZRange(timeslot_start, timeslot_end)168        job_start = make_aware(datetime.datetime(2019, 12, 27, 12, 0))169        job_end = job_start + datetime.timedelta(hours=3)170        job_duration = DateTimeTZRange(job_start, job_end)171        # create a time slot for the executor172        samples.sample_timeslot(executor, period=timeslot_period)173        # create a job within the duration174        job_kwargs = dict(175            creator=creator,176            executor=executor,177            duration=job_duration,178            type='turnover',179            price=30.05,180        )181        job = Job.objects.create(**job_kwargs)182        self.assertEqual(job.creator, job_kwargs['creator'])183        self.assertEqual(job.executor, job_kwargs['executor'])184        self.assertEqual(job.duration, job_kwargs['duration'])185        self.assertEqual(Job.objects.all().count(), 1)186    def test_create_new_job_successful2(self):187        """ Test to create a new job  in which  user has188           valid timeslots in different time ranges189          e.g ==> 12-16 and 19-23190          Job duration ==> 13-16191        """192        creator = samples.sample_user()193        executor = samples.sample_user(email='tito123@pluto.com')194        timeslot_start1 = make_aware(datetime.datetime(2019, 12, 27, 12, 0))195        timeslot_end1 = timeslot_start1 + datetime.timedelta(hours=4)  # 16196        timeslot_period1 = DateTimeTZRange(timeslot_start1, timeslot_end1)197        timeslot_start2 = make_aware(datetime.datetime(2019, 12, 27, 19, 0))198        timeslot_end2 = timeslot_start2 + datetime.timedelta(hours=4)  # 23199        timeslot_period2 = DateTimeTZRange(timeslot_start2, timeslot_end2)200        job_start = make_aware(datetime.datetime(2019, 12, 27, 13, 0))201        job_end = job_start + datetime.timedelta(hours=3)202        job_duration = DateTimeTZRange(job_start, job_end)203        # create a time slot for the executor204        samples.sample_timeslot(executor, period=timeslot_period1)205        samples.sample_timeslot(executor, period=timeslot_period2)206        # create a job within the duration207        job_kwargs = dict(208            creator=creator,209            executor=executor,210            duration=job_duration,211            type='turnover',212            price=30.05,213        )214        job = Job.objects.create(**job_kwargs)215        self.assertEqual(job.creator, job_kwargs['creator'])216        self.assertEqual(job.executor, job_kwargs['executor'])217        self.assertEqual(job.duration, job_kwargs['duration'])218        self.assertEqual(Job.objects.all().count(), 1)219    def test_create_new_job_successful3(self):220        """ Test to create a new job  in which  user has221          have a valid timeslots that are continuous222          e.g ==> 12-4 and 4-8223          and job time overlaps between the timeslots224        """225        creator = samples.sample_user()226        executor = samples.sample_user(email='tito123@pluto.com')227        timeslot_start1 = make_aware(datetime.datetime(2019, 12, 27, 12, 0))228        timeslot_end1 = timeslot_start1 + datetime.timedelta(hours=4)  # 16229        timeslot_period1 = DateTimeTZRange(timeslot_start1, timeslot_end1)230        timeslot_start2 = make_aware(datetime.datetime(2019, 12, 27, 16, 0))231        timeslot_end2 = timeslot_start2 + datetime.timedelta(hours=4)  # 20232        timeslot_period2 = DateTimeTZRange(timeslot_start2, timeslot_end2)233        # offset timeslot (not in job duration range)234        timeslot_start3 = make_aware(datetime.datetime(2019, 12, 27, 8, 0))235        timeslot_end3 = timeslot_start3 + datetime.timedelta(hours=3)  # 11236        timeslot_period3 = DateTimeTZRange(timeslot_start3, timeslot_end3)237        job_start = make_aware(datetime.datetime(2019, 12, 27, 14, 0))238        job_end = job_start + datetime.timedelta(hours=18)  # 18239        job_duration = DateTimeTZRange(job_start, job_end)240        # create  timeslots for the executor241        samples.sample_timeslot(executor, period=timeslot_period1)242        samples.sample_timeslot(executor, period=timeslot_period2)243        samples.sample_timeslot(executor, period=timeslot_period3)244        # create a job with start time end time out of the245        # executor's time slot246        job_kwargs = dict(247            creator=creator,248            executor=executor,249            duration=job_duration,250            type='turnover',251            price=30.05,252        )253        job = Job.objects.create(**job_kwargs)254        self.assertEqual(job.creator, job_kwargs['creator'])255        self.assertEqual(job.executor, job_kwargs['executor'])256        self.assertEqual(job.duration, job_kwargs['duration'])257        self.assertEqual(Job.objects.all().count(), 1)258    def test_create_job_non_continuous_timeslot_unsuccessful(self):259        """ Test to create a new job  in which  user has260          have a valid 2 timeslots that are not continuous261          e.g ==> 12-16 and 17-21262          and job time overlaps between the timeslots263          e.g ==> 15-18264        """265        creator = samples.sample_user()266        executor = samples.sample_user(email='tito123@pluto.com')267        timeslot_start1 = make_aware(datetime.datetime(2019, 12, 27, 12, 0))268        timeslot_end1 = timeslot_start1 + datetime.timedelta(hours=4)  # 16269        timeslot_period1 = DateTimeTZRange(timeslot_start1, timeslot_end1)270        timeslot_start2 = make_aware(datetime.datetime(2019, 12, 27, 17, 0))271        timeslot_end2 = timeslot_start2 + datetime.timedelta(hours=4)  # 21272        timeslot_period2 = DateTimeTZRange(timeslot_start2, timeslot_end2)273        job_start = make_aware(datetime.datetime(2019, 12, 27, 15, 0))274        job_end = job_start + datetime.timedelta(hours=3)  # 18275        job_duration = DateTimeTZRange(job_start, job_end)276        # create  timeslots for the executor277        samples.sample_timeslot(executor, period=timeslot_period1)278        samples.sample_timeslot(executor, period=timeslot_period2)279        # create a job with start time/ end time out of280        # the executor's time slot281        job_kwargs = dict(282            creator=creator,283            executor=executor,284            duration=job_duration,285            type='turnover',286            price=30.05,287        )288        with self.assertRaises(TimeSlotsNotContinuousError):289            Job.objects.create(**job_kwargs)290    def test_create_job_non_continuous_timeslot_unsuccessful2(self):291        """ Test to create a new job  in which  user has292          have a valid 3 timeslots that are not continuous293          e.g ==> 11-12 and 13-15 and 15-18294          and job time overlaps between the timeslots295          e.g ==> 11-17296        """297        creator = samples.sample_user()298        executor = samples.sample_user(email='tito123@pluto.com')299        timeslot_start1 = make_aware(datetime.datetime(2019, 12, 27, 11, 0))300        timeslot_end1 = timeslot_start1 + datetime.timedelta(hours=1)  # 12301        timeslot_period1 = DateTimeTZRange(timeslot_start1, timeslot_end1)302        timeslot_start2 = make_aware(datetime.datetime(2019, 12, 27, 13, 0))303        timeslot_end2 = timeslot_start2 + datetime.timedelta(hours=2)  # 15304        timeslot_period2 = DateTimeTZRange(timeslot_start2, timeslot_end2)305        timeslot_start3 = make_aware(datetime.datetime(2019, 12, 27, 15, 0))306        timeslot_end3 = timeslot_start3 + datetime.timedelta(hours=3)  # 18307        timeslot_period3 = DateTimeTZRange(timeslot_start3, timeslot_end3)308        job_start = make_aware(datetime.datetime(2019, 12, 27, 11, 0))309        job_end = job_start + datetime.timedelta(hours=6)  # 17310        job_duration = DateTimeTZRange(job_start, job_end)311        # create  timeslots for the executor312        samples.sample_timeslot(executor, period=timeslot_period1)313        samples.sample_timeslot(executor, period=timeslot_period2)314        samples.sample_timeslot(executor, period=timeslot_period3)315        # create a job with start time/ end time out of the316        # executor's time slot317        job_kwargs = dict(318            creator=creator,319            executor=executor,320            duration=job_duration,321            type='turnover',322            price=30.05,323        )324        with self.assertRaises(TimeSlotsNotContinuousError):...test_simulation_multi.py
Source:test_simulation_multi.py  
1#2# Copyright 2013 The Regents of The University California3# 4# Licensed under the Apache License, Version 2.0 (the "License");5# you may not use this file except in compliance with the License.6# You may obtain a copy of the License at7# 8#   http://www.apache.org/licenses/LICENSE-2.09# 10# Unless required by applicable law or agreed to in writing, software11# distributed under the License is distributed on an "AS IS" BASIS,12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.13# See the License for the specific language governing permissions and14# limitations under the License.15#16import unittest17import simulation_multi18class TestMultiGetSimulation(unittest.TestCase):19    def setUp(self):20        simulation_multi.TOTAL_WORKERS = 1021        self.simulation = simulation_multi.Simulation(5, "", 0.9)22    def test_basic(self):23        TASKS_PER_JOB = 324        # ADd a job.25        JOB_ID = 826        JOB_START = 1027        job = simulation_multi.Job(TASKS_PER_JOB, JOB_START)28        self.simulation.jobs[JOB_ID] = job29        self.assertEqual(job.num_tasks, TASKS_PER_JOB)30        self.assertEqual(len(job.unscheduled_tasks), TASKS_PER_JOB)31        # At this point, all workers are idle.32        worker = self.simulation.workers[0]33        # Add a probe for job 0 at 10 millis in.34        events = worker.add_probe(JOB_ID, JOB_START)35        # Should return 4 events: one task ending event for each slot and one no-op.36        self.assertEqual(len(events), 4)37        self.assertEqual(worker.num_free_slots, 0)38        # Run the nop-op event.39        events[-1][1].run(JOB_START + 2)40        self.assertEqual(worker.num_free_slots, 1)41        JOB_ID_2 = 1542        JOB_START_2 = 1543        job2 = simulation_multi.Job(TASKS_PER_JOB, JOB_START_2)44        self.simulation.jobs[JOB_ID_2] = job245        events = worker.add_probe(JOB_ID_2, JOB_START_2)46        self.assertEqual(worker.num_free_slots, 0)47        # One job start event48        self.assertEqual(len(events), 1)49        # Fake the completion of the first task some time later;50        # should launch the second one.51        new_events = events[0][1].run(JOB_START_2 + 5)52        # Don't keep the job around at this point.53        self.assertEqual(len(new_events), 0)54    def test_multiple_slots_released(self):55        """Ensures that multiple slots are released when a noop comes back."""56        JOB_ID = 2057        JOB_START= 5058        job = simulation_multi.Job(2, JOB_START)59        self.simulation.jobs[JOB_ID] = job60        worker = self.simulation.workers[1]61        self.assertEqual(worker.num_free_slots, 4)62        events = worker.add_probe(JOB_ID, JOB_START)63        self.assertEqual(worker.num_free_slots, 0)64        # The events shoudl include 2 task end events and 1 noop.65        self.assertEqual(len(events), 3)66        # Run the noop event.67        events[-1][1].run(events[-1][0])68        self.assertEqual(worker.num_free_slots, 2)69if __name__ == '__main__':...get_pause.py
Source:get_pause.py  
1from django.utils.datetime_safe import datetime2def get_pause(model_settings) -> bool:3    '''4    ÑÑнкÑÐ¸Ñ Ð¿ÑовеÑÑÐµÑ Ð¼Ð¾Ð¶Ð½Ð¾ ли ÑейÑÐ°Ñ Ð¾Ð±Ð¼ÐµÐ½Ð½Ð¸ÐºÑ ÑабоÑаÑÑ Ð¸Ð»Ð¸ он ÑÑÐ¾Ð¸Ñ Ð½Ð° паÑзе5    '''6    if not bool(model_settings):7        return True8    if model_settings.pause or model_settings.reload_exchange:9        return True10    now_hour = datetime.now().hour11    job_start = model_settings.job_start12    job_end = model_settings.job_end13    if job_start == job_end or (job_start == 0 and job_end == 24) or (job_start == 24 and job_end == 0):14        return False15    elif job_start < job_end:16        if now_hour < job_start or now_hour > job_end:17            return True18    else:  # job_start > job_end:19        if now_hour < job_start and now_hour >= job_end:20            return True...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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
