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

How to use get_services method in Airtest

Best Python code snippet using Airtest

test.py

Source:test.py

...35    ssl.__ssl__.SSLContext.wrap_socket = wrap_socket36except AttributeError:37    pass38import postservices39def get_services():40    return postservices.TrackingService.service_classes41if __name__ == '__main__':42    import sys43    import codecs44    import locale45    import logging46    # logging.basicConfig(level=logging.DEBUG, format="%(asctime)s - %(levelname)s - %(message)s")47    # sys.stdout = codecs.getwriter(locale.getpreferredencoding())(sys.stdout)48    # service = get_services()['RussianpostRuService']('CG077165712US')49    # service = get_services()['CanparComService']('D100301463601')50    # service = get_services()['ThailandPostCoThService']('RR051974496TH')51    # service = get_services()['BelpostByService']('BV021258988BY')52    # service = get_services()['EmspostRuService']('EB006267877IT')53    # service = get_services()['EmsPostService']('EB006267877IT')54    # service = get_services()['PpxTrackComService']('RRD050599000070847')55    # service = get_services()['CorreosCl']('RP316339727SG')56    # service = get_services()['RussianpostRuService']('RM144847659CN')57    # service = get_services()['AsendiaUsaComService']('LM586057034US')58    # service = PostaSiService('RA316816466SI')59    # service = ParcelForceNetService('EK034679967GB')60    # service = PostdanmarkDkService('TS123456789DK')61    # service = SwisspostChService('RU470249572CH')62    # service = UspsComService('420917419405515901033236726959')63    # service = get_services()['UspsComService']('LN800389358US')64    # service = get_services()['UspsComService']('9400110200883101818261')65    # service = get_services()['UspsComService']('9241990101549234628999')66    # service = get_services()['CyprusPostGovCyService']('CP353366219FR')67    # service = get_services()['CorreosDeMexicoGobMxService']('RB918786998CN')68    # service = get_services()['CttPtService']('RE640520385SE')69    # service = get_services()['CttPtService']('RJ216862295CN')70    # service = get_services()['TransgroupComService']('23B514653')71    # service = get_services()['PosteItService']('EB006267877IT')72    # service = get_services()['FedexComService']('718909941913')73    # service = get_services()['UpsComService']('1Z88Y7Y21203375521')74    service = get_services()['UpsComService']('1Z18424E4406080438')75    # service = get_services()['YanwenComCnService']('11901885104')76    # service = get_services()['OnTracComService']('C11229305254319')77    # service = get_services()['JapanpostJp']('EG483450686JP')78    # service = UkrposhtaComService('RB041697087UA')79    # service = get_services()['DhlDeService']('00340433836170068547')80    # service = get_services()['HermesWorldTrackingService']('77305107155045')81    # service = CorreiosComService('RD024810847SE')82    # service = PrivpakSchenkerNuService('4199854203482')83    # service = SchenkerNuService('6273636354')84    # service = DhlmultishippingSeService('6288206664')85    # service = get_services()['DhlComService']('5008590300')86    # service = get_services()['CanadaPostCaService']('305260344086')87    # service = CorreosEsService('RA514556931CN')88    # service = get_services()['SingPostComService']('RF427535022SG')89    # service = get_services()['IParcelComService']('AEIIND10002594777')90    # service = get_services()['IParcelComService']('AEIPHX60004469610')91    print u'\t'.join((u'Operation', 'Datetime', 'Location'))92    for i in service.fetch():...

test_heartbeats.py

Source:test_heartbeats.py

1from unittest import TestCase, mock2from collections import namedtuple3import datetime4from gobworkflow.heartbeats import on_heartbeat, check_services5class TestHeartbeats(TestCase):6    @mock.patch('gobworkflow.heartbeats.get_services')7    @mock.patch('gobworkflow.heartbeats.update_service')8    def test_on_heartbeat(self, update_service, get_services):9        service = {10            "name": "AnyService",11            "is_alive": True,12            "host": None,13            "pid": None,14            "timestamp": datetime.datetime.now().isoformat(),15        }16        msg = {17            "threads": [18                {19                    "name": "thread1",20                    "is_alive": True21                },22                {23                    "name": "thread2",24                    "is_alive": False25                }26            ]27        }28        msg.update(service)29        on_heartbeat(msg)30        self.assertEqual(update_service.call_count, 1)31        service_parameter, tasks = update_service.call_args[0]32        self.assertEqual(service_parameter, service)33        self.assertEqual(len(tasks), len(msg["threads"]))34    @mock.patch('gobworkflow.heartbeats.update_service')35    @mock.patch('gobworkflow.heartbeats.get_services')36    @mock.patch('gobworkflow.heartbeats.remove_service')37    @mock.patch('gobworkflow.heartbeats.mark_service_dead')38    def test_check_services(self, mark_service_dead, remove_service, get_services, update_service):39        service = {40            "name": "AnyService",41            "is_alive": True,42            "timestamp": (datetime.datetime.now() - datetime.timedelta(hours=1)).isoformat(),43        }44        msg = {45            "threads": [46                {47                    "name": "thread1",48                    "is_alive": True49                },50                {51                    "name": "thread2",52                    "is_alive": False53                }54            ]55        }56        msg.update(service)57        get_services.return_value = []58        on_heartbeat(msg)59        # Assure that the service has been marked dead because of a heartbeat timeout60        self.assertEqual(get_services.call_count, 1)61        self.assertEqual(remove_service.call_count, 0)62        self.assertEqual(mark_service_dead.call_count, 0)63        Service = namedtuple('Service', ['timestamp'])64        service = Service(datetime.datetime.utcnow())65        get_services.return_value = [service]66        get_services.reset_mock()67        on_heartbeat(msg)68        # Assure that the service has been marked dead because of a heartbeat timeout69        self.assertEqual(get_services.call_count, 1)70        self.assertEqual(remove_service.call_count, 0)71        self.assertEqual(mark_service_dead.call_count, 0)72        service = Service(datetime.datetime.utcnow() - datetime.timedelta(minutes=15))73        get_services.return_value = [service]74        get_services.reset_mock()75        on_heartbeat(msg)76        # Assure that the service has been marked dead because of a heartbeat timeout77        self.assertEqual(get_services.call_count, 1)78        self.assertEqual(remove_service.call_count, 0)79        self.assertEqual(mark_service_dead.call_count, 1)80        service = Service(datetime.datetime.utcnow() - datetime.timedelta(days=1))81        get_services.return_value = [service]82        get_services.reset_mock()83        mark_service_dead.reset_mock()84        on_heartbeat(msg)85        # Assure that the service has been marked dead because of a heartbeat timeout86        self.assertEqual(get_services.call_count, 1)87        self.assertEqual(remove_service.call_count, 1)...

test_unit.py

Source:test_unit.py

...6    service = TimedService('dummy')7    service.start()8    time.sleep(0.009)9    service.end()10    assert get_services() == [service]11    assert service.duration == 912    assert service.name == 'dummy'13def test_timed_context_manager(discard):14    with timed('dummy') as service:15        time.sleep(0.008)16        assert get_services() == [service]17    assert get_services()[0].duration == 818    assert get_services()[0].name == 'dummy'19def test_timed_decorator(discard):20    @timed_wrapper('dummy')21    def sleepy_function():22        time.sleep(0.007)23    sleepy_function()24    assert get_services()[0].duration == 725    assert get_services()[0].name == 'dummy'26def test_discard_all_services(discard):27    service = TimedService('drummy')28    service.start()29    time.sleep(0.009)30    service.end()31    assert get_services() == [service]32    discard_all_services()33    assert get_services() == []34def test_add_service(discard):35    service = TimedService('drummy')36    add_service(service)...

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.