我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用charmhelpers.core.host.service_running()。
def test_pauses_a_running_upstart_service(self, service, systemd, service_running): """Pause on a running service will call service stop.""" service_name = 'foo-service' service.side_effect = [True] systemd.return_value = False service_running.return_value = True tempdir = mkdtemp(prefix="test_pauses_an_upstart_service") conf_path = os.path.join(tempdir, "{}.conf".format(service_name)) # Just needs to exist with open(conf_path, "w") as fh: fh.write("") self.addCleanup(rmtree, tempdir) self.assertTrue(host.service_pause(service_name, init_dir=tempdir)) service.assert_called_with('stop', service_name) override_path = os.path.join( tempdir, "{}.override".format(service_name)) with open(override_path, "r") as fh: override_contents = fh.read() self.assertEqual("manual\n", override_contents)
def test_pauses_a_stopped_upstart_service(self, service, systemd, service_running): """Pause on a stopped service will not call service stop.""" service_name = 'foo-service' service.side_effect = [True] systemd.return_value = False service_running.return_value = False tempdir = mkdtemp(prefix="test_pauses_an_upstart_service") conf_path = os.path.join(tempdir, "{}.conf".format(service_name)) # Just needs to exist with open(conf_path, "w") as fh: fh.write("") self.addCleanup(rmtree, tempdir) self.assertTrue(host.service_pause(service_name, init_dir=tempdir)) # Stop isn't called because service is already stopped self.assertRaises( AssertionError, service.assert_called_with, 'stop', service_name) override_path = os.path.join( tempdir, "{}.override".format(service_name)) with open(override_path, "r") as fh: override_contents = fh.read() self.assertEqual("manual\n", override_contents)
def test_pauses_a_running_sysv_service(self, service, check_call, systemd, service_running): """Pause calls service stop on a running sysv service.""" service_name = 'foo-service' service.side_effect = [True] systemd.return_value = False service_running.return_value = True tempdir = mkdtemp(prefix="test_pauses_a_sysv_service") sysv_path = os.path.join(tempdir, service_name) # Just needs to exist with open(sysv_path, "w") as fh: fh.write("") self.addCleanup(rmtree, tempdir) self.assertTrue(host.service_pause( service_name, init_dir=tempdir, initd_dir=tempdir)) service.assert_called_with('stop', service_name) check_call.assert_called_with(["update-rc.d", service_name, "disable"])
def test_resumes_a_running_upstart_service(self, service, check_output, systemd, service_running): """When the service is already running, service start isn't called.""" service_name = 'foo-service' service.side_effect = [True] systemd.return_value = False service_running.return_value = True tempdir = mkdtemp(prefix="test_resumes_an_upstart_service") conf_path = os.path.join(tempdir, "{}.conf".format(service_name)) with open(conf_path, "w") as fh: fh.write("") self.addCleanup(rmtree, tempdir) self.assertTrue(host.service_resume(service_name, init_dir=tempdir)) # Start isn't called because service is already running self.assertFalse(service.called) override_path = os.path.join( tempdir, "{}.override".format(service_name)) self.assertFalse(os.path.exists(override_path))
def test_resumes_a_stopped_upstart_service(self, service, check_output, systemd, service_running): """When the service is stopped, service start is called.""" check_output.return_value = b'foo-service stop/waiting' service_name = 'foo-service' service.side_effect = [True] systemd.return_value = False service_running.return_value = False tempdir = mkdtemp(prefix="test_resumes_an_upstart_service") conf_path = os.path.join(tempdir, "{}.conf".format(service_name)) with open(conf_path, "w") as fh: fh.write("") self.addCleanup(rmtree, tempdir) self.assertTrue(host.service_resume(service_name, init_dir=tempdir)) service.assert_called_with('start', service_name) override_path = os.path.join( tempdir, "{}.override".format(service_name)) self.assertFalse(os.path.exists(override_path))
def test_resumes_a_sysv_service(self, service, check_call, systemd, service_running): """When process is in a stop/waiting state, service start is called.""" service_name = 'foo-service' service.side_effect = [True] systemd.return_value = False service_running.return_value = False tempdir = mkdtemp(prefix="test_resumes_a_sysv_service") sysv_path = os.path.join(tempdir, service_name) # Just needs to exist with open(sysv_path, "w") as fh: fh.write("") self.addCleanup(rmtree, tempdir) self.assertTrue(host.service_resume( service_name, init_dir=tempdir, initd_dir=tempdir)) service.assert_called_with('start', service_name) check_call.assert_called_with(["update-rc.d", service_name, "enable"])
def test_resume_a_running_sysv_service(self, service, check_call, systemd, service_running): """When process is already running, service start isn't called.""" service_name = 'foo-service' systemd.return_value = False service_running.return_value = True tempdir = mkdtemp(prefix="test_resumes_a_sysv_service") sysv_path = os.path.join(tempdir, service_name) # Just needs to exist with open(sysv_path, "w") as fh: fh.write("") self.addCleanup(rmtree, tempdir) self.assertTrue(host.service_resume( service_name, init_dir=tempdir, initd_dir=tempdir)) # Start isn't called because service is already running self.assertFalse(service.called) check_call.assert_called_with(["update-rc.d", service_name, "enable"])
def update_status(): if service_running('plumgrid'): status_set('active', 'Unit is ready') else: status_set('blocked', 'plumgrid service not running')
def service_stop(service_name): """ Wrapper around host.service_stop to prevent spurious "unknown service" messages in the logs. """ if host.service_running(service_name): host.service_stop(service_name)
def service_restart(service_name): """ Wrapper around host.service_restart to prevent spurious "unknown service" messages in the logs. """ if host.service_available(service_name): if host.service_running(service_name): host.service_restart(service_name) else: host.service_start(service_name) # Convenience aliases
