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

How to use _check_service_running method in lisa

Best Python code snippet using lisa_python

test_vms_exist.py

Source:test_vms_exist.py

...26        stdout, stderr = vm.run("uname -r")27        kernel_version = stdout.decode("utf-8").rstrip()28        assert kernel_version.endswith("-grsec-workstation")29        assert kernel_version == EXPECTED_KERNEL_VERSION30    def _check_service_running(self, vm, service):31        """32        Ensures a given service is running inside a given VM.33        Uses systemctl is-active to query the service state.34        """35        cmd = "systemctl is-active {}".format(service)36        stdout, stderr = vm.run(cmd)37        service_status = stdout.decode("utf-8").rstrip()38        assert service_status == "active"39    def test_sd_whonix_config(self):40        vm = self.app.domains["sd-whonix"]41        nvm = vm.netvm42        self.assertTrue(nvm.name == "sys-firewall")43        wanted_kernelopts = "nopat apparmor=1 security=apparmor"44        self.assertEqual(vm.kernelopts, wanted_kernelopts)45        self.assertTrue(vm.template == "whonix-gw-16")46        self.assertTrue(vm.provides_network)47        self.assertTrue(vm.autostart is True)48        self.assertFalse(vm.template_for_dispvms)49        self.assertTrue("sd-workstation" in vm.tags)50    def test_sd_proxy_config(self):51        vm = self.app.domains["sd-proxy"]52        nvm = vm.netvm53        self.assertTrue(nvm.name == "sd-whonix")54        self.assertTrue(vm.template == "sd-small-{}-template".format(DEBIAN_VERSION))55        self.assertTrue(vm.autostart is True)56        self.assertFalse(vm.provides_network)57        self.assertFalse(vm.template_for_dispvms)58        self.assertTrue("sd-workstation" in vm.tags)59    def test_sd_app_config(self):60        vm = self.app.domains["sd-app"]61        nvm = vm.netvm62        self.assertTrue(nvm is None)63        self.assertTrue(vm.template == "sd-small-{}-template".format(DEBIAN_VERSION))64        self.assertFalse(vm.provides_network)65        self.assertFalse(vm.template_for_dispvms)66        self._check_kernel(vm)67        self._check_service_running(vm, "paxctld")68        self.assertTrue("sd-workstation" in vm.tags)69        self.assertTrue("sd-client" in vm.tags)70        # Check the size of the private volume71        # Should be 10GB72        # >>> 1024 * 1024 * 10 * 102473        size = self.config["vmsizes"]["sd_app"]74        vol = vm.volumes["private"]75        self.assertEqual(vol.size, size * 1024 * 1024 * 1024)76    def test_sd_viewer_config(self):77        vm = self.app.domains["sd-viewer"]78        nvm = vm.netvm79        self.assertTrue(nvm is None)80        self.assertTrue(vm.template == "sd-large-{}-template".format(DEBIAN_VERSION))81        self.assertFalse(vm.provides_network)82        self.assertTrue(vm.template_for_dispvms)83        # sd-viewer should not be able to create other disposable VMs84        self.assertIsNone(vm.default_dispvm)85        self._check_kernel(vm)86        self._check_service_running(vm, "paxctld")87        self.assertTrue("sd-workstation" in vm.tags)88    def test_sd_gpg_config(self):89        vm = self.app.domains["sd-gpg"]90        nvm = vm.netvm91        self.assertTrue(nvm is None)92        # No sd-gpg-template, since keyring is managed in $HOME93        self.assertTrue(vm.template == "sd-small-{}-template".format(DEBIAN_VERSION))94        self.assertTrue(vm.autostart is True)95        self.assertFalse(vm.provides_network)96        self.assertFalse(vm.template_for_dispvms)97        self._check_kernel(vm)98        self.assertTrue("sd-workstation" in vm.tags)99    def test_sd_log_config(self):100        vm = self.app.domains["sd-log"]101        nvm = vm.netvm102        self.assertTrue(nvm is None)103        self.assertTrue(vm.template == "sd-small-{}-template".format(DEBIAN_VERSION))104        self.assertTrue(vm.autostart is True)105        self.assertFalse(vm.provides_network)106        self.assertFalse(vm.template_for_dispvms)107        self._check_kernel(vm)108        self._check_service_running(vm, "paxctld")109        self._check_service_running(vm, "securedrop-log")110        self.assertFalse(vm.template_for_dispvms)111        self.assertTrue("sd-workstation" in vm.tags)112        # Check the size of the private volume113        # Should be same of config.json114        # >>> 1024 * 1024 * 5 * 1024115        size = self.config["vmsizes"]["sd_log"]116        vol = vm.volumes["private"]117        self.assertEqual(vol.size, size * 1024 * 1024 * 1024)118    def test_sd_workstation_template(self):119        vm = self.app.domains["securedrop-workstation-{}".format(DEBIAN_VERSION)]120        nvm = vm.netvm121        self.assertTrue(nvm is None)122        self.assertTrue(vm.virt_mode == "hvm")123        self.assertTrue(vm.kernel == "")124        self.assertTrue("sd-workstation" in vm.tags)125        self._check_kernel(vm)126        self._check_service_running(vm, "paxctld")127    def test_sd_proxy_template(self):128        vm = self.app.domains["sd-small-{}-template".format(DEBIAN_VERSION)]129        nvm = vm.netvm130        self.assertTrue(nvm is None)131        self.assertTrue("sd-workstation" in vm.tags)132    def sd_app_template(self):133        vm = self.app.domains["sd-small-{}-template".format(DEBIAN_VERSION)]134        nvm = vm.netvm135        self.assertTrue(nvm is None)136        self.assertTrue("sd-workstation" in vm.tags)137        self._check_kernel(vm)138    def sd_viewer_template(self):139        vm = self.app.domains["sd-large-{}-template".format(DEBIAN_VERSION)]140        nvm = vm.netvm...

service.py

Source:service.py

...29        self._internal_tool.stop_service(name)  # type: ignore30    def enable_service(self, name: str) -> None:31        self._internal_tool.enable_service(name)  # type: ignore32    def check_service_status(self, name: str) -> bool:33        return self._internal_tool._check_service_running(name)  # type: ignore34    def check_service_exists(self, name: str) -> bool:35        return self._internal_tool._check_service_exists(name)  # type: ignore36class ServiceInternal(Tool):37    @property38    def command(self) -> str:39        return "service"40    @property41    def can_install(self) -> bool:42        return False43    def _check_exists(self) -> bool:44        return True45    def _check_service_exists(self, name: str) -> bool:46        cmd_result = self.run(f"{name} status", shell=True, sudo=True, force_run=True)47        if "unrecognized service" in cmd_result.stdout:48            return False49        return True50    def _check_service_running(self, name: str) -> bool:51        cmd_result = self.run(f"{name} status", shell=True, sudo=True, force_run=True)52        return (53            "unrecognized service" not in cmd_result.stdout54            and 0 == cmd_result.exit_code55        )56    def stop_service(self, name: str) -> None:57        if self._check_service_running(name):58            cmd_result = self.run(f"{name} stop", shell=True, sudo=True, force_run=True)59            cmd_result.assert_exit_code()60    def restart_service(self, name: str, ignore_exit_code: int = 0) -> None:61        cmd_result = self.run(f"{name} restart", shell=True, sudo=True, force_run=True)62        # optionally ignore exit code if it matches our expected non-zero value63        _check_error_codes(cmd_result, ignore_exit_code)64class Systemctl(Tool):65    @property66    def command(self) -> str:67        return "systemctl"68    @property69    def can_install(self) -> bool:70        return False71    def stop_service(self, name: str) -> None:72        if self._check_service_running(name):73            cmd_result = self.run(f"stop {name}", shell=True, sudo=True, force_run=True)74            cmd_result.assert_exit_code()75    def restart_service(self, name: str, ignore_exit_code: int = 0) -> None:76        cmd_result = self.run(f"restart {name}", shell=True, sudo=True, force_run=True)77        _check_error_codes(cmd_result, ignore_exit_code)78    def enable_service(self, name: str) -> None:79        cmd_result = self.run(f"enable {name}", shell=True, sudo=True, force_run=True)80        cmd_result.assert_exit_code()81    def hibernate(self) -> None:82        self.run_async("hibernate", sudo=True, force_run=True)83    def _check_exists(self) -> bool:84        return True85    def _check_service_exists(self, name: str) -> bool:86        cmd_result = self.run(87            f"--full --no-pager status {name}", shell=True, sudo=True, force_run=True88        )89        if (90            "could not be found" in cmd_result.stdout91            or "not-found" in cmd_result.stdout92        ):93            return False94        return True95    def _check_service_running(self, name: str) -> bool:96        cmd_result = self.run(97            f"--full --no-pager status {name}", shell=True, sudo=True, force_run=True98        )99        return (100            "could not be found" not in cmd_result.stdout101            or "not-found" in cmd_result.stdout102        ) and 0 == cmd_result.exit_code103def _check_error_codes(cmd_result: ExecutableResult, error_code: int = 0) -> None:...

debug_cloud_controller.py

Source:debug_cloud_controller.py

...5    def debug(self):6        clcs = self.component_deployer.roles['clc'] 7        with hide('everything'):8            for clc in clcs:9                self._check_service_running(clc)10                self._services_enabled(clc)11                self._psql_available(clc)12                self._db_size_check(clc)13                self._var_lib_euca_size_check(clc)14                self._memory_usage(clc)15        return (self.passed, self.failed)16    def _check_service_running(self, clc):17        clc_service_state = self.run_command_on_host(18            'service eucalyptus-cloud status', host=clc)19        if re.search('running', clc_service_state):20            self.success(clc + ': CLC service running')21        else:22            self.failure(clc + ': CLC service not running')23    def _services_enabled(self, clc):24        describe_services = self.run_command_on_host('euca-describe-services',25                                                     host=clc)26        for state in ['DISABLED', 'BROKEN', 'NOTREADY']:27            search = re.search('.*' + state + '.*', describe_services)28            if search:29                self.failure(clc + ': Some services are in ' + state)30                print search.group()...

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.