Python charmhelpers.core.host 模块,service_pause() 实例源码
我们从Python开源项目中,提取了以下10个代码示例,用于说明如何使用charmhelpers.core.host.service_pause()。
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_pauses_a_stopped_sysv_service(self, service, check_call,
systemd, service_running):
"""Pause does not call service stop on a stopped sysv service."""
service_name = 'foo-service'
service.side_effect = [True]
systemd.return_value = False
service_running.return_value = False
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))
# Stop isn't called because service is already stopped
self.assertRaises(
AssertionError, service.assert_called_with, 'stop', service_name)
check_call.assert_called_with(["update-rc.d", service_name, "disable"])
def pause(args):
"""Pause all the swift services.
@raises Exception if any services fail to stop
"""
for service in args.services:
stopped = service_pause(service)
if not stopped:
raise Exception("{} didn't stop cleanly.".format(service))
set_unit_paused()
assess_status(CONFIGS, args.services)
def disable_services(self):
'''Disable all services related to gnocchi'''
for svc in self.services:
host.service_pause(svc)
def install(self):
super(GnocchiCharm, self).install()
# NOTE(jamespage): always pause gnocchi-api service as we force
# execution with Apache2+mod_wsgi
host.service_pause('gnocchi-api')
def pause(args):
"""Pause all the swift services.
@raises Exception if any services fail to stop
"""
for service in args.services:
stopped = service_pause(service)
if not stopped:
raise Exception("{} didn't stop cleanly.".format(service))
with HookData()():
kv().set('unit-paused', True)
set_os_workload_status(CONFIGS, REQUIRED_INTERFACES,
charm_func=assess_status)
def start_ntpmon():
"""
Start the ntpmon daemon process.
If ntp is not installed, do nothing.
"""
if os.path.exists(ntp_conf):
hookenv.log(ntp_conf + ' present; enabling and starting ntpmon')
host.service_resume(service_name)
else:
hookenv.log(ntp_conf + ' not present; disabling ntpmon')
host.service_pause(service_name)
set_state('ntpmon.started')
def test_pauses_a_running_systemd_unit(self, service, systemd,
service_running):
"""Pause on a running systemd unit will be stopped and disabled."""
service_name = 'foo-service'
service_running.return_value = True
systemd.return_value = True
self.assertTrue(host.service_pause(service_name))
service.assert_has_calls([
call('stop', service_name),
call('disable', service_name),
call('mask', service_name)])