<s> import os <EOL> import unittest <EOL> import traceback <EOL> import yaml <EOL> from ztpserver . app import enable_handler_console <EOL> from ztpserver . topology import load_neighbordb , load_file <EOL> from ztpserver . topology import Node <EOL> from ztpserver . validators import NeighbordbValidator <EOL> from ztpserver . constants import CONTENT_TYPE_YAML <EOL> from server_test_lib import enable_logging , log , random_string <EOL> TEST_DIR = '<STR_LIT>' <EOL> ID = random_string ( ) <EOL> def debug ( exc ) : <EOL> raise exc <EOL> def load_node ( node , content_type = CONTENT_TYPE_YAML ) : <EOL> try : <EOL> if not hasattr ( node , '<STR_LIT>' ) : <EOL> node = load_file ( node , content_type , ID ) <EOL> if '<STR_LIT>' in node : <EOL> for symbol in [ '<STR_LIT::>' , '<STR_LIT:.>' ] : <EOL> node [ '<STR_LIT>' ] = str ( node [ '<STR_LIT>' ] ) . replace ( symbol , '<STR_LIT>' ) <EOL> return Node ( ** node ) <EOL> except TypeError : <EOL> log . error ( '<STR_LIT>' ) <EOL> except KeyError as err : <EOL> log . error ( '<STR_LIT>' % err ) <EOL> class NeighbordbTest ( unittest . TestCase ) : <EOL> def __init__ ( self , test_name , ndb , ** kwargs ) : <EOL> self . ndb = ndb <EOL> self . node = kwargs . get ( '<STR_LIT>' ) <EOL> self . tag = kwargs . get ( '<STR_LIT>' ) <EOL> self . match = kwargs . get ( '<STR_LIT>' ) <EOL> self . filename = kwargs . get ( '<STR_LIT:filename>' ) <EOL> self . valid_patterns = kwargs . get ( '<STR_LIT>' , dict ( ) ) <EOL> if not self . valid_patterns . get ( '<STR_LIT>' ) : <EOL> self . valid_patterns [ '<STR_LIT>' ] = list ( ) <EOL> if not self . valid_patterns . get ( '<STR_LIT>' ) : <EOL> self . valid_patterns [ '<STR_LIT>' ] = list ( ) <EOL> self . invalid_patterns = kwargs . get ( '<STR_LIT>' ) <EOL> self . longMessage = True <EOL> self . maxDiff = None <EOL> super ( NeighbordbTest , self ) . __init__ ( test_name ) <EOL> def _load_neighbordb ( self ) : <EOL> return load_neighbordb ( ID , contents = self . ndb ) <EOL> def _validate_neighbordb ( self ) : <EOL> validator = NeighbordbValidator ( ID ) <EOL> validator . validate ( self . ndb ) <EOL> return ( validator . valid_patterns , validator . invalid_patterns ) <EOL> def neighbordb_pattern ( self ) : <EOL> try : <EOL> log . info ( '<STR_LIT>' ) <EOL> tag = '<STR_LIT>' % ( self . tag , self . filename ) <EOL> ( valid , failed ) = self . _validate_neighbordb ( ) <EOL> if self . invalid_patterns : <EOL> failed_names = sorted ( [ p [ <NUM_LIT:1> ] for p in failed ] ) <EOL> self . assertEqual ( failed_names , <EOL> sorted ( self . invalid_patterns ) , <EOL> tag ) <EOL> self . assertEqual ( len ( failed ) , len ( self . invalid_patterns ) , tag ) <EOL> p_all = self . valid_patterns . get ( '<STR_LIT>' ) + self . valid_patterns . get ( '<STR_LIT>' ) <EOL> if p_all : <EOL> valid_actual = sorted ( [ str ( p [ <NUM_LIT:1> ] ) for p in valid ] ) <EOL> valid_configured = sorted ( [ str ( x ) for x in p_all ] ) <EOL> self . assertEqual ( valid_actual , valid_configured , tag ) <EOL> log . info ( '<STR_LIT>' ) <EOL> except AssertionError as exc : <EOL> print exc <EOL> print traceback . format_exc ( ) <EOL> debug ( exc ) <EOL> def neighbordb_topology ( self ) : <EOL> try : <EOL> log . info ( '<STR_LIT>' ) <EOL> tag = '<STR_LIT>' % ( self . tag , self . filename ) <EOL> topo = self . _load_neighbordb ( ) <EOL> self . assertIsNotNone ( topo , tag ) <EOL> p_nodes = [ p . name for p in topo . get_patterns ( ) <EOL> if topo . is_node_pattern ( p ) ] <EOL> p_globals = [ p . name for p in topo . get_patterns ( ) <EOL> if topo . is_global_pattern ( p ) ] <EOL> self . assertEqual ( sorted ( self . valid_patterns [ '<STR_LIT>' ] ) , <EOL> sorted ( p_nodes ) , <EOL> tag ) <EOL> self . assertEqual ( sorted ( self . valid_patterns [ '<STR_LIT>' ] ) , <EOL> sorted ( p_globals ) , <EOL> tag ) <EOL> log . info ( '<STR_LIT>' ) <EOL> return topo <EOL> except AssertionError as exc : <EOL> print exc <EOL> print traceback . format_exc ( ) <EOL> debug ( exc ) <EOL> def node_pass ( self ) : <EOL> try : <EOL> log . info ( '<STR_LIT>' ) <EOL> tag = '<STR_LIT>' % ( self . tag , self . filename ) <EOL> node = load_node ( self . node ) <EOL> self . assertIsNotNone ( node , tag ) <EOL> neighbordb = self . _load_neighbordb ( ) <EOL> self . assertIsNotNone ( neighbordb , tag ) <EOL> result = neighbordb . match_node ( node ) <EOL> self . assertTrue ( result , tag ) <EOL> self . assertEqual ( result [ <NUM_LIT:0> ] . name , self . match , tag ) <EOL> log . info ( '<STR_LIT>' ) <EOL> except AssertionError as exc : <EOL> print exc <EOL> print traceback . format_exc ( ) <EOL> debug ( exc ) <EOL> def node_fail ( self ) : <EOL> try : <EOL> log . info ( '<STR_LIT>' ) <EOL> tag = '<STR_LIT>' % ( self . tag , self . filename ) <EOL> node = load_node ( self . node ) <EOL> self . assertIsNotNone ( node , tag ) <EOL> neighbordb = self . _load_neighbordb ( ) <EOL> self . assertIsNotNone ( neighbordb , tag ) <EOL> result = neighbordb . match_node ( node ) <EOL> self . assertFalse ( result , tag ) <EOL> log . info ( '<STR_LIT>' ) <EOL> except AssertionError as exc : <EOL> print exc <EOL> print traceback . format_exc ( ) <EOL> debug ( exc ) <EOL> def get_test_list ( filepath ) : <EOL> test_list = os . environ . get ( '<STR_LIT>' , None ) <EOL> if not test_list : <EOL> log . debug ( '<STR_LIT>' , TEST_DIR ) <EOL> test_list = [ os . path . join ( filepath , f ) for f in os . listdir ( filepath ) <EOL> if f . endswith ( '<STR_LIT>' ) ] <EOL> else : <EOL> test_list = [ os . path . join ( filepath , f ) for f in test_list . split ( '<STR_LIT:U+002C>' ) ] <EOL> return test_list <EOL> def load_tests ( loader , tests , pattern ) : <EOL> try : <EOL> log . info ( '<STR_LIT>' ) <EOL> suite = unittest . TestSuite ( ) <EOL> test_list = get_test_list ( TEST_DIR ) <EOL> log . info ( '<STR_LIT>' , test_list ) <EOL> for test in test_list : <EOL> log . info ( '<STR_LIT>' , test ) <EOL> harness = yaml . load ( open ( test ) ) <EOL> if harness . get ( '<STR_LIT>' ) : <EOL> enable_handler_console ( ) <EOL> tests = harness . get ( '<STR_LIT>' , [ '<STR_LIT>' ] ) <EOL> name = harness . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> filename = name . split ( '<STR_LIT:/>' ) [ - <NUM_LIT:1> ] <EOL> if filename in harness : <EOL> ndb = harness . get ( filename ) <EOL> else : <EOL> assert os . path . exists ( name ) <EOL> ndb = yaml . load ( open ( name ) ) <EOL> kwargs = dict ( ) <EOL> kwargs [ '<STR_LIT>' ] = harness . get ( '<STR_LIT>' , dict ( ) ) <EOL> kwargs [ '<STR_LIT>' ] = harness . get ( '<STR_LIT>' ) <EOL> kwargs [ '<STR_LIT>' ] = harness . get ( '<STR_LIT>' , filename ) <EOL> kwargs [ '<STR_LIT:filename>' ] = test <EOL> for test in tests : <EOL> assert test in [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> suite . addTest ( NeighbordbTest ( '<STR_LIT>' % test , <EOL> ndb , ** kwargs ) ) <EOL> if '<STR_LIT>' in harness and '<STR_LIT>' in tests : <EOL> for key in harness [ '<STR_LIT>' ] . keys ( ) : <EOL> assert key in [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> entries = harness [ '<STR_LIT>' ] [ key ] <EOL> if not entries is None : <EOL> for entry in entries : <EOL> filename = entry [ '<STR_LIT:name>' ] . split ( '<STR_LIT:/>' ) [ - <NUM_LIT:1> ] <EOL> kwargs [ '<STR_LIT>' ] = harness . get ( filename , <EOL> entry [ '<STR_LIT:name>' ] ) <EOL> kwargs [ '<STR_LIT>' ] = entry . get ( '<STR_LIT>' ) <EOL> kwargs [ '<STR_LIT>' ] = '<STR_LIT>' % ( harness . get ( '<STR_LIT>' ) , <EOL> filename ) <EOL> log . info ( '<STR_LIT>' , name ) <EOL> suite . addTest ( NeighbordbTest ( '<STR_LIT>' % key , <EOL> ndb , ** kwargs ) ) <EOL> except Exception as exc : <EOL> log . exception ( '<STR_LIT>' % <EOL> exc ) <EOL> else : <EOL> return suite <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> enable_logging ( ) <EOL> unittest . main ( ) </s>
