<s> import os <EOL> import sys <EOL> import copy <EOL> import time <EOL> import socket <EOL> import lxml . etree <EOL> import Bcfg2 . Server <EOL> import Bcfg2 . Server . Plugin <EOL> from mock import Mock , MagicMock , patch <EOL> path = os . path . dirname ( __file__ ) <EOL> while path != "<STR_LIT:/>" : <EOL> if os . path . basename ( path ) . lower ( ) . startswith ( "<STR_LIT:test>" ) : <EOL> sys . path . append ( path ) <EOL> if os . path . basename ( path ) == "<STR_LIT>" : <EOL> break <EOL> path = os . path . dirname ( path ) <EOL> from common import * <EOL> from Bcfg2 . Server . Plugins . Metadata import load_django_models <EOL> from TestPlugin import TestXMLFileBacked , TestMetadata as _TestMetadata , TestClientRunHooks , TestDatabaseBacked <EOL> load_django_models ( ) <EOL> from Bcfg2 . Server . Plugins . Metadata import * <EOL> def get_clients_test_tree ( ) : <EOL> return lxml . etree . XML ( '''<STR_LIT>''' ) . getroottree ( ) <EOL> def get_groups_test_tree ( ) : <EOL> return lxml . etree . XML ( '''<STR_LIT>''' ) . getroottree ( ) <EOL> def get_metadata_object ( core = None ) : <EOL> if core is None : <EOL> core = Mock ( ) <EOL> core . metadata_cache = MagicMock ( ) <EOL> set_setup_default ( "<STR_LIT:password>" ) <EOL> @ patchIf ( not isinstance ( os . makedirs , Mock ) , "<STR_LIT>" , Mock ( ) ) <EOL> @ patchIf ( not isinstance ( lxml . etree . Element , Mock ) , <EOL> "<STR_LIT>" , Mock ( ) ) <EOL> def inner ( ) : <EOL> return Metadata ( core ) <EOL> return inner ( ) <EOL> class TestMetadataDB ( DBModelTestCase ) : <EOL> if HAS_DJANGO : <EOL> models = [ MetadataClientModel ] <EOL> class TestClientVersions ( TestDatabaseBacked ) : <EOL> test_clients = dict ( client1 = "<STR_LIT>" , <EOL> client2 = "<STR_LIT>" , <EOL> client3 = "<STR_LIT>" , <EOL> client4 = "<STR_LIT>" , <EOL> client5 = None , <EOL> client6 = None ) <EOL> @ skipUnless ( HAS_DJANGO , "<STR_LIT>" ) <EOL> def setUp ( self ) : <EOL> TestDatabaseBacked . setUp ( self ) <EOL> self . test_obj = ClientVersions <EOL> syncdb ( TestMetadataDB ) <EOL> for client , version in self . test_clients . items ( ) : <EOL> MetadataClientModel ( hostname = client , version = version ) . save ( ) <EOL> def test__contains ( self ) : <EOL> v = self . get_obj ( ) <EOL> self . assertIn ( "<STR_LIT>" , v ) <EOL> self . assertIn ( "<STR_LIT>" , v ) <EOL> self . assertNotIn ( "<STR_LIT>" , v ) <EOL> def test_keys ( self ) : <EOL> v = self . get_obj ( ) <EOL> self . assertItemsEqual ( self . test_clients . keys ( ) , v . keys ( ) ) <EOL> def test__setitem ( self ) : <EOL> v = self . get_obj ( ) <EOL> v [ "<STR_LIT>" ] = "<STR_LIT>" <EOL> self . assertIn ( "<STR_LIT>" , v ) <EOL> self . assertEqual ( v [ '<STR_LIT>' ] , "<STR_LIT>" ) <EOL> client = MetadataClientModel . objects . get ( hostname = "<STR_LIT>" ) <EOL> self . assertEqual ( client . version , "<STR_LIT>" ) <EOL> new = "<STR_LIT>" <EOL> v [ new ] = "<STR_LIT>" <EOL> self . assertIn ( new , v ) <EOL> self . assertEqual ( v [ new ] , "<STR_LIT>" ) <EOL> client = MetadataClientModel . objects . get ( hostname = new ) <EOL> self . assertEqual ( client . version , "<STR_LIT>" ) <EOL> new2 = "<STR_LIT>" <EOL> v [ new2 ] = None <EOL> self . assertIn ( new2 , v ) <EOL> self . assertEqual ( v [ new2 ] , None ) <EOL> client = MetadataClientModel . objects . get ( hostname = new2 ) <EOL> self . assertEqual ( client . version , None ) <EOL> def test__getitem ( self ) : <EOL> v = self . get_obj ( ) <EOL> self . assertEqual ( v [ '<STR_LIT>' ] , "<STR_LIT>" ) <EOL> self . assertIsNone ( v [ '<STR_LIT>' ] ) <EOL> expected = KeyError <EOL> try : <EOL> v [ '<STR_LIT>' ] <EOL> except expected : <EOL> pass <EOL> except : <EOL> err = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> self . assertFalse ( True , "<STR_LIT>" % <EOL> ( err . __class__ . __name__ , <EOL> expected . __class__ . __name__ ) ) <EOL> else : <EOL> self . assertFalse ( True , <EOL> "<STR_LIT>" % expected . __class__ . __name__ ) <EOL> def test__len ( self ) : <EOL> v = self . get_obj ( ) <EOL> self . assertEqual ( len ( v ) , MetadataClientModel . objects . count ( ) ) <EOL> def test__iter ( self ) : <EOL> v = self . get_obj ( ) <EOL> self . assertItemsEqual ( [ h for h in iter ( v ) ] , v . keys ( ) ) <EOL> def test__delitem ( self ) : <EOL> v = self . get_obj ( ) <EOL> new = "<STR_LIT>" <EOL> v [ new ] = "<STR_LIT>" <EOL> del v [ new ] <EOL> self . assertIn ( new , v ) <EOL> self . assertIsNone ( v [ new ] ) <EOL> class TestXMLMetadataConfig ( TestXMLFileBacked ) : <EOL> test_obj = XMLMetadataConfig <EOL> path = os . path . join ( datastore , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def get_obj ( self , basefile = "<STR_LIT>" , core = None ) : <EOL> self . metadata = get_metadata_object ( core = core ) <EOL> @ patchIf ( not isinstance ( lxml . etree . Element , Mock ) , <EOL> "<STR_LIT>" , Mock ( ) ) <EOL> def inner ( ) : <EOL> return XMLMetadataConfig ( self . metadata , basefile ) <EOL> return inner ( ) <EOL> @ patch ( "<STR_LIT>" , Mock ( ) ) <EOL> def test__init ( self ) : <EOL> xmc = self . get_obj ( ) <EOL> self . assertNotIn ( call ( xmc . basefile ) , <EOL> xmc . fam . AddMonitor . call_args_list ) <EOL> def test_xdata ( self ) : <EOL> config = self . get_obj ( ) <EOL> expected = Bcfg2 . Server . Plugin . MetadataRuntimeError <EOL> try : <EOL> config . xdata <EOL> except expected : <EOL> pass <EOL> except : <EOL> err = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> self . assertFalse ( True , "<STR_LIT>" % <EOL> ( err . __class__ . __name__ , <EOL> expected . __class__ . __name__ ) ) <EOL> else : <EOL> self . assertFalse ( True , <EOL> "<STR_LIT>" % expected . __class__ . __name__ ) <EOL> pass <EOL> config . data = "<STR_LIT>" <EOL> self . assertEqual ( config . xdata , "<STR_LIT>" ) <EOL> def test_base_xdata ( self ) : <EOL> config = self . get_obj ( ) <EOL> expected = Bcfg2 . Server . Plugin . MetadataRuntimeError <EOL> try : <EOL> config . base_xdata <EOL> except expected : <EOL> pass <EOL> except : <EOL> err = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> self . assertFalse ( True , "<STR_LIT>" % <EOL> ( err . __class__ . __name__ , <EOL> expected . __class__ . __name__ ) ) <EOL> else : <EOL> self . assertFalse ( True , <EOL> "<STR_LIT>" % expected . __class__ . __name__ ) <EOL> pass <EOL> config . basedata = "<STR_LIT>" <EOL> self . assertEqual ( config . base_xdata , "<STR_LIT>" ) <EOL> def test_add_monitor ( self ) : <EOL> config = self . get_obj ( ) <EOL> config . fam = Mock ( ) <EOL> fname = "<STR_LIT>" <EOL> fpath = os . path . join ( self . metadata . data , fname ) <EOL> config . extras = [ ] <EOL> config . add_monitor ( fpath ) <EOL> config . fam . AddMonitor . assert_called_with ( fpath , config . metadata ) <EOL> self . assertItemsEqual ( config . extras , [ fpath ] ) <EOL> def test_Index ( self ) : <EOL> pass <EOL> @ patch ( "<STR_LIT>" ) <EOL> @ patch ( "<STR_LIT>" ) <EOL> def test_load_xml ( self , mock_follow , mock_parse ) : <EOL> config = self . get_obj ( "<STR_LIT>" ) <EOL> def reset ( ) : <EOL> mock_parse . reset_mock ( ) <EOL> mock_follow . reset_mock ( ) <EOL> config . data = None <EOL> config . basedata = None <EOL> reset ( ) <EOL> config . load_xml ( ) <EOL> mock_follow . assert_called_with ( xdata = mock_parse . return_value ) <EOL> mock_parse . assert_called_with ( os . path . join ( config . basedir , <EOL> "<STR_LIT>" ) , <EOL> parser = Bcfg2 . Server . XMLParser ) <EOL> self . assertFalse ( mock_parse . return_value . xinclude . called ) <EOL> self . assertEqual ( config . data , mock_parse . return_value ) <EOL> self . assertIsNotNone ( config . basedata ) <EOL> reset ( ) <EOL> mock_parse . side_effect = lxml . etree . XMLSyntaxError ( None , None , None , <EOL> None ) <EOL> config . load_xml ( ) <EOL> mock_parse . assert_called_with ( os . path . join ( config . basedir , <EOL> "<STR_LIT>" ) , <EOL> parser = Bcfg2 . Server . XMLParser ) <EOL> self . assertIsNone ( config . data ) <EOL> self . assertIsNone ( config . basedata ) <EOL> reset ( ) <EOL> mock_parse . side_effect = None <EOL> def follow_xincludes ( xdata = None ) : <EOL> config . extras = [ Mock ( ) , Mock ( ) ] <EOL> mock_follow . side_effect = follow_xincludes <EOL> config . load_xml ( ) <EOL> mock_follow . assert_called_with ( xdata = mock_parse . return_value ) <EOL> mock_parse . assert_called_with ( os . path . join ( config . basedir , <EOL> "<STR_LIT>" ) , <EOL> parser = Bcfg2 . Server . XMLParser ) <EOL> mock_parse . return_value . xinclude . assert_any_call ( ) <EOL> self . assertEqual ( config . data , mock_parse . return_value ) <EOL> self . assertIsNotNone ( config . basedata ) <EOL> @ patch ( "<STR_LIT>" ) <EOL> def test_write ( self , mock_write_xml ) : <EOL> config = self . get_obj ( "<STR_LIT>" ) <EOL> config . basedata = "<STR_LIT>" <EOL> config . write ( ) <EOL> mock_write_xml . assert_called_with ( os . path . join ( self . metadata . data , <EOL> "<STR_LIT>" ) , <EOL> "<STR_LIT>" ) <EOL> @ patch ( '<STR_LIT>' , Mock ( return_value = False ) ) <EOL> @ patch ( '<STR_LIT>' , Mock ( ) ) <EOL> @ patch ( "<STR_LIT>" ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_write_xml ( self , mock_readlink , mock_islink , mock_rename , <EOL> mock_unlink , mock_fdopen , mock_open , mock_load_xml ) : <EOL> fname = "<STR_LIT>" <EOL> config = self . get_obj ( fname ) <EOL> fpath = os . path . join ( self . metadata . data , fname ) <EOL> tmpfile = "<STR_LIT>" % fpath <EOL> linkdest = os . path . join ( self . metadata . data , "<STR_LIT>" ) <EOL> def reset ( ) : <EOL> mock_readlink . reset_mock ( ) <EOL> mock_islink . reset_mock ( ) <EOL> mock_rename . reset_mock ( ) <EOL> mock_unlink . reset_mock ( ) <EOL> mock_fdopen . reset_mock ( ) <EOL> mock_open . reset_mock ( ) <EOL> mock_load_xml . reset_mock ( ) <EOL> mock_islink . return_value = False <EOL> config . write_xml ( fpath , get_clients_test_tree ( ) ) <EOL> mock_open . assert_called_with ( tmpfile , <EOL> os . O_CREAT | os . O_EXCL | os . O_WRONLY ) <EOL> mock_fdopen . assert_called_with ( mock_open . return_value , '<STR_LIT:w>' ) <EOL> self . assertTrue ( mock_fdopen . return_value . write . called ) <EOL> mock_islink . assert_called_with ( fpath ) <EOL> mock_rename . assert_called_with ( tmpfile , fpath ) <EOL> mock_load_xml . assert_called_with ( ) <EOL> def rv ( fname , mode ) : <EOL> mock_open . side_effect = None <EOL> raise OSError ( <NUM_LIT> , fname ) <EOL> reset ( ) <EOL> mock_open . side_effect = rv <EOL> config . write_xml ( fpath , get_clients_test_tree ( ) ) <EOL> self . assertItemsEqual ( mock_open . call_args_list , <EOL> [ call ( tmpfile , <EOL> os . O_CREAT | os . O_EXCL | os . O_WRONLY ) , <EOL> call ( tmpfile , <EOL> os . O_CREAT | os . O_EXCL | os . O_WRONLY ) ] ) <EOL> mock_fdopen . assert_called_with ( mock_open . return_value , '<STR_LIT:w>' ) <EOL> self . assertTrue ( mock_fdopen . return_value . write . called ) <EOL> mock_islink . assert_called_with ( fpath ) <EOL> mock_rename . assert_called_with ( tmpfile , fpath ) <EOL> mock_load_xml . assert_called_with ( ) <EOL> reset ( ) <EOL> mock_open . side_effect = None <EOL> mock_islink . return_value = True <EOL> mock_readlink . return_value = linkdest <EOL> config . write_xml ( fpath , get_clients_test_tree ( ) ) <EOL> mock_rename . assert_called_with ( tmpfile , linkdest ) <EOL> mock_load_xml . assert_called_with ( ) <EOL> reset ( ) <EOL> mock_rename . side_effect = OSError <EOL> self . assertRaises ( Bcfg2 . Server . Plugin . MetadataRuntimeError , <EOL> config . write_xml , fpath , get_clients_test_tree ( ) ) <EOL> mock_unlink . assert_called_with ( tmpfile ) <EOL> reset ( ) <EOL> mock_open . return_value . write . side_effect = IOError <EOL> self . assertRaises ( Bcfg2 . Server . Plugin . MetadataRuntimeError , <EOL> config . write_xml , fpath , get_clients_test_tree ( ) ) <EOL> mock_unlink . assert_called_with ( tmpfile ) <EOL> reset ( ) <EOL> mock_open . side_effect = OSError <EOL> self . assertRaises ( Bcfg2 . Server . Plugin . MetadataRuntimeError , <EOL> config . write_xml , fpath , get_clients_test_tree ( ) ) <EOL> reset ( ) <EOL> mock_fdopen . side_effect = OSError <EOL> self . assertRaises ( Bcfg2 . Server . Plugin . MetadataRuntimeError , <EOL> config . write_xml , fpath , get_clients_test_tree ( ) ) <EOL> @ patch ( "<STR_LIT>" , Mock ( ) ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_find_xml_for_xpath ( self , mock_parse ) : <EOL> config = self . get_obj ( "<STR_LIT>" ) <EOL> config . basedata = get_groups_test_tree ( ) <EOL> xpath = "<STR_LIT>" <EOL> self . assertItemsEqual ( config . find_xml_for_xpath ( xpath ) , <EOL> dict ( filename = os . path . join ( self . metadata . data , <EOL> "<STR_LIT>" ) , <EOL> xmltree = get_groups_test_tree ( ) , <EOL> xquery = get_groups_test_tree ( ) . xpath ( xpath ) ) ) <EOL> self . assertEqual ( config . find_xml_for_xpath ( "<STR_LIT>" ) , dict ( ) ) <EOL> config . extras = [ os . path . join ( self . metadata . data , p ) <EOL> for p in [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ] <EOL> def parse_side_effect ( fname , parser = Bcfg2 . Server . XMLParser ) : <EOL> if fname == os . path . join ( self . metadata . data , "<STR_LIT>" ) : <EOL> return get_clients_test_tree ( ) <EOL> else : <EOL> return lxml . etree . XML ( "<STR_LIT>" ) . getroottree ( ) <EOL> mock_parse . side_effect = parse_side_effect <EOL> xpath = "<STR_LIT>" <EOL> self . assertItemsEqual ( config . find_xml_for_xpath ( xpath ) , <EOL> dict ( filename = os . path . join ( self . metadata . data , <EOL> "<STR_LIT>" ) , <EOL> xmltree = get_clients_test_tree ( ) , <EOL> xquery = get_clients_test_tree ( ) . xpath ( xpath ) ) ) <EOL> @ patch ( "<STR_LIT>" ) <EOL> def test_HandleEvent ( self , mock_load_xml ) : <EOL> config = self . get_obj ( "<STR_LIT>" ) <EOL> evt = Mock ( ) <EOL> evt . filename = os . path . join ( self . metadata . data , "<STR_LIT>" ) <EOL> evt . code2str = Mock ( return_value = "<STR_LIT>" ) <EOL> self . assertTrue ( config . HandleEvent ( evt ) ) <EOL> mock_load_xml . assert_called_with ( ) <EOL> class TestClientMetadata ( Bcfg2TestCase ) : <EOL> def test_inGroup ( self ) : <EOL> cm = ClientMetadata ( "<STR_LIT>" , "<STR_LIT>" , [ "<STR_LIT>" , "<STR_LIT>" ] , <EOL> [ "<STR_LIT>" ] , [ ] , [ ] , [ ] , None , None , None , None ) <EOL> self . assertTrue ( cm . inGroup ( "<STR_LIT>" ) ) <EOL> self . assertFalse ( cm . inGroup ( "<STR_LIT>" ) ) <EOL> class TestMetadata ( _TestMetadata , TestClientRunHooks , TestDatabaseBacked ) : <EOL> test_obj = Metadata <EOL> def setUp ( self ) : <EOL> _TestMetadata . setUp ( self ) <EOL> TestClientRunHooks . setUp ( self ) <EOL> TestDatabaseBacked . setUp ( self ) <EOL> Bcfg2 . Options . setup . metadata_db = False <EOL> Bcfg2 . Options . setup . authentication = "<STR_LIT>" <EOL> def get_obj ( self , core = None ) : <EOL> return get_metadata_object ( core = core ) <EOL> @ skipUnless ( HAS_DJANGO , "<STR_LIT>" ) <EOL> def test__use_db ( self ) : <EOL> pass <EOL> def get_nonexistent_client ( self , metadata , prefix = "<STR_LIT>" ) : <EOL> if metadata is None : <EOL> metadata = self . load_clients_data ( ) <EOL> i = <NUM_LIT:0> <EOL> client_name = "<STR_LIT>" % ( prefix , i ) <EOL> while client_name in metadata . clients : <EOL> i += <NUM_LIT:1> <EOL> client_name = "<STR_LIT>" % ( prefix , i ) <EOL> return client_name <EOL> @ patch ( "<STR_LIT>" ) <EOL> def test__init ( self , mock_get_fam ) : <EOL> core = MagicMock ( ) <EOL> metadata = self . get_obj ( core = core ) <EOL> self . assertEqual ( len ( metadata . states ) , <NUM_LIT:2> ) <EOL> mock_get_fam . return_value . AddMonitor . assert_any_call ( <EOL> os . path . join ( metadata . data , "<STR_LIT>" ) , <EOL> metadata ) <EOL> mock_get_fam . return_value . AddMonitor . assert_any_call ( <EOL> os . path . join ( metadata . data , "<STR_LIT>" ) , <EOL> metadata ) <EOL> @ patch ( '<STR_LIT>' , Mock ( ) ) <EOL> @ patch ( '<STR_LIT>' % builtins ) <EOL> def test_init_repo ( self , mock_open ) : <EOL> Metadata . init_repo ( datastore , <EOL> groups_xml = "<STR_LIT>" , clients_xml = "<STR_LIT>" ) <EOL> mock_open . assert_any_call ( os . path . join ( datastore , "<STR_LIT>" , <EOL> "<STR_LIT>" ) , "<STR_LIT:w>" ) <EOL> mock_open . assert_any_call ( os . path . join ( datastore , "<STR_LIT>" , <EOL> "<STR_LIT>" ) , "<STR_LIT:w>" ) <EOL> def test_search_xdata ( self ) : <EOL> metadata = self . get_obj ( ) <EOL> tree = get_groups_test_tree ( ) <EOL> res = metadata . _search_xdata ( "<STR_LIT>" , "<STR_LIT>" , tree ) <EOL> self . assertIsInstance ( res , lxml . etree . _Element ) <EOL> self . assertEqual ( res . get ( "<STR_LIT:name>" ) , "<STR_LIT>" ) <EOL> metadata = self . get_obj ( ) <EOL> tree = get_clients_test_tree ( ) <EOL> res = metadata . _search_xdata ( "<STR_LIT>" , "<STR_LIT>" , tree , alias = True ) <EOL> self . assertIsInstance ( res , lxml . etree . _Element ) <EOL> self . assertNotEqual ( res . get ( "<STR_LIT:name>" ) , "<STR_LIT>" ) <EOL> metadata = self . get_obj ( ) <EOL> tree = get_clients_test_tree ( ) <EOL> res = metadata . _search_xdata ( "<STR_LIT>" , <EOL> self . get_nonexistent_client ( metadata ) , <EOL> tree , alias = True ) <EOL> self . assertIsNone ( res ) <EOL> def search_xdata ( self , tag , name , tree , alias = False ) : <EOL> metadata = self . get_obj ( ) <EOL> res = metadata . _search_xdata ( tag , name , tree , alias = alias ) <EOL> self . assertIsInstance ( res , lxml . etree . _Element ) <EOL> if not alias : <EOL> self . assertEqual ( res . get ( "<STR_LIT:name>" ) , name ) <EOL> def test_search_group ( self ) : <EOL> tree = get_groups_test_tree ( ) <EOL> self . search_xdata ( "<STR_LIT>" , "<STR_LIT>" , tree ) <EOL> def test_search_bundle ( self ) : <EOL> tree = get_groups_test_tree ( ) <EOL> self . search_xdata ( "<STR_LIT>" , "<STR_LIT>" , tree ) <EOL> def test_search_client ( self ) : <EOL> tree = get_clients_test_tree ( ) <EOL> self . search_xdata ( "<STR_LIT>" , "<STR_LIT>" , tree , alias = True ) <EOL> self . search_xdata ( "<STR_LIT>" , "<STR_LIT>" , tree , alias = True ) <EOL> def test_add_group ( self ) : <EOL> metadata = self . get_obj ( ) <EOL> metadata . groups_xml . write = Mock ( ) <EOL> metadata . groups_xml . load_xml = Mock ( ) <EOL> metadata . groups_xml . data = lxml . etree . XML ( '<STR_LIT>' ) . getroottree ( ) <EOL> metadata . groups_xml . basedata = copy . copy ( metadata . groups_xml . data ) <EOL> metadata . add_group ( "<STR_LIT>" , dict ( ) ) <EOL> metadata . groups_xml . write . assert_any_call ( ) <EOL> grp = metadata . search_group ( "<STR_LIT>" , metadata . groups_xml . base_xdata ) <EOL> self . assertIsNotNone ( grp ) <EOL> self . assertEqual ( grp . attrib , dict ( name = '<STR_LIT>' ) ) <EOL> metadata . groups_xml . basedata = copy . copy ( metadata . groups_xml . data ) <EOL> metadata . add_group ( "<STR_LIT>" , dict ( foo = '<STR_LIT:bar>' ) ) <EOL> metadata . groups_xml . write . assert_any_call ( ) <EOL> grp = metadata . search_group ( "<STR_LIT>" , metadata . groups_xml . base_xdata ) <EOL> self . assertIsNotNone ( grp ) <EOL> self . assertEqual ( grp . attrib , dict ( name = '<STR_LIT>' , foo = '<STR_LIT:bar>' ) ) <EOL> metadata . groups_xml . basedata = copy . copy ( metadata . groups_xml . data ) <EOL> metadata . groups_xml . write . reset_mock ( ) <EOL> self . assertRaises ( Bcfg2 . Server . Plugin . MetadataConsistencyError , <EOL> metadata . add_group , <EOL> "<STR_LIT>" , dict ( ) ) <EOL> self . assertFalse ( metadata . groups_xml . write . called ) <EOL> def test_update_group ( self ) : <EOL> metadata = self . get_obj ( ) <EOL> metadata . groups_xml . write_xml = Mock ( ) <EOL> metadata . groups_xml . load_xml = Mock ( ) <EOL> metadata . groups_xml . data = copy . deepcopy ( get_groups_test_tree ( ) ) <EOL> metadata . groups_xml . basedata = copy . copy ( metadata . groups_xml . data ) <EOL> metadata . update_group ( "<STR_LIT>" , dict ( foo = "<STR_LIT:bar>" ) ) <EOL> grp = metadata . search_group ( "<STR_LIT>" , metadata . groups_xml . base_xdata ) <EOL> self . assertIsNotNone ( grp ) <EOL> self . assertIn ( "<STR_LIT:foo>" , grp . attrib ) <EOL> self . assertEqual ( grp . get ( "<STR_LIT:foo>" ) , "<STR_LIT:bar>" ) <EOL> self . assertTrue ( metadata . groups_xml . write_xml . called ) <EOL> self . assertRaises ( Bcfg2 . Server . Plugin . MetadataConsistencyError , <EOL> metadata . update_group , <EOL> "<STR_LIT>" , dict ( ) ) <EOL> def test_remove_group ( self ) : <EOL> metadata = self . get_obj ( ) <EOL> metadata . groups_xml . write_xml = Mock ( ) <EOL> metadata . groups_xml . load_xml = Mock ( ) <EOL> metadata . groups_xml . data = copy . deepcopy ( get_groups_test_tree ( ) ) <EOL> metadata . groups_xml . basedata = copy . copy ( metadata . groups_xml . data ) <EOL> metadata . remove_group ( "<STR_LIT>" ) <EOL> grp = metadata . search_group ( "<STR_LIT>" , metadata . groups_xml . base_xdata ) <EOL> self . assertIsNone ( grp ) <EOL> self . assertTrue ( metadata . groups_xml . write_xml . called ) <EOL> self . assertRaises ( Bcfg2 . Server . Plugin . MetadataConsistencyError , <EOL> metadata . remove_group , <EOL> "<STR_LIT>" ) <EOL> def test_add_bundle ( self ) : <EOL> metadata = self . get_obj ( ) <EOL> metadata . groups_xml . write = Mock ( ) <EOL> metadata . groups_xml . load_xml = Mock ( ) <EOL> metadata . groups_xml . data = lxml . etree . XML ( '<STR_LIT>' ) . getroottree ( ) <EOL> metadata . groups_xml . basedata = copy . copy ( metadata . groups_xml . data ) <EOL> metadata . add_bundle ( "<STR_LIT>" ) <EOL> metadata . groups_xml . write . assert_any_call ( ) <EOL> bundle = metadata . search_bundle ( "<STR_LIT>" , <EOL> metadata . groups_xml . base_xdata ) <EOL> self . assertIsNotNone ( bundle ) <EOL> self . assertEqual ( bundle . attrib , dict ( name = '<STR_LIT>' ) ) <EOL> metadata . groups_xml . basedata = copy . copy ( metadata . groups_xml . data ) <EOL> metadata . groups_xml . write . reset_mock ( ) <EOL> self . assertRaises ( Bcfg2 . Server . Plugin . MetadataConsistencyError , <EOL> metadata . add_bundle , <EOL> "<STR_LIT>" ) <EOL> self . assertFalse ( metadata . groups_xml . write . called ) <EOL> def test_remove_bundle ( self ) : <EOL> metadata = self . get_obj ( ) <EOL> metadata . groups_xml . write_xml = Mock ( ) <EOL> metadata . groups_xml . load_xml = Mock ( ) <EOL> metadata . groups_xml . data = copy . deepcopy ( get_groups_test_tree ( ) ) <EOL> metadata . groups_xml . basedata = copy . copy ( metadata . groups_xml . data ) <EOL> metadata . remove_bundle ( "<STR_LIT>" ) <EOL> grp = metadata . search_bundle ( "<STR_LIT>" , metadata . groups_xml . base_xdata ) <EOL> self . assertIsNone ( grp ) <EOL> self . assertTrue ( metadata . groups_xml . write_xml . called ) <EOL> self . assertRaises ( Bcfg2 . Server . Plugin . MetadataConsistencyError , <EOL> metadata . remove_bundle , <EOL> "<STR_LIT>" ) <EOL> def test_add_client ( self ) : <EOL> metadata = self . get_obj ( ) <EOL> metadata . clients_xml . write = Mock ( ) <EOL> metadata . clients_xml . load_xml = Mock ( ) <EOL> metadata . clients_xml . data = lxml . etree . XML ( '<STR_LIT>' ) . getroottree ( ) <EOL> metadata . clients_xml . basedata = copy . copy ( metadata . clients_xml . data ) <EOL> new1 = self . get_nonexistent_client ( metadata ) <EOL> new1_client = metadata . add_client ( new1 , dict ( ) ) <EOL> metadata . clients_xml . write . assert_any_call ( ) <EOL> grp = metadata . search_client ( new1 , metadata . clients_xml . base_xdata ) <EOL> self . assertIsNotNone ( grp ) <EOL> self . assertEqual ( grp . attrib , dict ( name = new1 ) ) <EOL> metadata . clients_xml . basedata = copy . copy ( metadata . clients_xml . data ) <EOL> metadata . _handle_clients_xml_event ( Mock ( ) ) <EOL> new2 = self . get_nonexistent_client ( metadata ) <EOL> metadata . add_client ( new2 , dict ( foo = '<STR_LIT:bar>' ) ) <EOL> metadata . clients_xml . write . assert_any_call ( ) <EOL> grp = metadata . search_client ( new2 , metadata . clients_xml . base_xdata ) <EOL> self . assertIsNotNone ( grp ) <EOL> self . assertEqual ( grp . attrib , dict ( name = new2 , foo = '<STR_LIT:bar>' ) ) <EOL> metadata . clients_xml . basedata = copy . copy ( metadata . clients_xml . data ) <EOL> metadata . clients_xml . write . reset_mock ( ) <EOL> self . assertXMLEqual ( metadata . add_client ( new1 , dict ( ) ) , <EOL> new1_client ) <EOL> self . assertFalse ( metadata . clients_xml . write . called ) <EOL> def test_update_client ( self ) : <EOL> metadata = self . get_obj ( ) <EOL> metadata . clients_xml . write_xml = Mock ( ) <EOL> metadata . clients_xml . load_xml = Mock ( ) <EOL> metadata . clients_xml . data = copy . deepcopy ( get_clients_test_tree ( ) ) <EOL> metadata . clients_xml . basedata = copy . copy ( metadata . clients_xml . data ) <EOL> metadata . update_client ( "<STR_LIT>" , dict ( foo = "<STR_LIT:bar>" ) ) <EOL> grp = metadata . search_client ( "<STR_LIT>" , metadata . clients_xml . base_xdata ) <EOL> self . assertIsNotNone ( grp ) <EOL> self . assertIn ( "<STR_LIT:foo>" , grp . attrib ) <EOL> self . assertEqual ( grp . get ( "<STR_LIT:foo>" ) , "<STR_LIT:bar>" ) <EOL> self . assertTrue ( metadata . clients_xml . write_xml . called ) <EOL> new = self . get_nonexistent_client ( metadata ) <EOL> self . assertRaises ( Bcfg2 . Server . Plugin . MetadataConsistencyError , <EOL> metadata . update_client , <EOL> new , dict ( ) ) <EOL> def load_clients_data ( self , metadata = None , xdata = None ) : <EOL> if metadata is None : <EOL> metadata = self . get_obj ( ) <EOL> metadata . clients_xml . data = xdata or copy . deepcopy ( get_clients_test_tree ( ) ) <EOL> metadata . clients_xml . basedata = copy . copy ( metadata . clients_xml . data ) <EOL> evt = Mock ( ) <EOL> evt . filename = os . path . join ( datastore , "<STR_LIT>" , "<STR_LIT>" ) <EOL> evt . code2str = Mock ( return_value = "<STR_LIT>" ) <EOL> metadata . HandleEvent ( evt ) <EOL> return metadata <EOL> def test_handle_clients_xml_event ( self ) : <EOL> metadata = self . get_obj ( ) <EOL> metadata . profiles = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> metadata . clients_xml = Mock ( ) <EOL> metadata . clients_xml . xdata = copy . deepcopy ( get_clients_test_tree ( ) ) <EOL> metadata . _handle_clients_xml_event ( Mock ( ) ) <EOL> if not Bcfg2 . Options . setup . metadata_db : <EOL> self . assertItemsEqual ( metadata . clients , <EOL> dict ( [ ( c . get ( "<STR_LIT:name>" ) , c . get ( "<STR_LIT>" ) ) <EOL> for c in get_clients_test_tree ( ) . findall ( "<STR_LIT>" ) ] ) ) <EOL> aliases = dict ( [ ( a . get ( "<STR_LIT:name>" ) , a . getparent ( ) . get ( "<STR_LIT:name>" ) ) <EOL> for a in get_clients_test_tree ( ) . findall ( "<STR_LIT>" ) ] ) <EOL> self . assertItemsEqual ( metadata . aliases , aliases ) <EOL> raliases = dict ( [ ( c . get ( "<STR_LIT:name>" ) , set ( ) ) <EOL> for c in get_clients_test_tree ( ) . findall ( "<STR_LIT>" ) ] ) <EOL> for alias in get_clients_test_tree ( ) . findall ( "<STR_LIT>" ) : <EOL> raliases [ alias . getparent ( ) . get ( "<STR_LIT:name>" ) ] . add ( alias . get ( "<STR_LIT:name>" ) ) <EOL> self . assertItemsEqual ( metadata . raliases , raliases ) <EOL> self . assertEqual ( metadata . secure , <EOL> [ c . get ( "<STR_LIT:name>" ) <EOL> for c in get_clients_test_tree ( ) . findall ( "<STR_LIT>" ) ] ) <EOL> self . assertEqual ( metadata . floating , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> addresses = dict ( [ ( c . get ( "<STR_LIT:address>" ) , [ ] ) <EOL> for c in get_clients_test_tree ( ) . findall ( "<STR_LIT>" ) ] ) <EOL> raddresses = dict ( ) <EOL> for client in get_clients_test_tree ( ) . findall ( "<STR_LIT>" ) : <EOL> addresses [ client . get ( "<STR_LIT:address>" ) ] . append ( client . get ( "<STR_LIT:name>" ) ) <EOL> try : <EOL> raddresses [ client . get ( "<STR_LIT:name>" ) ] . append ( client . get ( "<STR_LIT:address>" ) ) <EOL> except KeyError : <EOL> raddresses [ client . get ( "<STR_LIT:name>" ) ] = [ client . get ( "<STR_LIT:address>" ) ] <EOL> for alias in get_clients_test_tree ( ) . findall ( "<STR_LIT>" ) : <EOL> addresses [ alias . get ( "<STR_LIT:address>" ) ] . append ( alias . getparent ( ) . get ( "<STR_LIT:name>" ) ) <EOL> try : <EOL> raddresses [ alias . getparent ( ) . get ( "<STR_LIT:name>" ) ] . append ( alias . get ( "<STR_LIT:address>" ) ) <EOL> except KeyError : <EOL> raddresses [ alias . getparent ( ) . get ( "<STR_LIT:name>" ) ] = alias . get ( "<STR_LIT:address>" ) <EOL> self . assertItemsEqual ( metadata . addresses , addresses ) <EOL> self . assertItemsEqual ( metadata . raddresses , raddresses ) <EOL> self . assertTrue ( metadata . states [ '<STR_LIT>' ] ) <EOL> def load_groups_data ( self , metadata = None , xdata = None ) : <EOL> if metadata is None : <EOL> metadata = self . get_obj ( ) <EOL> metadata . groups_xml . data = xdata or copy . deepcopy ( get_groups_test_tree ( ) ) <EOL> metadata . groups_xml . basedata = copy . copy ( metadata . groups_xml . data ) <EOL> evt = Mock ( ) <EOL> evt . filename = os . path . join ( datastore , "<STR_LIT>" , "<STR_LIT>" ) <EOL> evt . code2str = Mock ( return_value = "<STR_LIT>" ) <EOL> metadata . HandleEvent ( evt ) <EOL> return metadata <EOL> def test_handle_groups_xml_event ( self ) : <EOL> metadata = self . get_obj ( ) <EOL> metadata . groups_xml = Mock ( ) <EOL> metadata . groups_xml . xdata = get_groups_test_tree ( ) <EOL> metadata . _handle_groups_xml_event ( Mock ( ) ) <EOL> self . assertTrue ( metadata . states [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( metadata . groups [ '<STR_LIT>' ] . is_public ) <EOL> self . assertTrue ( metadata . groups [ '<STR_LIT>' ] . is_public ) <EOL> self . assertFalse ( metadata . groups [ '<STR_LIT>' ] . is_public ) <EOL> self . assertTrue ( metadata . groups [ '<STR_LIT>' ] . is_profile ) <EOL> self . assertTrue ( metadata . groups [ '<STR_LIT>' ] . is_profile ) <EOL> self . assertFalse ( metadata . groups [ '<STR_LIT>' ] . is_profile ) <EOL> self . assertItemsEqual ( metadata . groups . keys ( ) , <EOL> set ( g . get ( "<STR_LIT:name>" ) <EOL> for g in get_groups_test_tree ( ) . findall ( "<STR_LIT>" ) ) ) <EOL> self . assertEqual ( metadata . groups [ '<STR_LIT>' ] . category , '<STR_LIT>' ) <EOL> self . assertEqual ( metadata . groups [ '<STR_LIT>' ] . category , '<STR_LIT>' ) <EOL> self . assertEqual ( metadata . groups [ '<STR_LIT>' ] . category , '<STR_LIT>' ) <EOL> self . assertEqual ( metadata . groups [ '<STR_LIT>' ] . category , '<STR_LIT>' ) <EOL> self . assertEqual ( metadata . default , "<STR_LIT>" ) <EOL> all_groups = set ( ) <EOL> negated_groups = set ( ) <EOL> for group in get_groups_test_tree ( ) . xpath ( "<STR_LIT>" ) + get_groups_test_tree ( ) . xpath ( "<STR_LIT>" ) : <EOL> if group . tag == '<STR_LIT>' and not group . getchildren ( ) : <EOL> if group . get ( "<STR_LIT>" , "<STR_LIT:false>" ) . lower ( ) == '<STR_LIT:true>' : <EOL> negated_groups . add ( group . get ( "<STR_LIT:name>" ) ) <EOL> else : <EOL> all_groups . add ( group . get ( "<STR_LIT:name>" ) ) <EOL> self . assertItemsEqual ( metadata . ordered_groups , all_groups ) <EOL> self . assertItemsEqual ( metadata . group_membership . keys ( ) , all_groups ) <EOL> self . assertItemsEqual ( metadata . negated_groups . keys ( ) , negated_groups ) <EOL> @ patch ( "<STR_LIT>" , Mock ( ) ) <EOL> def test_set_profile ( self ) : <EOL> metadata = self . get_obj ( ) <EOL> if '<STR_LIT>' in metadata . states : <EOL> metadata . states [ '<STR_LIT>' ] = False <EOL> self . assertRaises ( Bcfg2 . Server . Plugin . MetadataRuntimeError , <EOL> metadata . set_profile , <EOL> None , None , None ) <EOL> self . load_groups_data ( metadata = metadata ) <EOL> self . load_clients_data ( metadata = metadata ) <EOL> self . assertRaises ( Bcfg2 . Server . Plugin . MetadataConsistencyError , <EOL> metadata . set_profile , <EOL> "<STR_LIT>" , "<STR_LIT>" , None ) <EOL> self . assertRaises ( Bcfg2 . Server . Plugin . MetadataConsistencyError , <EOL> metadata . set_profile , <EOL> "<STR_LIT>" , "<STR_LIT>" , None ) <EOL> @ patch ( "<STR_LIT>" , Mock ( ) ) <EOL> def test_set_profile_db ( self ) : <EOL> metadata = self . load_clients_data ( metadata = self . load_groups_data ( ) ) <EOL> if metadata . _use_db : <EOL> profile = "<STR_LIT>" <EOL> client_name = self . get_nonexistent_client ( metadata ) <EOL> metadata . set_profile ( client_name , profile , None ) <EOL> self . assertIn ( client_name , metadata . clients ) <EOL> self . assertRaises ( Bcfg2 . Server . Plugin . PluginExecutionError , <EOL> metadata . set_profile , <EOL> client_name , profile , None ) <EOL> @ patch ( "<STR_LIT>" , Mock ( ) ) <EOL> @ patch ( "<STR_LIT>" ) <EOL> @ patch ( "<STR_LIT>" ) <EOL> def test_set_profile_xml ( self , mock_update_client , mock_add_client ) : <EOL> metadata = self . load_clients_data ( metadata = self . load_groups_data ( ) ) <EOL> if not metadata . _use_db : <EOL> metadata . clients_xml . write = Mock ( ) <EOL> metadata . core . build_metadata = Mock ( ) <EOL> metadata . core . build_metadata . side_effect = lambda c : metadata . get_initial_metadata ( c ) <EOL> metadata . set_profile ( "<STR_LIT>" , "<STR_LIT>" , None ) <EOL> mock_update_client . assert_called_with ( "<STR_LIT>" , <EOL> dict ( profile = "<STR_LIT>" ) ) <EOL> self . assertEqual ( metadata . clientgroups [ "<STR_LIT>" ] , [ "<STR_LIT>" ] ) <EOL> metadata . clients_xml . write . reset_mock ( ) <EOL> new1 = self . get_nonexistent_client ( metadata ) <EOL> metadata . set_profile ( new1 , "<STR_LIT>" , None ) <EOL> mock_add_client . assert_called_with ( new1 , dict ( profile = "<STR_LIT>" ) ) <EOL> metadata . clients_xml . write . assert_any_call ( ) <EOL> self . assertEqual ( metadata . clientgroups [ new1 ] , [ "<STR_LIT>" ] ) <EOL> metadata . clients_xml . write . reset_mock ( ) <EOL> new2 = self . get_nonexistent_client ( metadata ) <EOL> metadata . session_cache [ ( '<STR_LIT>' , None ) ] = ( None , new2 ) <EOL> metadata . set_profile ( "<STR_LIT>" , "<STR_LIT>" , ( '<STR_LIT>' , None ) ) <EOL> mock_add_client . assert_called_with ( new2 , <EOL> dict ( uuid = '<STR_LIT>' , <EOL> profile = "<STR_LIT>" , <EOL> address = '<STR_LIT>' ) ) <EOL> metadata . clients_xml . write . assert_any_call ( ) <EOL> self . assertEqual ( metadata . clientgroups [ "<STR_LIT>" ] , [ "<STR_LIT>" ] ) <EOL> @ patch ( "<STR_LIT>" , Mock ( ) ) <EOL> @ patch ( "<STR_LIT>" ) <EOL> def test_resolve_client ( self , mock_getnameinfo ) : <EOL> metadata = self . load_clients_data ( metadata = self . load_groups_data ( ) ) <EOL> metadata . session_cache [ ( '<STR_LIT>' , None ) ] = ( time . time ( ) , '<STR_LIT>' ) <EOL> self . assertEqual ( metadata . resolve_client ( ( '<STR_LIT>' , None ) ) , '<STR_LIT>' ) <EOL> self . assertRaises ( Bcfg2 . Server . Plugin . MetadataConsistencyError , <EOL> metadata . resolve_client , <EOL> ( '<STR_LIT>' , None ) ) <EOL> self . assertEqual ( metadata . resolve_client ( ( '<STR_LIT>' , None ) ) , '<STR_LIT>' ) <EOL> metadata . session_cache [ ( '<STR_LIT>' , None ) ] = ( time . time ( ) - <NUM_LIT:100> , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( metadata . resolve_client ( ( '<STR_LIT>' , None ) ) , '<STR_LIT>' ) <EOL> self . assertEqual ( metadata . resolve_client ( ( '<STR_LIT>' , None ) , <EOL> cleanup_cache = True ) , '<STR_LIT>' ) <EOL> self . assertEqual ( metadata . session_cache , dict ( ) ) <EOL> mock_getnameinfo . return_value = ( '<STR_LIT>' , [ ] , [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( metadata . resolve_client ( ( '<STR_LIT>' , <NUM_LIT> ) ) , '<STR_LIT>' ) <EOL> mock_getnameinfo . assert_called_with ( ( '<STR_LIT>' , <NUM_LIT> ) , socket . NI_NAMEREQD ) <EOL> mock_getnameinfo . reset_mock ( ) <EOL> mock_getnameinfo . return_value = ( '<STR_LIT>' , [ ] , [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( metadata . resolve_client ( ( '<STR_LIT>' , <NUM_LIT> ) ) , '<STR_LIT>' ) <EOL> mock_getnameinfo . assert_called_with ( ( '<STR_LIT>' , <NUM_LIT> ) , socket . NI_NAMEREQD ) <EOL> mock_getnameinfo . reset_mock ( ) <EOL> mock_getnameinfo . return_value = None <EOL> mock_getnameinfo . side_effect = socket . herror <EOL> self . assertRaises ( Bcfg2 . Server . Plugin . MetadataConsistencyError , <EOL> metadata . resolve_client , <EOL> ( '<STR_LIT>' , <NUM_LIT> ) ) <EOL> mock_getnameinfo . assert_called_with ( ( '<STR_LIT>' , <NUM_LIT> ) , socket . NI_NAMEREQD ) <EOL> @ patch ( "<STR_LIT>" , Mock ( ) ) <EOL> @ patch ( "<STR_LIT>" , Mock ( ) ) <EOL> @ patch ( "<STR_LIT>" ) <EOL> def test_get_initial_metadata ( self , mock_clientmetadata ) : <EOL> metadata = self . get_obj ( ) <EOL> if '<STR_LIT>' in metadata . states : <EOL> metadata . states [ '<STR_LIT>' ] = False <EOL> self . assertRaises ( Bcfg2 . Server . Plugin . MetadataRuntimeError , <EOL> metadata . get_initial_metadata , None ) <EOL> self . load_groups_data ( metadata = metadata ) <EOL> self . load_clients_data ( metadata = metadata ) <EOL> metadata . get_initial_metadata ( "<STR_LIT>" ) <EOL> mock_clientmetadata . assert_called_with ( "<STR_LIT>" , "<STR_LIT>" , <EOL> set ( [ "<STR_LIT>" ] ) , set ( ) , set ( ) , <EOL> set ( [ "<STR_LIT>" ] ) , <EOL> dict ( category1 = '<STR_LIT>' ) , None , <EOL> '<STR_LIT>' , None , <EOL> metadata . query ) <EOL> metadata . get_initial_metadata ( "<STR_LIT>" ) <EOL> mock_clientmetadata . assert_called_with ( "<STR_LIT>" , "<STR_LIT>" , <EOL> set ( [ "<STR_LIT>" ] ) , <EOL> set ( [ "<STR_LIT>" , "<STR_LIT>" ] ) , <EOL> set ( ) , set ( [ "<STR_LIT>" ] ) , <EOL> dict ( category1 = "<STR_LIT>" ) , <EOL> None , None , None , <EOL> metadata . query ) <EOL> imd = metadata . get_initial_metadata ( "<STR_LIT>" ) <EOL> mock_clientmetadata . assert_called_with ( "<STR_LIT>" , "<STR_LIT>" , <EOL> set ( [ "<STR_LIT>" ] ) , set ( ) , <EOL> set ( [ '<STR_LIT>' ] ) , <EOL> set ( [ "<STR_LIT>" ] ) , <EOL> dict ( category1 = "<STR_LIT>" ) , <EOL> '<STR_LIT>' , '<STR_LIT>' , None , <EOL> metadata . query ) <EOL> new1 = self . get_nonexistent_client ( metadata ) <EOL> imd = metadata . get_initial_metadata ( new1 ) <EOL> mock_clientmetadata . assert_called_with ( new1 , "<STR_LIT>" , set ( [ "<STR_LIT>" ] ) , <EOL> set ( ) , set ( ) , set ( ) , <EOL> dict ( category1 = "<STR_LIT>" ) , None , <EOL> None , None , metadata . query ) <EOL> imd = metadata . get_initial_metadata ( "<STR_LIT>" ) <EOL> mock_clientmetadata . assert_called_with ( "<STR_LIT>" , "<STR_LIT>" , <EOL> set ( [ "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" ] ) , <EOL> set ( ) , <EOL> set ( ) , set ( [ "<STR_LIT>" ] ) , <EOL> dict ( category1 = "<STR_LIT>" ) , <EOL> None , None , None , metadata . query ) <EOL> imd = metadata . get_initial_metadata ( "<STR_LIT>" ) <EOL> mock_clientmetadata . assert_called_with ( "<STR_LIT>" , "<STR_LIT>" , <EOL> set ( [ "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" ] ) , <EOL> set ( [ "<STR_LIT>" , "<STR_LIT>" ] ) , <EOL> set ( ) , set ( ) , <EOL> dict ( category1 = "<STR_LIT>" ) , None , <EOL> "<STR_LIT>" , None , <EOL> metadata . query ) <EOL> metadata . default = None <EOL> new2 = self . get_nonexistent_client ( metadata ) <EOL> self . assertRaises ( Bcfg2 . Server . Plugin . MetadataConsistencyError , <EOL> metadata . get_initial_metadata , new2 ) <EOL> @ patch ( "<STR_LIT>" , Mock ( ) ) <EOL> def test_merge_groups ( self ) : <EOL> metadata = self . get_obj ( ) <EOL> self . load_groups_data ( metadata = metadata ) <EOL> self . load_clients_data ( metadata = metadata ) <EOL> self . assertEqual ( metadata . _merge_groups ( "<STR_LIT>" , set ( [ "<STR_LIT>" ] ) , <EOL> categories = dict ( group1 = "<STR_LIT>" ) ) , <EOL> ( set ( [ "<STR_LIT>" ] ) , dict ( group1 = "<STR_LIT>" ) ) ) <EOL> self . assertEqual ( metadata . _merge_groups ( "<STR_LIT>" , <EOL> set ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) , <EOL> categories = dict ( group1 = "<STR_LIT>" ) ) , <EOL> ( set ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) , <EOL> dict ( group1 = "<STR_LIT>" ) ) ) <EOL> @ patch ( "<STR_LIT>" , Mock ( ) ) <EOL> def test_get_all_group_names ( self ) : <EOL> metadata = self . load_groups_data ( ) <EOL> self . assertItemsEqual ( metadata . get_all_group_names ( ) , <EOL> set ( [ g . get ( "<STR_LIT:name>" ) <EOL> for g in get_groups_test_tree ( ) . findall ( "<STR_LIT>" ) ] ) ) <EOL> @ patch ( "<STR_LIT>" , Mock ( ) ) <EOL> def test_get_all_groups_in_category ( self ) : <EOL> metadata = self . load_groups_data ( ) <EOL> self . assertItemsEqual ( metadata . get_all_groups_in_category ( "<STR_LIT>" ) , <EOL> set ( [ g . get ( "<STR_LIT:name>" ) <EOL> for g in get_groups_test_tree ( ) . findall ( "<STR_LIT>" ) ] ) ) <EOL> @ patch ( "<STR_LIT>" , Mock ( ) ) <EOL> def test_get_client_names_by_profiles ( self ) : <EOL> metadata = self . load_clients_data ( metadata = self . load_groups_data ( ) ) <EOL> metadata . core . build_metadata = Mock ( ) <EOL> metadata . core . build_metadata . side_effect = lambda c : metadata . get_initial_metadata ( c ) <EOL> self . assertItemsEqual ( metadata . get_client_names_by_profiles ( [ "<STR_LIT>" ] ) , <EOL> [ c . get ( "<STR_LIT:name>" ) <EOL> for c in get_clients_test_tree ( ) . findall ( "<STR_LIT>" ) ] ) <EOL> @ patch ( "<STR_LIT>" , Mock ( ) ) <EOL> def test_get_client_names_by_groups ( self ) : <EOL> metadata = self . load_clients_data ( metadata = self . load_groups_data ( ) ) <EOL> metadata . core . build_metadata = Mock ( ) <EOL> metadata . core . build_metadata . side_effect = lambda c : metadata . get_initial_metadata ( c ) <EOL> self . assertItemsEqual ( metadata . get_client_names_by_groups ( [ "<STR_LIT>" ] ) , <EOL> [ c . get ( "<STR_LIT:name>" ) <EOL> for c in get_clients_test_tree ( ) . findall ( "<STR_LIT>" ) ] ) <EOL> @ patch ( "<STR_LIT>" , Mock ( ) ) <EOL> def test_merge_additional_groups ( self ) : <EOL> metadata = self . load_clients_data ( metadata = self . load_groups_data ( ) ) <EOL> imd = metadata . get_initial_metadata ( "<STR_LIT>" ) <EOL> oldgroups = imd . groups <EOL> metadata . merge_additional_groups ( imd , [ "<STR_LIT>" ] ) <EOL> self . assertEqual ( imd . groups , oldgroups ) <EOL> oldgroups = imd . groups <EOL> metadata . merge_additional_groups ( imd , [ "<STR_LIT>" ] ) <EOL> self . assertEqual ( imd . groups , oldgroups ) <EOL> oldgroups = imd . groups <EOL> oldbundles = imd . bundles <EOL> metadata . merge_additional_groups ( imd , [ "<STR_LIT>" ] ) <EOL> self . assertEqual ( imd . groups , oldgroups . union ( [ "<STR_LIT>" ] ) ) <EOL> self . assertEqual ( imd . bundles , oldbundles . union ( [ "<STR_LIT>" ] ) ) <EOL> oldgroups = imd . groups <EOL> metadata . merge_additional_groups ( imd , [ "<STR_LIT>" ] ) <EOL> self . assertEqual ( imd . groups , oldgroups . union ( [ "<STR_LIT>" ] ) ) <EOL> self . assertIn ( "<STR_LIT>" , imd . categories ) <EOL> self . assertEqual ( imd . categories [ "<STR_LIT>" ] , "<STR_LIT>" ) <EOL> imd = metadata . get_initial_metadata ( "<STR_LIT>" ) <EOL> oldgroups = imd . groups <EOL> metadata . merge_additional_groups ( imd , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> self . assertItemsEqual ( imd . groups , <EOL> oldgroups . union ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) ) <EOL> imd = metadata . get_initial_metadata ( "<STR_LIT>" ) <EOL> oldgroups = imd . groups <EOL> metadata . merge_additional_groups ( imd , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> self . assertItemsEqual ( imd . groups , <EOL> oldgroups . union ( [ "<STR_LIT>" , "<STR_LIT>" ] ) ) <EOL> @ patch ( "<STR_LIT>" , Mock ( ) ) <EOL> def test_merge_additional_data ( self ) : <EOL> metadata = self . load_clients_data ( metadata = self . load_groups_data ( ) ) <EOL> imd = metadata . get_initial_metadata ( "<STR_LIT>" ) <EOL> pattern = "<STR_LIT>" <EOL> for i in range ( <NUM_LIT:0> , <NUM_LIT:100> ) : <EOL> connector = pattern % i <EOL> if not hasattr ( imd , connector ) : <EOL> break <EOL> self . assertFalse ( hasattr ( imd , connector ) , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> metadata . merge_additional_data ( imd , connector , "<STR_LIT>" ) <EOL> self . assertEqual ( getattr ( imd , connector ) , "<STR_LIT>" ) <EOL> self . assertIn ( connector , imd . connectors ) <EOL> @ patch ( "<STR_LIT>" , Mock ( ) ) <EOL> @ patch ( "<STR_LIT>" ) <EOL> def test_validate_client_address ( self , mock_resolve_client ) : <EOL> metadata = self . load_clients_data ( metadata = self . load_groups_data ( ) ) <EOL> self . assertTrue ( metadata . validate_client_address ( "<STR_LIT>" , <EOL> ( None , None ) ) ) <EOL> self . assertTrue ( metadata . validate_client_address ( "<STR_LIT>" , <EOL> ( "<STR_LIT>" , None ) ) ) <EOL> self . assertFalse ( metadata . validate_client_address ( "<STR_LIT>" , <EOL> ( "<STR_LIT>" , None ) ) ) <EOL> self . assertTrue ( metadata . validate_client_address ( "<STR_LIT>" , <EOL> ( "<STR_LIT>" , None ) ) ) <EOL> mock_resolve_client . return_value = "<STR_LIT>" <EOL> self . assertTrue ( metadata . validate_client_address ( "<STR_LIT>" , <EOL> ( "<STR_LIT>" , None ) ) ) <EOL> mock_resolve_client . assert_called_with ( ( "<STR_LIT>" , None ) ) <EOL> mock_resolve_client . reset_mock ( ) <EOL> self . assertFalse ( metadata . validate_client_address ( "<STR_LIT>" , <EOL> ( "<STR_LIT>" , None ) ) ) <EOL> @ patch ( "<STR_LIT>" , Mock ( ) ) <EOL> @ patch ( "<STR_LIT>" ) <EOL> @ patch ( "<STR_LIT>" ) <EOL> def test_AuthenticateConnection ( self , mock_resolve_client , <EOL> mock_validate_client_address ) : <EOL> metadata = self . load_clients_data ( metadata = self . load_groups_data ( ) ) <EOL> metadata . password = "<STR_LIT>" <EOL> cert = dict ( subject = [ [ ( "<STR_LIT>" , "<STR_LIT>" ) ] ] ) <EOL> mock_validate_client_address . return_value = False <EOL> self . assertFalse ( metadata . AuthenticateConnection ( cert , "<STR_LIT:root>" , None , <EOL> "<STR_LIT>" ) ) <EOL> mock_validate_client_address . return_value = True <EOL> self . assertTrue ( metadata . AuthenticateConnection ( cert , "<STR_LIT:root>" , None , <EOL> "<STR_LIT>" ) ) <EOL> self . assertIn ( "<STR_LIT>" , metadata . session_cache ) <EOL> self . assertEqual ( metadata . session_cache [ "<STR_LIT>" ] [ <NUM_LIT:1> ] , "<STR_LIT>" ) <EOL> cert = dict ( subject = [ [ ( "<STR_LIT>" , "<STR_LIT>" ) ] ] ) <EOL> self . assertTrue ( metadata . AuthenticateConnection ( cert , "<STR_LIT:root>" , None , <EOL> "<STR_LIT>" ) ) <EOL> self . assertNotIn ( "<STR_LIT>" , metadata . session_cache ) <EOL> cert = dict ( subject = [ [ ( "<STR_LIT>" , "<STR_LIT>" ) ] ] ) <EOL> mock_resolve_client . return_value = "<STR_LIT>" <EOL> self . assertTrue ( metadata . AuthenticateConnection ( None , "<STR_LIT:root>" , <EOL> "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> mock_resolve_client . side_effect = Bcfg2 . Server . Plugin . MetadataConsistencyError <EOL> self . assertFalse ( metadata . AuthenticateConnection ( None , "<STR_LIT:root>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ) ) <EOL> self . assertFalse ( metadata . AuthenticateConnection ( None , '<STR_LIT>' , None , <EOL> "<STR_LIT>" ) ) <EOL> self . assertTrue ( metadata . AuthenticateConnection ( None , '<STR_LIT>' , <EOL> "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> self . assertIn ( "<STR_LIT>" , metadata . session_cache ) <EOL> self . assertEqual ( metadata . session_cache [ "<STR_LIT>" ] [ <NUM_LIT:1> ] , "<STR_LIT>" ) <EOL> self . assertTrue ( metadata . AuthenticateConnection ( None , '<STR_LIT>' , <EOL> "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> self . assertFalse ( metadata . AuthenticateConnection ( None , '<STR_LIT>' , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ) ) <EOL> self . assertTrue ( metadata . AuthenticateConnection ( None , '<STR_LIT>' , <EOL> "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> self . assertFalse ( metadata . AuthenticateConnection ( None , "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ) ) <EOL> @ patch ( "<STR_LIT>" , Mock ( ) ) <EOL> @ patch ( "<STR_LIT>" ) <EOL> def test_end_statistics ( self , mock_update_client ) : <EOL> metadata = self . load_clients_data ( metadata = self . load_groups_data ( ) ) <EOL> md = Mock ( ) <EOL> md . hostname = "<STR_LIT>" <EOL> metadata . end_statistics ( md ) <EOL> mock_update_client . assert_called_with ( md . hostname , dict ( auth = '<STR_LIT>' ) ) <EOL> mock_update_client . reset_mock ( ) <EOL> md . hostname = "<STR_LIT>" <EOL> metadata . end_statistics ( md ) <EOL> self . assertFalse ( mock_update_client . called ) <EOL> def test_viz ( self ) : <EOL> pass <EOL> class TestMetadataBase ( TestMetadata ) : <EOL> """<STR_LIT>""" <EOL> __test__ = False <EOL> @ skipUnless ( HAS_DJANGO , "<STR_LIT>" ) <EOL> def setUp ( self ) : <EOL> _TestMetadata . setUp ( self ) <EOL> TestClientRunHooks . setUp ( self ) <EOL> TestDatabaseBacked . setUp ( self ) <EOL> Bcfg2 . Options . setup . metadata_db = True <EOL> syncdb ( TestMetadataDB ) <EOL> def load_clients_data ( self , metadata = None , xdata = None ) : <EOL> if metadata is None : <EOL> metadata = get_obj ( ) <EOL> for client in get_clients_test_tree ( ) . findall ( "<STR_LIT>" ) : <EOL> metadata . add_client ( client . get ( "<STR_LIT:name>" ) ) <EOL> return metadata <EOL> def get_nonexistent_client ( self , _ , prefix = "<STR_LIT>" ) : <EOL> clients = [ o . hostname for o in MetadataClientModel . objects . all ( ) ] <EOL> i = <NUM_LIT:0> <EOL> client_name = "<STR_LIT>" % ( prefix , i ) <EOL> while client_name in clients : <EOL> i += <NUM_LIT:1> <EOL> client_name = "<STR_LIT>" % ( prefix , i ) <EOL> return client_name <EOL> @ patch ( '<STR_LIT>' ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test__init ( self , mock_get_fam , mock_exists ) : <EOL> mock_exists . return_value = False <EOL> metadata = self . get_obj ( ) <EOL> self . assertIsInstance ( metadata , Bcfg2 . Server . Plugin . DatabaseBacked ) <EOL> mock_get_fam . return_value . AddMonitor . assert_called_with ( <EOL> os . path . join ( metadata . data , "<STR_LIT>" ) , <EOL> metadata ) <EOL> mock_exists . return_value = True <EOL> mock_get_fam . reset_mock ( ) <EOL> metadata = self . get_obj ( ) <EOL> mock_get_fam . return_value . AddMonitor . assert_any_call ( <EOL> os . path . join ( metadata . data , "<STR_LIT>" ) , <EOL> metadata ) <EOL> mock_get_fam . return_value . AddMonitor . assert_any_call ( <EOL> os . path . join ( metadata . data , "<STR_LIT>" ) , <EOL> metadata ) <EOL> def test_add_group ( self ) : <EOL> pass <EOL> def test_add_bundle ( self ) : <EOL> pass <EOL> def test_add_client ( self ) : <EOL> metadata = self . get_obj ( ) <EOL> hostname = self . get_nonexistent_client ( metadata ) <EOL> client = metadata . add_client ( hostname ) <EOL> self . assertIsInstance ( client , MetadataClientModel ) <EOL> self . assertEqual ( client . hostname , hostname ) <EOL> self . assertIn ( hostname , metadata . clients ) <EOL> self . assertIn ( hostname , metadata . list_clients ( ) ) <EOL> self . assertItemsEqual ( metadata . clients , <EOL> [ c . hostname <EOL> for c in MetadataClientModel . objects . all ( ) ] ) <EOL> def test_update_group ( self ) : <EOL> pass <EOL> def test_update_bundle ( self ) : <EOL> pass <EOL> def test_update_client ( self ) : <EOL> pass <EOL> def test_list_clients ( self ) : <EOL> metadata = self . get_obj ( ) <EOL> self . assertItemsEqual ( metadata . list_clients ( ) , <EOL> [ c . hostname <EOL> for c in MetadataClientModel . objects . all ( ) ] ) <EOL> def test_remove_group ( self ) : <EOL> pass <EOL> def test_remove_bundle ( self ) : <EOL> pass <EOL> def test_remove_client ( self ) : <EOL> metadata = self . get_obj ( ) <EOL> client_name = self . get_nonexistent_client ( metadata ) <EOL> self . assertRaises ( Bcfg2 . Server . Plugin . MetadataConsistencyError , <EOL> metadata . remove_client , <EOL> client_name ) <EOL> metadata . add_client ( client_name ) <EOL> metadata . remove_client ( client_name ) <EOL> self . assertNotIn ( client_name , metadata . clients ) <EOL> self . assertNotIn ( client_name , metadata . list_clients ( ) ) <EOL> self . assertItemsEqual ( metadata . clients , <EOL> [ c . hostname <EOL> for c in MetadataClientModel . objects . all ( ) ] ) <EOL> def test_process_statistics ( self ) : <EOL> pass <EOL> class TestMetadata_NoClientsXML ( TestMetadataBase ) : <EOL> """<STR_LIT>""" <EOL> __test__ = True <EOL> def load_groups_data ( self , metadata = None , xdata = None ) : <EOL> if metadata is None : <EOL> metadata = self . get_obj ( ) <EOL> if not xdata : <EOL> xdata = copy . deepcopy ( get_groups_test_tree ( ) ) <EOL> for client in get_clients_test_tree ( ) . findall ( "<STR_LIT>" ) : <EOL> newclient = lxml . etree . SubElement ( xdata . getroot ( ) , <EOL> "<STR_LIT>" , name = client . get ( "<STR_LIT:name>" ) ) <EOL> lxml . etree . SubElement ( newclient , "<STR_LIT>" , <EOL> name = client . get ( "<STR_LIT>" ) ) <EOL> metadata . groups_xml . data = xdata <EOL> metadata . groups_xml . basedata = copy . copy ( metadata . groups_xml . data ) <EOL> evt = Mock ( ) <EOL> evt . filename = os . path . join ( datastore , "<STR_LIT>" , "<STR_LIT>" ) <EOL> evt . code2str = Mock ( return_value = "<STR_LIT>" ) <EOL> metadata . HandleEvent ( evt ) <EOL> return metadata <EOL> @ patch ( "<STR_LIT>" , Mock ( ) ) <EOL> @ patch ( "<STR_LIT>" , Mock ( ) ) <EOL> @ patch ( "<STR_LIT>" ) <EOL> def test_get_initial_metadata ( self , mock_clientmetadata ) : <EOL> metadata = self . get_obj ( ) <EOL> if '<STR_LIT>' in metadata . states : <EOL> metadata . states [ '<STR_LIT>' ] = False <EOL> self . assertRaises ( Bcfg2 . Server . Plugin . MetadataRuntimeError , <EOL> metadata . get_initial_metadata , None ) <EOL> self . load_groups_data ( metadata = metadata ) <EOL> self . load_clients_data ( metadata = metadata ) <EOL> metadata . get_initial_metadata ( "<STR_LIT>" ) <EOL> mock_clientmetadata . assert_called_with ( "<STR_LIT>" , "<STR_LIT>" , <EOL> set ( [ "<STR_LIT>" ] ) , set ( ) , set ( ) , <EOL> set ( ) , dict ( category1 = '<STR_LIT>' ) , <EOL> None , None , None , metadata . query ) <EOL> metadata . get_initial_metadata ( "<STR_LIT>" ) <EOL> mock_clientmetadata . assert_called_with ( "<STR_LIT>" , "<STR_LIT>" , <EOL> set ( [ "<STR_LIT>" ] ) , <EOL> set ( [ "<STR_LIT>" , "<STR_LIT>" ] ) , <EOL> set ( ) , set ( ) , <EOL> dict ( category1 = "<STR_LIT>" ) , None , <EOL> None , None , metadata . query ) <EOL> new1 = self . get_nonexistent_client ( metadata ) <EOL> imd = metadata . get_initial_metadata ( new1 ) <EOL> mock_clientmetadata . assert_called_with ( new1 , "<STR_LIT>" , set ( [ "<STR_LIT>" ] ) , <EOL> set ( ) , set ( ) , set ( ) , <EOL> dict ( category1 = "<STR_LIT>" ) , None , <EOL> None , None , metadata . query ) <EOL> imd = metadata . get_initial_metadata ( "<STR_LIT>" ) <EOL> mock_clientmetadata . assert_called_with ( "<STR_LIT>" , "<STR_LIT>" , <EOL> set ( [ "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" ] ) , <EOL> set ( ) , set ( ) , set ( ) , <EOL> dict ( category1 = "<STR_LIT>" ) , None , <EOL> None , None , metadata . query ) <EOL> imd = metadata . get_initial_metadata ( "<STR_LIT>" ) <EOL> mock_clientmetadata . assert_called_with ( "<STR_LIT>" , "<STR_LIT>" , <EOL> set ( [ "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" ] ) , <EOL> set ( [ "<STR_LIT>" , "<STR_LIT>" ] ) , <EOL> set ( ) , set ( ) , <EOL> dict ( category1 = "<STR_LIT>" ) , None , <EOL> None , None , metadata . query ) <EOL> metadata . default = None <EOL> new2 = self . get_nonexistent_client ( metadata ) <EOL> self . assertRaises ( Bcfg2 . Server . Plugin . MetadataConsistencyError , <EOL> metadata . get_initial_metadata , <EOL> new2 ) <EOL> @ patch ( "<STR_LIT>" , Mock ( ) ) <EOL> @ patch ( "<STR_LIT>" ) <EOL> def test_validate_client_address ( self , mock_resolve_client ) : <EOL> metadata = self . load_clients_data ( metadata = self . load_groups_data ( ) ) <EOL> mock_resolve_client . return_value = "<STR_LIT>" <EOL> self . assertTrue ( metadata . validate_client_address ( "<STR_LIT>" , <EOL> ( "<STR_LIT>" , None ) ) ) <EOL> mock_resolve_client . assert_called_with ( ( "<STR_LIT>" , None ) ) <EOL> mock_resolve_client . reset_mock ( ) <EOL> self . assertFalse ( metadata . validate_client_address ( "<STR_LIT>" , <EOL> ( "<STR_LIT>" , None ) ) ) <EOL> @ patch ( "<STR_LIT>" , Mock ( ) ) <EOL> @ patch ( "<STR_LIT>" ) <EOL> @ patch ( "<STR_LIT>" ) <EOL> def test_AuthenticateConnection ( self , mock_resolve_client , <EOL> mock_validate_client_address ) : <EOL> metadata = self . load_clients_data ( metadata = self . load_groups_data ( ) ) <EOL> metadata . password = "<STR_LIT>" <EOL> cert = dict ( subject = [ [ ( "<STR_LIT>" , "<STR_LIT>" ) ] ] ) <EOL> mock_validate_client_address . return_value = False <EOL> self . assertFalse ( metadata . AuthenticateConnection ( cert , "<STR_LIT:root>" , None , <EOL> "<STR_LIT>" ) ) <EOL> mock_validate_client_address . return_value = True <EOL> self . assertTrue ( metadata . AuthenticateConnection ( cert , "<STR_LIT:root>" , <EOL> metadata . password , <EOL> "<STR_LIT>" ) ) <EOL> cert = dict ( subject = [ [ ( "<STR_LIT>" , "<STR_LIT>" ) ] ] ) <EOL> mock_resolve_client . return_value = "<STR_LIT>" <EOL> self . assertTrue ( metadata . AuthenticateConnection ( None , "<STR_LIT:root>" , <EOL> "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> mock_resolve_client . side_effect = Bcfg2 . Server . Plugin . MetadataConsistencyError <EOL> self . assertFalse ( metadata . AuthenticateConnection ( None , "<STR_LIT:root>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ) ) <EOL> @ patch ( "<STR_LIT>" , Mock ( ) ) <EOL> @ patch ( "<STR_LIT>" ) <EOL> def test_resolve_client ( self , mock_getnameinfo ) : <EOL> metadata = self . load_clients_data ( metadata = self . load_groups_data ( ) ) <EOL> metadata . session_cache [ ( '<STR_LIT>' , None ) ] = ( time . time ( ) , '<STR_LIT>' ) <EOL> self . assertEqual ( metadata . resolve_client ( ( '<STR_LIT>' , None ) ) , '<STR_LIT>' ) <EOL> metadata . session_cache [ ( '<STR_LIT>' , None ) ] = ( time . time ( ) - <NUM_LIT:100> , <EOL> '<STR_LIT>' ) <EOL> mock_getnameinfo . return_value = ( "<STR_LIT>" , [ ] , [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( metadata . resolve_client ( ( '<STR_LIT>' , None ) , <EOL> cleanup_cache = True ) , '<STR_LIT>' ) <EOL> self . assertEqual ( metadata . session_cache , dict ( ) ) <EOL> mock_getnameinfo . return_value = ( '<STR_LIT>' , [ ] , [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( metadata . resolve_client ( ( '<STR_LIT>' , <NUM_LIT> ) , socket . NI_NAMEREQD ) , '<STR_LIT>' ) <EOL> mock_getnameinfo . assert_called_with ( ( '<STR_LIT>' , <NUM_LIT> ) , socket . NI_NAMEREQD ) <EOL> mock_getnameinfo . reset_mock ( ) <EOL> mock_getnameinfo . return_value = None <EOL> mock_getnameinfo . side_effect = socket . herror <EOL> self . assertRaises ( Bcfg2 . Server . Plugin . MetadataConsistencyError , <EOL> metadata . resolve_client , <EOL> ( '<STR_LIT>' , <NUM_LIT> ) , socket . NI_NAMEREQD ) <EOL> mock_getnameinfo . assert_called_with ( ( '<STR_LIT>' , <NUM_LIT> ) , socket . NI_NAMEREQD ) <EOL> def test_handle_clients_xml_event ( self ) : <EOL> pass <EOL> def test_end_statistics ( self ) : <EOL> pass <EOL> class TestMetadata_ClientsXML ( TestMetadataBase ) : <EOL> """<STR_LIT>""" <EOL> __test__ = True <EOL> def load_clients_data ( self , metadata = None , xdata = None ) : <EOL> if metadata is None : <EOL> metadata = self . get_obj ( ) <EOL> fam = Bcfg2 . Server . FileMonitor . _FAM <EOL> Bcfg2 . Server . FileMonitor . _FAM = MagicMock ( ) <EOL> @ patchIf ( not isinstance ( lxml . etree . Element , Mock ) , <EOL> "<STR_LIT>" , Mock ( ) ) <EOL> def inner ( ) : <EOL> metadata . clients_xml = metadata . _handle_file ( "<STR_LIT>" ) <EOL> inner ( ) <EOL> metadata = TestMetadata . load_clients_data ( self , metadata = metadata , <EOL> xdata = xdata ) <EOL> rv = TestMetadataBase . load_clients_data ( self , metadata = metadata , <EOL> xdata = xdata ) <EOL> Bcfg2 . Server . FileMonitor . _FAM = fam <EOL> return rv </s>
