Merge "Implemented automatic updates plugin"

This commit is contained in:
Jenkins 2017-03-17 12:58:59 +00:00 committed by Gerrit Code Review
commit 28c8cce1f9
6 changed files with 218 additions and 0 deletions

View File

@ -267,6 +267,10 @@ class GlobalOptions(conf_base.Options):
help='Copies the userdata to the given file path. The path '
'can include environment variables that will be expanded,'
' e.g. "%%SYSTEMDRIVE%%\\CloudbaseInit\\UserData.bin"'),
cfg.BoolOpt(
'enable_automatic_updates', default=None,
help='If set, enables or disables automatic operating '
'system updates.'),
]
self._cli_options = [

View File

@ -202,6 +202,10 @@ class BaseMetadataService(object):
def get_use_avma_licensing(self):
pass
def get_enable_automatic_updates(self):
"""Check if the metadata provider enforces automatic updates."""
pass
class BaseHTTPMetadataService(BaseMetadataService):

View File

@ -0,0 +1,38 @@
# Copyright (c) 2017 Cloudbase Solutions Srl
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from oslo_log import log as oslo_logging
from cloudbaseinit import conf as cloudbaseinit_conf
from cloudbaseinit.plugins.common import base
from cloudbaseinit.utils.windows import updates
CONF = cloudbaseinit_conf.CONF
LOG = oslo_logging.getLogger(__name__)
class WindowsAutoUpdatesPlugin(base.BasePlugin):
def execute(self, service, shared_data):
enable_updates = service.get_enable_automatic_updates()
if enable_updates is None:
enable_updates = CONF.enable_automatic_updates
if enable_updates is not None:
LOG.info("Configuring automatic updates: %s", enable_updates)
updates.set_automatic_updates(enable_updates)
return base.PLUGIN_EXECUTION_DONE, False
def get_os_requirements(self):
return 'win32', (5, 2)

View File

@ -0,0 +1,66 @@
# Copyright (c) 2017 Cloudbase Solutions Srl
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import importlib
import unittest
try:
import unittest.mock as mock
except ImportError:
import mock
from cloudbaseinit import conf as cloudbaseinit_conf
from cloudbaseinit.plugins.common import base
from cloudbaseinit.tests import testutils
CONF = cloudbaseinit_conf.CONF
MODPATH = "cloudbaseinit.plugins.windows.updates"
class WindowsAutoUpdatesPluginTest(unittest.TestCase):
def setUp(self):
self.mock_win32com = mock.MagicMock()
patcher = mock.patch.dict(
"sys.modules",
{
"win32com": self.mock_win32com
}
)
patcher.start()
self.addCleanup(patcher.stop)
updates = importlib.import_module(MODPATH)
self._updates_plugin = updates.WindowsAutoUpdatesPlugin()
self.snatcher = testutils.LogSnatcher(MODPATH)
@testutils.ConfPatcher("enable_automatic_updates", True)
@mock.patch("cloudbaseinit.utils.windows.updates.set_automatic_updates")
def test_execute(self, mock_set_updates):
mock_service = mock.Mock()
mock_shared_data = mock.Mock()
mock_service.get_enable_automatic_updates.return_value = True
expected_res = (base.PLUGIN_EXECUTION_DONE, False)
expected_logs = ["Configuring automatic updates: %s" % True]
with self.snatcher:
res = self._updates_plugin.execute(mock_service, mock_shared_data)
self.assertEqual(res, expected_res)
self.assertEqual(self.snatcher.output, expected_logs)
mock_service.get_enable_automatic_updates.assert_called_once_with()
mock_set_updates.assert_called_once_with(True)
def test_get_os_requirements(self):
expected_res = ('win32', (5, 2))
requirements_res = self._updates_plugin.get_os_requirements()
self.assertEqual(requirements_res, expected_res)

View File

@ -0,0 +1,63 @@
# Copyright (c) 2017 Cloudbase Solutions Srl
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import importlib
import unittest
try:
import unittest.mock as mock
except ImportError:
import mock
MODPATH = "cloudbaseinit.utils.windows.updates"
class UpdatesUtilTest(unittest.TestCase):
def setUp(self):
self._win32_com = mock.MagicMock()
self._module_patcher = mock.patch.dict(
'sys.modules', {
'win32com': self._win32_com})
self._win32_com_client = self._win32_com.client
self._module_patcher.start()
self._updates = importlib.import_module(MODPATH)
def tearDown(self):
self._module_patcher.stop()
@mock.patch("cloudbaseinit.osutils.factory.get_os_utils")
def _test_set_automatic_updates(self, mock_get_os_utils, enabled=True):
mock_osutils = mock.Mock()
mock_updates = mock.Mock()
mock_get_os_utils.return_value = mock_osutils
mock_osutils.check_os_version.return_value = False
self._win32_com_client.Dispatch.return_value = mock_updates
if not enabled:
self._updates.set_automatic_updates(enabled)
self.assertEqual(mock_updates.Settings.NotificationLevel, 1)
mock_updates.Settings.Save.assert_called_once_with()
else:
self._updates.set_automatic_updates(enabled)
mock_get_os_utils.assert_called_once_with()
self.assertIsNotNone(
mock_updates.SettingsScheduledInstallationTime)
mock_updates.Settings.Save.assert_called_once_with()
def test_set_automatic_no_updates(self):
self._test_set_automatic_updates(enabled=False)
def test_set_automatic_updates(self):
self._test_set_automatic_updates()

View File

@ -0,0 +1,43 @@
# Copyright (c) 2017 Cloudbase Solutions Srl
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import random
from win32com import client
from cloudbaseinit.osutils import factory as osutils_factory
AU_DISABLED = 1
AU_SCHEDULED_INSTALLATION = 4
MIN_INSTALL_HOUR = 1
MAX_INSTALL_HOUR = 5
def set_automatic_updates(enabled):
# TODO(alexpilotti): the following settings are ignored on
# Windows 10 / Windows Server 2016 build 14393
auto_update = client.Dispatch("Microsoft.Update.AutoUpdate")
if enabled:
auto_update.Settings.NotificationLevel = AU_SCHEDULED_INSTALLATION
osutils = osutils_factory.get_os_utils()
if not osutils.check_os_version(6, 2):
# NOTE(alexpilotti): this setting is not supported starting
# with Windows 8 / Windows Server 2012
hour = random.randint(MIN_INSTALL_HOUR, MAX_INSTALL_HOUR)
auto_update.SettingsScheduledInstallationTime = hour
else:
auto_update.Settings.NotificationLevel = AU_DISABLED
auto_update.Settings.Save()