#NEWLINE# @ classmethod #NEWLINE# def _parse_test( cls , buf ) : #NEWLINE# #INDENT# def __test_pkt_from_json( test ) : #NEWLINE# #INDENT# data = eval ( '/' . join ( test ) ) #NEWLINE# data . serialize ( ) #NEWLINE# return six . binary_type ( data . data ) #NEWLINE# #UNINDENT# target_dp = DummyDatapath ( OfTester . target_ver ) #NEWLINE# tester_dp = DummyDatapath ( OfTester . tester_ver ) #NEWLINE# description = buf . get ( KEY_DESC ) #NEWLINE# prerequisite = [ ] #NEWLINE# if ( KEY_PREREQ not in buf ) : #NEWLINE# #INDENT# raise ValueError ( ( 'a test requires a "%s" block' % KEY_PREREQ ) ) #NEWLINE# #UNINDENT# for flow in buf [ KEY_PREREQ ] : #NEWLINE# #INDENT# msg = ofproto_parser . ofp_msg_from_jsondict ( target_dp , flow ) #NEWLINE# msg . serialize ( ) #NEWLINE# prerequisite . append ( msg ) #NEWLINE# #UNINDENT# tests = [ ] #NEWLINE# if ( KEY_TESTS not in buf ) : #NEWLINE# #INDENT# raise ValueError ( ( 'a test requires a "%s" block.' % KEY_TESTS ) ) #NEWLINE# #UNINDENT# for test in buf [ KEY_TESTS ] : #NEWLINE# #INDENT# if ( len ( test ) != 2 ) : #NEWLINE# #INDENT# raise ValueError ( ( '"%s" block requires "%s" field and one of "%s" or "%s" or "%s" field.' % ( KEY_TESTS , KEY_INGRESS , KEY_EGRESS , KEY_PKT_IN , KEY_TBL_MISS ) ) ) #NEWLINE# #UNINDENT# test_pkt = { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } #NEWLINE# if ( KEY_INGRESS not in test ) : #NEWLINE# #INDENT# raise ValueError ( ( 'a test requires "%s" field.' % KEY_INGRESS ) ) #NEWLINE# #UNINDENT# if isinstance ( test [ KEY_INGRESS ] , list ) : #NEWLINE# #INDENT# test_pkt [ KEY_INGRESS ] = __test_pkt_from_json ( test [ KEY_INGRESS ] ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if isinstance ( test [ KEY_INGRESS ] , dict ) : #NEWLINE# #INDENT# test_pkt [ KEY_PACKETS ] = { #NEWLINE# #INDENT# 'packet_text' : test [ KEY_INGRESS ] [ KEY_PACKETS ] [ KEY_DATA ] , #NEWLINE# 'packet_binary' : __test_pkt_from_json ( test [ KEY_INGRESS ] [ KEY_PACKETS ] [ KEY_DATA ] ) , #NEWLINE# KEY_DURATION_TIME : test [ KEY_INGRESS ] [ KEY_PACKETS ] . get ( KEY_DURATION_TIME , DEFAULT_DURATION_TIME ) , #NEWLINE# KEY_PKTPS : test [ KEY_INGRESS ] [ KEY_PACKETS ] . get ( KEY_PKTPS , DEFAULT_PKTPS ) , #NEWLINE# 'randomize' : ( True in [ ( line . find ( 'randint' ) != ( - 1 ) ) for line in test [ KEY_INGRESS ] [ KEY_PACKETS ] [ KEY_DATA ] ] ) , #NEWLINE# #UNINDENT# } #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# raise ValueError ( ( 'invalid format: "%s" field' % KEY_INGRESS ) ) #NEWLINE# #UNINDENT# #UNINDENT# if ( KEY_EGRESS in test ) : #NEWLINE# #INDENT# if isinstance ( test [ KEY_EGRESS ] , list ) : #NEWLINE# #INDENT# test_pkt [ KEY_EGRESS ] = __test_pkt_from_json ( test [ KEY_EGRESS ] ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if isinstance ( test [ KEY_EGRESS ] , dict ) : #NEWLINE# #INDENT# throughputs = [ ] #NEWLINE# for throughput in test [ KEY_EGRESS ] [ KEY_THROUGHPUT ] : #NEWLINE# #INDENT# one = { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } #NEWLINE# mod = { #NEWLINE# #INDENT# 'OFPFlowMod' : { #NEWLINE# #INDENT# 'cookie' : THROUGHPUT_COOKIE , #NEWLINE# 'priority' : THROUGHPUT_PRIORITY , #NEWLINE# 'match' : { #NEWLINE# #INDENT# 'OFPMatch' : target_dp [ KEY_MATCH ] , #NEWLINE# #UNINDENT# } , #NEWLINE# #UNINDENT# } , #NEWLINE# #UNINDENT# } #NEWLINE# msg = ofproto_parser . ofp_msg_from_jsondict ( tester_dp , mod ) #NEWLINE# one [ KEY_FLOW ] = msg #NEWLINE# one [ KEY_KBPS ] = throughput . get ( KEY_KBPS ) #NEWLINE# one [ KEY_PKTPS ] = throughput . get ( KEY_PKTPS ) #NEWLINE# if ( not ( bool ( one [ KEY_KBPS ] ) != bool ( one [ KEY_PKTPS ] ) ) ) : #NEWLINE# #INDENT# raise ValueError ( ( '"%s" requires either "%s" or "%s".' % ( KEY_THROUGHPUT , KEY_KBPS , KEY_PKTPS ) ) ) #NEWLINE# #UNINDENT# throughputs . append ( one ) #NEWLINE# #UNINDENT# test_pkt [ KEY_THROUGHPUT ] = throughputs #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# raise ValueError ( ( 'invalid format: "%s" field' % KEY_EGRESS ) ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# if ( KEY_PKT_IN in test ) : #NEWLINE# #INDENT# test_pkt [ KEY_PKT_IN ] = __test_pkt_from_json ( test [ KEY_PKT_IN ] ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if ( KEY_TBL_MISS in test ) : #NEWLINE# #INDENT# test_pkt [ KEY_TBL_MISS ] = test [ KEY_TBL_MISS ] #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# tests . append ( test_pkt ) #NEWLINE# #UNINDENT# return ( description , prerequisite , tests )