<s> import contextlib <EOL> import functools <EOL> import logging as base_logging <EOL> import os <EOL> import shutil <EOL> import tempfile <EOL> import unittest <EOL> try : <EOL> import unittest . mock as mock <EOL> except ImportError : <EOL> import mock <EOL> from oslo_config import cfg <EOL> from oslo_log import log as oslo_logging <EOL> from cloudbaseinit import exception <EOL> CONF = cfg . CONF <EOL> @ contextlib . contextmanager <EOL> def create_tempdir ( ) : <EOL> """<STR_LIT>""" <EOL> tempdir = tempfile . mkdtemp ( prefix = "<STR_LIT>" ) <EOL> try : <EOL> yield tempdir <EOL> finally : <EOL> shutil . rmtree ( tempdir ) <EOL> @ contextlib . contextmanager <EOL> def create_tempfile ( content = None ) : <EOL> """<STR_LIT>""" <EOL> with create_tempdir ( ) as temp : <EOL> fd , path = tempfile . mkstemp ( dir = temp ) <EOL> os . close ( fd ) <EOL> if content : <EOL> with open ( path , '<STR_LIT:w>' ) as stream : <EOL> stream . write ( content ) <EOL> yield path <EOL> class SnatchHandler ( base_logging . Handler ) : <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( SnatchHandler , self ) . __init__ ( * args , ** kwargs ) <EOL> self . output = [ ] <EOL> def emit ( self , record ) : <EOL> msg = self . format ( record ) <EOL> self . output . append ( msg ) <EOL> class LogSnatcher ( object ) : <EOL> """<STR_LIT>""" <EOL> @ property <EOL> def output ( self ) : <EOL> return self . _snatch_handler . output <EOL> def __init__ ( self , logger_name ) : <EOL> self . _logger_name = logger_name <EOL> self . _snatch_handler = SnatchHandler ( ) <EOL> self . _logger = oslo_logging . getLogger ( self . _logger_name ) <EOL> self . _previous_level = self . _logger . logger . getEffectiveLevel ( ) <EOL> def __enter__ ( self ) : <EOL> self . _logger . logger . setLevel ( base_logging . DEBUG ) <EOL> self . _logger . handlers . append ( self . _snatch_handler ) <EOL> return self <EOL> def __exit__ ( self , * args ) : <EOL> self . _logger . handlers . remove ( self . _snatch_handler ) <EOL> self . _logger . logger . setLevel ( self . _previous_level ) <EOL> class ConfPatcher ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , key , value , conf = CONF ) : <EOL> self . _original_value = conf . get ( key ) <EOL> self . _key = key <EOL> self . _value = value <EOL> self . _conf = conf <EOL> def __call__ ( self , func , * args , ** kwargs ) : <EOL> def _wrapped_f ( * args , ** kwargs ) : <EOL> with self : <EOL> return func ( * args , ** kwargs ) <EOL> functools . update_wrapper ( _wrapped_f , func ) <EOL> return _wrapped_f <EOL> def __enter__ ( self ) : <EOL> self . _conf . set_override ( self . _key , self . _value ) <EOL> return self <EOL> def __exit__ ( self , exc_type , exc_val , exc_tb ) : <EOL> self . _conf . set_override ( self . _key , self . _original_value ) <EOL> class CloudbaseInitTestBase ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> @ contextlib . contextmanager <EOL> def assert_raises_windows_message ( <EOL> self , expected_msg , error_code , <EOL> exc = exception . WindowsCloudbaseInitException ) : <EOL> """<STR_LIT>""" <EOL> with self . assertRaises ( exc ) as cm : <EOL> with mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> create = True ) as mock_format_error : <EOL> with mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> create = True ) as mock_get_last_error : <EOL> mock_format_error . return_value = "<STR_LIT:description>" <EOL> yield <EOL> if mock_get_last_error . called : <EOL> mock_get_last_error . assert_called_once_with ( ) <EOL> mock_format_error . assert_called_once_with ( <EOL> mock_get_last_error . return_value ) <EOL> else : <EOL> mock_format_error . assert_called_once_with ( error_code ) <EOL> expected_msg = expected_msg % mock_format_error . return_value <EOL> self . assertEqual ( expected_msg , cm . exception . args [ <NUM_LIT:0> ] ) </s>
