b957aadc47
The patch includes the following changes:
* add a new folder for cloud-config plugins, in userdataplugins.
It will contain plugins which can be processed by the cloud-config
user data plugin, right now, only write-files is supported.
* move the code related to cloudconfig functionalities in a
executor class. The meaning of this class is plain and simple
and it doesn't clutter the base plugin.
* add a from_yaml classmethod to the newly added cloud config executor.
In this way, we could add support for other formats as well,
the only thing required being adding a new factory method for
these formats.
* the tests aren't relying too much on mock, they are written
with feature-testing in mind.
Change-Id: I8f73e7aaafee74e8023bf7c5e13579982b5f9498
111 lines
3.2 KiB
Python
111 lines
3.2 KiB
Python
# Copyright 2014 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 contextlib
|
|
import logging as base_logging
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
|
|
from cloudbaseinit.openstack.common import log as logging
|
|
|
|
|
|
__all__ = (
|
|
'create_tempfile',
|
|
'create_tempdir',
|
|
'LogSnatcher',
|
|
)
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def create_tempdir():
|
|
"""Create a temporary directory.
|
|
|
|
This is a context manager, which creates a new temporary
|
|
directory and removes it when exiting from the context manager
|
|
block.
|
|
"""
|
|
tempdir = tempfile.mkdtemp(prefix="cloudbaseinit-tests")
|
|
try:
|
|
yield tempdir
|
|
finally:
|
|
shutil.rmtree(tempdir)
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def create_tempfile(content=None):
|
|
"""Create a temporary file.
|
|
|
|
This is a context manager, which uses `create_tempdir` to obtain a
|
|
temporary directory, where the file will be placed.
|
|
|
|
:param content:
|
|
Additionally, a string which will be written
|
|
in the new file.
|
|
"""
|
|
with create_tempdir() as temp:
|
|
fd, path = tempfile.mkstemp(dir=temp)
|
|
os.close(fd)
|
|
if content:
|
|
with open(path, 'w') as stream:
|
|
stream.write(content)
|
|
yield path
|
|
|
|
|
|
# This is similar with unittest.TestCase.assertLogs from Python 3.4.
|
|
class SnatchHandler(base_logging.Handler):
|
|
def __init__(self, *args, **kwargs):
|
|
super(SnatchHandler, self).__init__(*args, **kwargs)
|
|
self.output = []
|
|
|
|
def emit(self, record):
|
|
msg = self.format(record)
|
|
self.output.append(msg)
|
|
|
|
|
|
class LogSnatcher(object):
|
|
"""A context manager to capture emitted logged messages.
|
|
|
|
The class can be used as following::
|
|
|
|
with LogSnatcher('plugins.windows.createuser') as snatcher:
|
|
LOG.info("doing stuff")
|
|
LOG.info("doing stuff %s", 1)
|
|
LOG.warn("doing other stuff")
|
|
...
|
|
self.assertEqual(snatcher.output,
|
|
['INFO:unknown:doing stuff',
|
|
'INFO:unknown:doing stuff 1',
|
|
'WARN:unknown:doing other stuff'])
|
|
"""
|
|
|
|
@property
|
|
def output(self):
|
|
return self._snatch_handler.output
|
|
|
|
def __init__(self, logger_name):
|
|
self._logger_name = logger_name
|
|
self._snatch_handler = SnatchHandler()
|
|
self._logger = logging.getLogger(self._logger_name)
|
|
self._previous_level = self._logger.logger.getEffectiveLevel()
|
|
|
|
def __enter__(self):
|
|
self._logger.logger.setLevel(base_logging.DEBUG)
|
|
self._logger.handlers.append(self._snatch_handler)
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
self._logger.handlers.remove(self._snatch_handler)
|
|
self._logger.logger.setLevel(self._previous_level)
|