python类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)])