<s> from __future__ import print_function <EOL> import pytest <EOL> import random <EOL> import struct <EOL> from pymtl import * <EOL> from pclib . test import mk_test_case_table , run_sim <EOL> from pclib . test import TestSource , TestSink <EOL> from pclib . ifcs import MemMsg , MemReqMsg , MemRespMsg <EOL> from TestMemory import TestMemory <EOL> from pclib . ifcs import MemMsg4B , MemReqMsg4B , MemRespMsg4B <EOL> class TestHarness ( Model ) : <EOL> def __init__ ( s , nports , src_msgs , sink_msgs , stall_prob , latency , <EOL> src_delay , sink_delay ) : <EOL> mem_msgs = MemMsg4B ( ) <EOL> s . srcs = [ ] <EOL> for i in range ( nports ) : <EOL> s . srcs . append ( TestSource ( mem_msgs . req , src_msgs [ i ] , src_delay ) ) <EOL> s . mem = TestMemory ( mem_msgs , nports , stall_prob , latency ) <EOL> s . sinks = [ ] <EOL> for i in range ( nports ) : <EOL> s . sinks . append ( TestSink ( mem_msgs . resp , sink_msgs [ i ] , sink_delay ) ) <EOL> for i in range ( nports ) : <EOL> s . connect ( s . srcs [ i ] . out , s . mem . reqs [ i ] ) <EOL> s . connect ( s . sinks [ i ] . in_ , s . mem . resps [ i ] ) <EOL> def done ( s ) : <EOL> done_flag = <NUM_LIT:1> <EOL> for src , sink in zip ( s . srcs , s . sinks ) : <EOL> done_flag &= src . done and sink . done <EOL> return done_flag <EOL> def line_trace ( s ) : <EOL> return s . srcs [ <NUM_LIT:0> ] . line_trace ( ) + "<STR_LIT:U+0020>" + s . mem . line_trace ( ) + "<STR_LIT:U+0020>" + s . sinks [ <NUM_LIT:0> ] . line_trace ( ) <EOL> def req ( type_ , opaque , addr , len , data ) : <EOL> msg = MemReqMsg4B ( ) <EOL> if type_ == '<STR_LIT>' : msg . type_ = MemReqMsg . TYPE_READ <EOL> elif type_ == '<STR_LIT>' : msg . type_ = MemReqMsg . TYPE_WRITE <EOL> elif type_ == '<STR_LIT>' : msg . type_ = MemReqMsg . TYPE_AMO_ADD <EOL> elif type_ == '<STR_LIT>' : msg . type_ = MemReqMsg . TYPE_AMO_AND <EOL> elif type_ == '<STR_LIT>' : msg . type_ = MemReqMsg . TYPE_AMO_OR <EOL> elif type_ == '<STR_LIT>' : msg . type_ = MemReqMsg . TYPE_AMO_XCHG <EOL> elif type_ == '<STR_LIT>' : msg . type_ = MemReqMsg . TYPE_AMO_MIN <EOL> msg . addr = addr <EOL> msg . opaque = opaque <EOL> msg . len = len <EOL> msg . data = data <EOL> return msg <EOL> def resp ( type_ , opaque , len , data ) : <EOL> msg = MemRespMsg4B ( ) <EOL> if type_ == '<STR_LIT>' : msg . type_ = MemRespMsg . TYPE_READ <EOL> elif type_ == '<STR_LIT>' : msg . type_ = MemRespMsg . TYPE_WRITE <EOL> elif type_ == '<STR_LIT>' : msg . type_ = MemRespMsg . TYPE_AMO_ADD <EOL> elif type_ == '<STR_LIT>' : msg . type_ = MemRespMsg . TYPE_AMO_AND <EOL> elif type_ == '<STR_LIT>' : msg . type_ = MemRespMsg . TYPE_AMO_OR <EOL> elif type_ == '<STR_LIT>' : msg . type_ = MemRespMsg . TYPE_AMO_XCHG <EOL> elif type_ == '<STR_LIT>' : msg . type_ = MemRespMsg . TYPE_AMO_MIN <EOL> msg . opaque = opaque <EOL> msg . len = len <EOL> msg . data = data <EOL> return msg <EOL> def basic_msgs ( base_addr ) : <EOL> return [ <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr , <NUM_LIT:0> , <NUM_LIT> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr , <NUM_LIT:0> , <NUM_LIT:0> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT> ) , <EOL> ] <EOL> def stream_msgs ( base_addr ) : <EOL> msgs = [ ] <EOL> for i in range ( <NUM_LIT:20> ) : <EOL> msgs . extend ( [ <EOL> req ( '<STR_LIT>' , i , base_addr + <NUM_LIT:4> * i , <NUM_LIT:0> , i ) , resp ( '<STR_LIT>' , i , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> req ( '<STR_LIT>' , i , base_addr + <NUM_LIT:4> * i , <NUM_LIT:0> , <NUM_LIT:0> ) , resp ( '<STR_LIT>' , i , <NUM_LIT:0> , i ) , <EOL> ] ) <EOL> return msgs <EOL> def subword_rd_msgs ( base_addr ) : <EOL> return [ <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:0> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:0> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:2> , <NUM_LIT:1> , <NUM_LIT:0> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:3> , <NUM_LIT:1> , <NUM_LIT:0> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:0> , <NUM_LIT:2> , <NUM_LIT:0> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:2> , <NUM_LIT> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:0> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:2> , <NUM_LIT> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:2> , <NUM_LIT:2> , <NUM_LIT:0> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:2> , <NUM_LIT> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:0> , <NUM_LIT:3> , <NUM_LIT:0> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:3> , <NUM_LIT> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:1> , <NUM_LIT:3> , <NUM_LIT:0> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:3> , <NUM_LIT> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT> ) , <EOL> ] <EOL> def subword_wr_msgs ( base_addr ) : <EOL> return [ <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:2> , <NUM_LIT:1> , <NUM_LIT> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:3> , <NUM_LIT:1> , <NUM_LIT> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:0> , <NUM_LIT:2> , <NUM_LIT> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:2> , <NUM_LIT:2> , <NUM_LIT> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:0> , <NUM_LIT:3> , <NUM_LIT> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT> ) , <EOL> ] <EOL> def amo_msgs ( base_addr ) : <EOL> return [ <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr , <NUM_LIT:0> , <NUM_LIT> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:4> , <NUM_LIT:0> , <NUM_LIT> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:8> , <NUM_LIT:0> , <NUM_LIT> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:12> , <NUM_LIT:0> , <NUM_LIT> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:16> , <NUM_LIT:0> , <NUM_LIT> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:20> , <NUM_LIT:0> , <NUM_LIT> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr , <NUM_LIT:0> , <NUM_LIT> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr , <NUM_LIT:0> , <NUM_LIT:0> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:4> , <NUM_LIT:0> , <NUM_LIT> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:4> , <NUM_LIT:0> , <NUM_LIT:0> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:8> , <NUM_LIT:0> , <NUM_LIT> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:8> , <NUM_LIT:0> , <NUM_LIT:0> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:12> , <NUM_LIT:0> , <NUM_LIT> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:12> , <NUM_LIT:0> , <NUM_LIT:0> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:16> , <NUM_LIT:0> , <NUM_LIT> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:16> , <NUM_LIT:0> , <NUM_LIT:0> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:20> , <NUM_LIT:0> , <NUM_LIT> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT> ) , <EOL> req ( '<STR_LIT>' , <NUM_LIT> , base_addr + <NUM_LIT:20> , <NUM_LIT:0> , <NUM_LIT:0> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT> ) , <EOL> ] <EOL> def random_msgs ( base_addr ) : <EOL> rgen = random . Random ( ) <EOL> rgen . seed ( <NUM_LIT> ) <EOL> vmem = [ rgen . randint ( <NUM_LIT:0> , <NUM_LIT> ) for _ in range ( <NUM_LIT:20> ) ] <EOL> msgs = [ ] <EOL> for i in range ( <NUM_LIT:20> ) : <EOL> msgs . extend ( [ <EOL> req ( '<STR_LIT>' , i , base_addr + <NUM_LIT:4> * i , <NUM_LIT:0> , vmem [ i ] ) , resp ( '<STR_LIT>' , i , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> ] ) <EOL> for i in range ( <NUM_LIT:20> ) : <EOL> idx = rgen . randint ( <NUM_LIT:0> , <NUM_LIT> ) <EOL> if rgen . randint ( <NUM_LIT:0> , <NUM_LIT:1> ) : <EOL> correct_data = vmem [ idx ] <EOL> msgs . extend ( [ <EOL> req ( '<STR_LIT>' , i , base_addr + <NUM_LIT:4> * idx , <NUM_LIT:0> , <NUM_LIT:0> ) , resp ( '<STR_LIT>' , i , <NUM_LIT:0> , correct_data ) , <EOL> ] ) <EOL> else : <EOL> new_data = rgen . randint ( <NUM_LIT:0> , <NUM_LIT> ) <EOL> vmem [ idx ] = new_data <EOL> msgs . extend ( [ <EOL> req ( '<STR_LIT>' , i , base_addr + <NUM_LIT:4> * idx , <NUM_LIT:0> , new_data ) , resp ( '<STR_LIT>' , i , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> ] ) <EOL> return msgs <EOL> test_case_table = mk_test_case_table ( [ <EOL> ( "<STR_LIT>" ) , <EOL> [ "<STR_LIT>" , basic_msgs , <NUM_LIT:0.0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ] , <EOL> [ "<STR_LIT>" , stream_msgs , <NUM_LIT:0.0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ] , <EOL> [ "<STR_LIT>" , subword_rd_msgs , <NUM_LIT:0.0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ] , <EOL> [ "<STR_LIT>" , subword_wr_msgs , <NUM_LIT:0.0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ] , <EOL> [ "<STR_LIT>" , amo_msgs , <NUM_LIT:0.0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ] , <EOL> [ "<STR_LIT>" , random_msgs , <NUM_LIT:0.0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ] , <EOL> [ "<STR_LIT>" , random_msgs , <NUM_LIT:0.0> , <NUM_LIT:0> , <NUM_LIT:3> , <NUM_LIT> ] , <EOL> [ "<STR_LIT>" , stream_msgs , <NUM_LIT:0.5> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ] , <EOL> [ "<STR_LIT>" , stream_msgs , <NUM_LIT:0.0> , <NUM_LIT:4> , <NUM_LIT:0> , <NUM_LIT:0> ] , <EOL> [ "<STR_LIT>" , stream_msgs , <NUM_LIT:0.5> , <NUM_LIT:4> , <NUM_LIT:0> , <NUM_LIT:0> ] , <EOL> [ "<STR_LIT>" , random_msgs , <NUM_LIT:0.5> , <NUM_LIT:4> , <NUM_LIT:3> , <NUM_LIT> ] , <EOL> ] ) <EOL> @ pytest . mark . parametrize ( ** test_case_table ) <EOL> def test_1port ( test_params , dump_vcd ) : <EOL> msgs = test_params . msg_func ( <NUM_LIT> ) <EOL> run_sim ( TestHarness ( <NUM_LIT:1> , [ msgs [ : : <NUM_LIT:2> ] ] , [ msgs [ <NUM_LIT:1> : : <NUM_LIT:2> ] ] , <EOL> test_params . stall , test_params . lat , <EOL> test_params . src , test_params . sink ) , <EOL> dump_vcd ) <EOL> @ pytest . mark . parametrize ( ** test_case_table ) <EOL> def test_2port ( test_params , dump_vcd ) : <EOL> msgs0 = test_params . msg_func ( <NUM_LIT> ) <EOL> msgs1 = test_params . msg_func ( <NUM_LIT> ) <EOL> run_sim ( TestHarness ( <NUM_LIT:2> , [ msgs0 [ : : <NUM_LIT:2> ] , msgs1 [ : : <NUM_LIT:2> ] ] , <EOL> [ msgs0 [ <NUM_LIT:1> : : <NUM_LIT:2> ] , msgs1 [ <NUM_LIT:1> : : <NUM_LIT:2> ] ] , <EOL> test_params . stall , test_params . lat , <EOL> test_params . src , test_params . sink ) , <EOL> dump_vcd ) <EOL> def test_read_write_mem ( dump_vcd ) : <EOL> rgen = random . Random ( ) <EOL> rgen . seed ( <NUM_LIT> ) <EOL> data = [ rgen . randrange ( - ( <NUM_LIT:2> ** <NUM_LIT> ) , <NUM_LIT:2> ** <NUM_LIT> ) for _ in range ( <NUM_LIT:20> ) ] <EOL> data_bytes = struct . pack ( "<STR_LIT>" . format ( len ( data ) ) , * data ) <EOL> msgs = [ ] <EOL> for i , item in enumerate ( data ) : <EOL> msgs . extend ( [ <EOL> req ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT> + <NUM_LIT:4> * i , <NUM_LIT:0> , <NUM_LIT:0> ) , resp ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , item ) , <EOL> ] ) <EOL> th = TestHarness ( <NUM_LIT:1> , [ msgs [ : : <NUM_LIT:2> ] ] , [ msgs [ <NUM_LIT:1> : : <NUM_LIT:2> ] ] , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> th . mem . write_mem ( <NUM_LIT> , data_bytes ) <EOL> run_sim ( th , dump_vcd ) <EOL> result_bytes = th . mem . read_mem ( <NUM_LIT> , len ( data_bytes ) ) <EOL> result = list ( struct . unpack ( "<STR_LIT>" . format ( len ( data ) ) , result_bytes ) ) <EOL> assert result == data </s>
