python类service_resume()的实例源码

def resume(args):
    """Resume all the swift services.

    @raises Exception if any services fail to start
    """
    for service in args.services:
        started = service_resume(service)
        if not started:
            raise Exception("{} didn't start cleanly.".format(service))
    with HookData()():
        kv().set('unit-paused', False)
    set_os_workload_status(CONFIGS, REQUIRED_INTERFACES,
                           charm_func=assess_status)


# A dictionary of all the defined actions to callables (which take
# parsed arguments).

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 resume(args):
    """Resume all the swift services.

    @raises Exception if any services fail to start
    """
    for service in args.services:
        started = service_resume(service)
        if not started:
            raise Exception("{} didn't start cleanly.".format(service))
    clear_unit_paused()
    assess_status(CONFIGS, args.services)

def enable_services(self):
        '''Enable all services related to gnocchi'''
        for svc in self.services:
            host.service_resume(svc)

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_resumes_a_stopped_systemd_unit(self, service, systemd,
                                            service_running):
        """Resume on a stopped systemd unit will be started and enabled."""
        service_name = 'foo-service'
        service_running.return_value = False
        systemd.return_value = True
        self.assertTrue(host.service_resume(service_name))
        service.assert_has_calls([
            call('unmask', service_name),
            # Ensures a package starts up if disabled but not masked,
            # per lp:1692178
            call('enable', service_name),
            call('start', service_name)])