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

How to use get_health_check method in localstack

Best Python code snippet using localstack_python

test_scheduler.py

Source:test_scheduler.py

...106        self.assertEqual(device.current_job(), None)107    def test_without_health_checks(self):108        # Make sure that get_health_check does return None109        Device.get_health_check = lambda cls: None110        self.assertEqual(self.device01.get_health_check(), None)111        self.assertEqual(self.device02.get_health_check(), None)112        self.assertEqual(self.device03.get_health_check(), None)113        # Schedule without health check114        available_devices = schedule_health_checks(DummyLogger())[0]115        self.assertEqual(available_devices, {"panda": ["panda01", "panda03"]})116    def test_disabled_hc(self):117        # Make sure that get_health_check does return something118        Device.get_health_check = _minimal_valid_job119        self.assertNotEqual(self.device01.get_health_check(), None)120        self.assertNotEqual(self.device02.get_health_check(), None)121        self.assertNotEqual(self.device03.get_health_check(), None)122        self.device_type01.disable_health_check = True123        self.device_type01.save()124        available_devices = schedule_health_checks(DummyLogger())[0]125        self.assertEqual(available_devices, {"panda": ["panda01", "panda03"]})126    def test_no_devicedict(self):127        Device.get_health_check = _minimal_valid_job128        self.device_type02.disable_health_check = False129        self.device_type02.display = True130        self.device_type02.save()131        self.device04.state = Device.STATE_IDLE132        self.device04.health = Device.HEALTH_UNKNOWN133        self.device04.save()134        schedule_health_checks(DummyLogger())135        self.device04.refresh_from_db()136        self.assertFalse(self.device04.is_valid())137        self.assertEqual(self.device04.health, Device.HEALTH_BAD)138        self.assertIsNone(self.device04.current_job())139    def test_without_previous_hc_device_health_unknown(self):140        # Make sure that get_health_check does return something141        Device.get_health_check = _minimal_valid_job142        self.assertNotEqual(self.device01.get_health_check(), None)143        self.assertNotEqual(self.device02.get_health_check(), None)144        self.assertNotEqual(self.device03.get_health_check(), None)145        available_devices = schedule_health_checks(DummyLogger())[0]146        self.assertEqual(available_devices, {"panda": []})147        self._check_hc_scheduled(self.device01)148        self._check_hc_not_scheduled(self.device02)149        self._check_hc_scheduled(self.device03)150    def test_device_health_good(self):151        # Make sure that get_health_check does return something152        Device.get_health_check = _minimal_valid_job153        self.assertNotEqual(self.device01.get_health_check(), None)154        self.assertNotEqual(self.device02.get_health_check(), None)155        self.assertNotEqual(self.device03.get_health_check(), None)156        self.device01.health = Device.HEALTH_GOOD157        self.device01.save()158        self.device02.health = Device.HEALTH_GOOD159        self.device02.save()160        self.device03.health = Device.HEALTH_GOOD161        self.device03.save()162        available_devices = schedule_health_checks(DummyLogger())[0]163        self.assertEqual(available_devices, {"panda": ["panda03"]})164        self._check_hc_scheduled(self.device01)165        self._check_hc_not_scheduled(self.device02)166        self._check_hc_not_scheduled(self.device03)167    def test_device_health_looping(self):168        # Make sure that get_health_check does return something169        Device.get_health_check = _minimal_valid_job170        self.assertNotEqual(self.device01.get_health_check(), None)171        self.assertNotEqual(self.device02.get_health_check(), None)172        self.assertNotEqual(self.device03.get_health_check(), None)173        self.device01.health = Device.HEALTH_LOOPING174        self.device01.save()175        self.device02.health = Device.HEALTH_LOOPING176        self.device02.save()177        self.device03.health = Device.HEALTH_LOOPING178        self.device03.save()179        available_devices = schedule_health_checks(DummyLogger())[0]180        self.assertEqual(available_devices, {"panda": []})181        self._check_hc_scheduled(self.device01)182        self._check_hc_not_scheduled(self.device02)183        self._check_hc_scheduled(self.device03)184    def test_device_health_wrong(self):185        # Make sure that get_health_check does return something186        Device.get_health_check = _minimal_valid_job187        self.assertNotEqual(self.device01.get_health_check(), None)188        self.assertNotEqual(self.device02.get_health_check(), None)189        self.assertNotEqual(self.device03.get_health_check(), None)190        # HEALTH_(BAD|MAINTENANCE|RETIRED)191        for health in [192            Device.HEALTH_BAD,193            Device.HEALTH_MAINTENANCE,194            Device.HEALTH_RETIRED,195        ]:196            self.device01.health = health197            self.device01.save()198            self.device02.health = health199            self.device02.save()200            self.device03.health = health201            self.device03.save()202            available_devices = schedule_health_checks(DummyLogger())[0]203            self.assertEqual(available_devices, {"panda": []})204            self._check_hc_not_scheduled(self.device01)205            self._check_hc_not_scheduled(self.device02)206            self._check_hc_not_scheduled(self.device03)207    def test_health_frequency_hours(self):208        self.device_type01.health_denominator = DeviceType.HEALTH_PER_HOUR209        self.device_type01.health_frequency = 24210        self.device_type01.save()211        Device.get_health_check = _minimal_valid_job212        self.assertNotEqual(self.device01.get_health_check(), None)213        self.assertNotEqual(self.device02.get_health_check(), None)214        self.assertNotEqual(self.device03.get_health_check(), None)215        # Only device03 is available now216        self.device01.health = Device.HEALTH_BAD217        self.device01.save()218        self.assertTrue(self.device01.is_valid())219        self.device03.health = Device.HEALTH_GOOD220        self.device03.save()221        self.assertTrue(self.device03.is_valid())222        # Create a job that should be scheduled now223        j = TestJob.objects.create(224            requested_device_type=self.device_type01,225            submitter=self.user,226            definition=_minimal_valid_job(None),227        )228        schedule(DummyLogger())229        self.device01.refresh_from_db()230        j.refresh_from_db()231        self.assertEqual(j.state, TestJob.STATE_SCHEDULED)232        self.assertEqual(j.actual_device, self.device03)233        j.go_state_finished(TestJob.HEALTH_COMPLETE)234        j.save()235        # Create a job that should be scheduled after the health check236        j = TestJob.objects.create(237            requested_device_type=self.device_type01,238            submitter=self.user,239            definition=_minimal_valid_job(None),240        )241        self.device03.refresh_from_db()242        self.last_hc03.submit_time = timezone.now() - timedelta(hours=25)243        self.last_hc03.save()244        schedule(DummyLogger())245        self.device03.refresh_from_db()246        j.refresh_from_db()247        self.assertEqual(j.state, TestJob.STATE_SUBMITTED)248        current_hc = self.device03.current_job()249        self.assertTrue(current_hc.health_check)250        self.assertEqual(current_hc.state, TestJob.STATE_SCHEDULED)251    def test_health_frequency_jobs(self):252        self.device_type01.health_denominator = DeviceType.HEALTH_PER_JOB253        self.device_type01.health_frequency = 2254        self.device_type01.save()255        self.last_hc03.submit_time = timezone.now() - timedelta(hours=2)256        self.last_hc03.save()257        Device.get_health_check = _minimal_valid_job258        self.assertNotEqual(self.device01.get_health_check(), None)259        self.assertNotEqual(self.device02.get_health_check(), None)260        self.assertNotEqual(self.device03.get_health_check(), None)261        # Only device03 is available now262        self.device01.health = Device.HEALTH_BAD263        self.device01.save()264        self.device03.health = Device.HEALTH_GOOD265        self.device03.save()266        # Create a job that should be scheduled now267        j01 = TestJob.objects.create(268            requested_device_type=self.device_type01,269            submitter=self.user,270            definition=_minimal_valid_job(None),271        )272        j02 = TestJob.objects.create(273            requested_device_type=self.device_type01,274            submitter=self.user,275            definition=_minimal_valid_job(None),276        )277        j03 = TestJob.objects.create(278            requested_device_type=self.device_type01,279            submitter=self.user,280            definition=_minimal_valid_job(None),281        )282        schedule(DummyLogger())283        self.device03.refresh_from_db()284        j01.refresh_from_db()285        self.assertEqual(j01.state, TestJob.STATE_SCHEDULED)286        self.assertEqual(j01.actual_device, self.device03)287        j01.go_state_finished(TestJob.HEALTH_COMPLETE)288        j01.start_time = timezone.now() - timedelta(hours=1)289        j01.save()290        schedule(DummyLogger())291        self.device03.refresh_from_db()292        j02.refresh_from_db()293        self.assertEqual(j02.state, TestJob.STATE_SCHEDULED)294        self.assertEqual(j02.actual_device, self.device03)295        j02.go_state_finished(TestJob.HEALTH_COMPLETE)296        j02.start_time = timezone.now() - timedelta(hours=1)297        j02.save()298        schedule(DummyLogger())299        self.device03.refresh_from_db()300        j03.refresh_from_db()301        self.assertEqual(j03.state, TestJob.STATE_SUBMITTED)302        current_hc = self.device03.current_job()303        self.assertTrue(current_hc.health_check)304        self.assertEqual(current_hc.state, TestJob.STATE_SCHEDULED)305class TestVisibility(TestCase):306    def setUp(self):307        self.worker01 = Worker.objects.create(308            hostname="worker-01", state=Worker.STATE_ONLINE309        )310        self.worker02 = Worker.objects.create(311            hostname="worker-02", state=Worker.STATE_OFFLINE312        )313        self.worker03 = Worker.objects.create(314            hostname="worker-03", state=Worker.STATE_ONLINE315        )316        self.device_type01 = DeviceType.objects.create(name="panda")317        self.device01 = Device.objects.create(318            hostname="panda01",319            device_type=self.device_type01,320            worker_host=self.worker01,321            health=Device.HEALTH_UNKNOWN,322        )323        # This device should never be considered (his worker is OFFLINE)324        self.device02 = Device.objects.create(325            hostname="panda02",326            device_type=self.device_type01,327            worker_host=self.worker02,328            health=Device.HEALTH_UNKNOWN,329        )330        self.device03 = Device.objects.create(331            hostname="panda03",332            device_type=self.device_type01,333            worker_host=self.worker03,334            health=Device.HEALTH_UNKNOWN,335        )336        self.user = User.objects.create(username="user-01")337        self.device03.save()338        self.original_health_check = Device.get_health_check339        # Make sure that get_health_check does return something340        Device.get_health_check = _minimal_valid_job341    def tearDown(self):342        Device.get_health_check = self.original_health_check343        for job in TestJob.objects.filter(344            state__in=[TestJob.STATE_SUBMITTED, TestJob.STATE_SCHEDULING]345        ):346            job.go_state_finished(TestJob.HEALTH_COMPLETE)347    def _minimal_personal_job(self):348        return """349job_name: minimal valid job350visibility: personal351timeouts:352  job:353    minutes: 10354  action:355    minutes: 5356actions: []357"""358    def _check_hc_scheduled(self, device):359        device.refresh_from_db()360        self.assertEqual(device.state, Device.STATE_RESERVED)361        job = device.current_job()362        self.assertNotEqual(job, None)363        self.assertEqual(job.state, TestJob.STATE_SCHEDULED)364        self.assertEqual(job.health, TestJob.HEALTH_UNKNOWN)365        self.assertEqual(job.actual_device, device)366    def _check_hc_not_scheduled(self, device):367        device.refresh_from_db()368        self.assertEqual(device.state, Device.STATE_IDLE)369        self.assertEqual(device.current_job(), None)370    def _check_initial_state(self):371        self.assertNotEqual(self.device01.get_health_check(), None)372        self.assertNotEqual(self.device02.get_health_check(), None)373        self.assertNotEqual(self.device03.get_health_check(), None)374        self.assertEqual(self.device01.health, Device.HEALTH_UNKNOWN)375        self.assertEqual(self.device02.health, Device.HEALTH_UNKNOWN)376        self.assertEqual(self.device03.health, Device.HEALTH_UNKNOWN)377    def test_health_visibility(self):378        self._check_initial_state()379        self.device_type01.disable_health_check = False380        self.device_type01.save()381        schedule_health_checks(DummyLogger())[0]382        self._check_hc_scheduled(self.device01)383        self._check_hc_not_scheduled(self.device02)384        self._check_hc_scheduled(self.device03)385    def test_health_visibility_some_restricted(self):386        self._check_initial_state()387        self.device_type01.disable_health_check = False388        self.device_type01.save()389        schedule_health_checks(DummyLogger())[0]390        self._check_hc_scheduled(self.device01)391        self._check_hc_not_scheduled(self.device02)392        # device03 is restricted in setUp393        self._check_hc_scheduled(self.device03)394    def test_health_visibility_all_restricted(self):395        self._check_initial_state()396        self.device_type01.disable_health_check = False397        self.device_type01.save()398        schedule_health_checks(DummyLogger())[0]399        self._check_hc_scheduled(self.device01)400        self._check_hc_not_scheduled(self.device02)401        self._check_hc_scheduled(self.device03)402class TestPriorities(TestCase):403    def setUp(self):404        self.worker01 = Worker.objects.create(405            hostname="worker-01", state=Worker.STATE_ONLINE406        )407        self.device_type01 = DeviceType.objects.create(name="panda")408        self.device01 = Device.objects.create(409            hostname="panda01",410            device_type=self.device_type01,411            worker_host=self.worker01,412            health=Device.HEALTH_GOOD,413        )414        self.user = User.objects.create(username="user-01")415        self.original_health_check = Device.get_health_check416    def tearDown(self):417        Device.get_health_check = self.original_health_check418    def _check_job(self, job, state, actual_device=None):419        job.refresh_from_db()420        self.assertEqual(job.state, state)421        self.assertEqual(job.actual_device, actual_device)422    def test_low_medium_high_without_hc(self):423        # Disable health checks424        Device.get_health_check = lambda cls: None425        jobs = []426        for p in [427            TestJob.LOW,428            TestJob.MEDIUM,429            TestJob.HIGH,430            TestJob.MEDIUM,431            TestJob.LOW,432            40,433        ]:434            j = TestJob.objects.create(435                requested_device_type=self.device_type01,436                submitter=self.user,437                definition=_minimal_valid_job(None),438                priority=p,439            )440            jobs.append(j)441        log = DummyLogger()442        schedule(log)443        self.device01.refresh_from_db()444        self.assertEqual(self.device01.state, Device.STATE_RESERVED)445        self._check_job(jobs[0], TestJob.STATE_SUBMITTED)446        self._check_job(jobs[1], TestJob.STATE_SUBMITTED)447        self._check_job(jobs[2], TestJob.STATE_SCHEDULED, self.device01)448        self._check_job(jobs[3], TestJob.STATE_SUBMITTED)449        self._check_job(jobs[4], TestJob.STATE_SUBMITTED)450        self._check_job(jobs[5], TestJob.STATE_SUBMITTED)451        jobs[2].go_state_finished(TestJob.HEALTH_COMPLETE)452        jobs[2].save()453        self._check_job(jobs[2], TestJob.STATE_FINISHED, self.device01)454        schedule(log)455        self.device01.refresh_from_db()456        self.assertEqual(self.device01.state, Device.STATE_RESERVED)457        self._check_job(jobs[0], TestJob.STATE_SUBMITTED)458        self._check_job(jobs[1], TestJob.STATE_SCHEDULED, self.device01)459        self._check_job(jobs[2], TestJob.STATE_FINISHED, self.device01)460        self._check_job(jobs[3], TestJob.STATE_SUBMITTED)461        self._check_job(jobs[4], TestJob.STATE_SUBMITTED)462        self._check_job(jobs[5], TestJob.STATE_SUBMITTED)463        jobs[1].go_state_finished(TestJob.HEALTH_COMPLETE)464        jobs[1].save()465        self._check_job(jobs[1], TestJob.STATE_FINISHED, self.device01)466        schedule(log)467        self.device01.refresh_from_db()468        self.assertEqual(self.device01.state, Device.STATE_RESERVED)469        self._check_job(jobs[0], TestJob.STATE_SUBMITTED)470        self._check_job(jobs[1], TestJob.STATE_FINISHED, self.device01)471        self._check_job(jobs[2], TestJob.STATE_FINISHED, self.device01)472        self._check_job(jobs[3], TestJob.STATE_SCHEDULED, self.device01)473        self._check_job(jobs[4], TestJob.STATE_SUBMITTED)474        self._check_job(jobs[5], TestJob.STATE_SUBMITTED)475        jobs[3].go_state_finished(TestJob.HEALTH_COMPLETE)476        jobs[3].save()477        self._check_job(jobs[3], TestJob.STATE_FINISHED, self.device01)478        schedule(log)479        self.device01.refresh_from_db()480        self.assertEqual(self.device01.state, Device.STATE_RESERVED)481        self._check_job(jobs[0], TestJob.STATE_SUBMITTED)482        self._check_job(jobs[1], TestJob.STATE_FINISHED, self.device01)483        self._check_job(jobs[2], TestJob.STATE_FINISHED, self.device01)484        self._check_job(jobs[3], TestJob.STATE_FINISHED, self.device01)485        self._check_job(jobs[4], TestJob.STATE_SUBMITTED)486        self._check_job(jobs[5], TestJob.STATE_SCHEDULED, self.device01)487        jobs[5].go_state_finished(TestJob.HEALTH_COMPLETE)488        jobs[5].save()489        self._check_job(jobs[5], TestJob.STATE_FINISHED, self.device01)490        schedule(log)491        self.device01.refresh_from_db()492        self.assertEqual(self.device01.state, Device.STATE_RESERVED)493        self._check_job(jobs[0], TestJob.STATE_SCHEDULED, self.device01)494        self._check_job(jobs[1], TestJob.STATE_FINISHED, self.device01)495        self._check_job(jobs[2], TestJob.STATE_FINISHED, self.device01)496        self._check_job(jobs[3], TestJob.STATE_FINISHED, self.device01)497        self._check_job(jobs[4], TestJob.STATE_SUBMITTED)498        self._check_job(jobs[5], TestJob.STATE_FINISHED, self.device01)499        jobs[0].go_state_finished(TestJob.HEALTH_COMPLETE)500        jobs[0].save()501        self._check_job(jobs[0], TestJob.STATE_FINISHED, self.device01)502        schedule(log)503        self.device01.refresh_from_db()504        self.assertEqual(self.device01.state, Device.STATE_RESERVED)505        self._check_job(jobs[0], TestJob.STATE_FINISHED, self.device01)506        self._check_job(jobs[1], TestJob.STATE_FINISHED, self.device01)507        self._check_job(jobs[2], TestJob.STATE_FINISHED, self.device01)508        self._check_job(jobs[3], TestJob.STATE_FINISHED, self.device01)509        self._check_job(jobs[4], TestJob.STATE_SCHEDULED, self.device01)510        self._check_job(jobs[5], TestJob.STATE_FINISHED, self.device01)511    def test_low_medium_high_with_hc(self):512        # Enable health checks513        self.device_type01.health_denominator = DeviceType.HEALTH_PER_HOUR514        self.device_type01.health_frequency = 24515        self.device_type01.save()516        Device.get_health_check = _minimal_valid_job517        self.assertNotEqual(self.device01.get_health_check(), None)518        jobs = []519        for p in [520            TestJob.LOW,521            TestJob.MEDIUM,522            TestJob.HIGH,523            TestJob.MEDIUM,524            TestJob.LOW,525        ]:526            j = TestJob.objects.create(527                requested_device_type=self.device_type01,528                submitter=self.user,529                definition=_minimal_valid_job(None),530                priority=p,531            )...

gcp-compute-lb-add-group

Source:gcp-compute-lb-add-group

...16    backend_service_name = f'{lb_name}-backend-service'17    health_check_name = f'{lb_name}-health-checks'18    get_health_check = cli_gcp.compute().healthChecks().get(project=project, healthCheck=health_check_name).execute19    get_health_check = retry(get_health_check, exponent=1.2, allowed_exception_fn=lambda e: getattr(getattr(e, "resp", None), "status", None) == 404)20    health_check = get_health_check()21    group_name = f'{name}-instance-group-manager-{zone}{suffix}'22    cli_gcp.ensure.backend_has_instance_group(verbose, project, zone, backend_service_name, group_name, balancing_mode, cli_gcp.url(health_check))23if __name__ == '__main__':24    with cli_gcp.setup():...

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.