Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use stop_service method in lisa

Best Python code snippet using lisa_python

xdbus.py

Source:xdbus.py

...28            self.dbusMonitor.start()29            self.oe.dbg_log('xdbus::start_service', 'exit_function', self.oe.LOGDEBUG)30        except Exception as e:31            self.oe.dbg_log('xdbus::start_service', 'ERROR: (' + repr(e) + ')', self.oe.LOGERROR)32    def stop_service(self):33        try:34            self.oe.dbg_log('xdbus::stop_service', 'enter_function', self.oe.LOGDEBUG)35            if hasattr(self, 'dbusMonitor'):36                self.dbusMonitor.stop()37                del self.dbusMonitor38            self.oe.dbg_log('xdbus::stop_service', 'exit_function', self.oe.LOGDEBUG)39        except Exception as e:40            self.oe.dbg_log('xdbus::stop_service', 'ERROR: (' + repr(e) + ')')41    def exit(self):42        pass43    def restart(self):44        try:45            self.oe.dbg_log('xdbus::restart', 'enter_function', self.oe.LOGDEBUG)46            self.stop_service()47            self.start_service()48            self.oe.dbg_log('xdbus::restart', 'exit_function', self.oe.LOGDEBUG)49        except Exception as e:50            self.oe.dbg_log('xdbus::restart', 'ERROR: (' + repr(e) + ')')51class dbusMonitor(threading.Thread):52    def __init__(self, oeMain):53        try:54            oeMain.dbg_log('xdbus::dbusMonitor::__init__', 'enter_function', oeMain.LOGDEBUG)55            self.monitors = []56            self.oe = oeMain57            self.dbusSystemBus = oeMain.dbusSystemBus58            dbus.mainloop.glib.threads_init()59            DBusGMainLoop(set_as_default=True)60            self.mainLoop = GLib.MainLoop()...

dynamic.py

Source:dynamic.py

...13        return s14    def current_services(self):15        return list(self.services)16    def after_task(self, task, service):17        self.provider.stop_service(service)18        # self.services.discard(service)19class SingleEngine(EngineBase):20    def __init__(self, provider, **kwargs):21        super(SingleEngine, self).__init__(provider, **kwargs)22        self.conf = self.provider.configurations()[0]23        self.service = None24    def before_eval(self):25        self.service = self.provider.start_service(1, self.conf)26    def after_eval(self):27        self.provider.stop_service(self.service)28        self.service = None29    def which_service(self, task):30        return self.service31    def current_services(self):32        return [self.service]33class LimitEngine(EngineBase):34    def __init__(self, provider, n, **kwargs):35        super(LimitEngine, self).__init__(provider, **kwargs)36        self.conf = self.provider.configurations()[0]37        self.n = n38        self.services = []39        self.num_unscheduled = -140        self.le_lock = Semaphore()41    def before_eval(self):42        self.num_unscheduled = len(self.dag)43    def after_eval(self):44        for s in self.services:45            self.provider.stop_service(s)46        self.services = []47    def which_service(self, task):48        self.num_unscheduled -= 149        for s in self.services:50            if len(s.tasks) == 0:51                return s52        if len(self.services) < self.n:53            s = self.provider.start_service(len(self.services) + 1, self.conf)54            self.services.append(s)55        else:56            s = min(self.services, key=lambda x: len(x.tasks))57        return s58    def after_task(self, task, service):59        self.le_lock.acquire()60        if self.num_unscheduled == 0 and len(service.tasks) == 0:61            self.provider.stop_service(service)62            for s in self.services:63                if len(s.tasks) == 0:64                    self.provider.stop_service(s)65        self.le_lock.release()66    def current_services(self):...

test_webmgr.py

Source:test_webmgr.py

1# Copyright (c) 2017-2019 David Steele <dsteele@gmail.com>2#3# SPDX-License-Identifier: GPL-2.0-or-later4# License-Filename: LICENSE5import pytest6from mock import patch, call7from comitup import webmgr8def test_webmgr_callback_target():9    assert webmgr.callback_target() == webmgr.state_callback10@pytest.fixture()11def websvc_fxt(request):12    web_svc = webmgr.web_service13    def fin():14        webmgr.web_service = web_svc15    request.addfinalizer(fin)16@pytest.mark.parametrize("state, action, fn_fact, arg_fact", (17    ('HOTSPOT',    'start', lambda: webmgr.stop_service,18                            lambda: webmgr.web_service),        # noqa19    ('HOTSPOT',     'pass', lambda: webmgr.start_service,20                            lambda: webmgr.COMITUP_SERVICE),21    ('CONNECTING', 'start', lambda: webmgr.stop_service,22                            lambda: webmgr.COMITUP_SERVICE),23    ('CONNECTED',  'start', lambda: webmgr.start_service,24                            lambda: webmgr.web_service),25))26@pytest.mark.parametrize("svc", ("", "foo"))27@patch('comitup.webmgr.start_service')28@patch('comitup.webmgr.stop_service')29def test_webmgr_callback(stop_svc, start_svc, svc, state, action,30                         fn_fact, arg_fact, websvc_fxt):31    webmgr.web_service = svc32    webmgr.state_callback(state, action)33    if arg_fact():34        assert fn_fact().called35        assert fn_fact().called_with(call(arg_fact()))36    else:37        assert not fn_fact().called38    if svc:39        assert fn_fact().called40others = [(x, y) for x in ('HOTSPOT', 'CONNECTING', 'CONNECTED')41                 for y in ('fail', 'timeout')]                      # noqa42@pytest.mark.parametrize("state, action", [43        ('CONNECTING', 'pass'),44        ('CONNECTED', 'pass'),45    ] + others46)47@patch('comitup.webmgr.start_service')48@patch('comitup.webmgr.stop_service')49def test_webmgr_no_callback(stop_svc, start_svc, state, action, websvc_fxt):50    webmgr.web_service = 'foo'51    webmgr.state_callback(state, action)52    assert not stop_svc.called...

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.