Best Python code snippet using slash
rpctests.py
Source:rpctests.py
...11from common import TestCommon, TimeoutError12from results import RawResults, PassFailResult13class RpcTestCommon(TestCommon):14 benchmark_type = None15 def get_module_name(self):16 raise NotImplementedError17 def get_modules(self, build, machine):18 modules = super(RpcTestCommon, self).get_modules(build, machine)19 modules.add_module(self.get_module_name())20 return modules21 def run(self, build, machine, testdir):22 modules = self.get_modules(build, machine)23 args = []24 if self.benchmark_type != None:25 args = [str(self.benchmark_type)]26 modules.reset_module(self.get_module_name(), args)27 self.boot(machine, modules)28 return self.collect_data(machine)29 def process_data(self, testdir, rawiter):30 results = RawResults('core')31 times = []32 core = None33 for line in rawiter:34 m = re.match("Running \w+ between core \d+ and core (\d+)", line)35 if m:36 if times:37 results.add_group(core, times)38 core = int(m.group(1))39 times = []40 continue41 m = re.match("page \d+ took (\d+)", line)42 if m:43 assert(core is not None)44 times.append(int(m.group(1)))45 if len(times) != 0:46 results.add_group(core, times)47 return results48@tests.add_test49class TransportLatencyTest(RpcTestCommon):50 ''' Transport Latency microbenchmark '''51 name = "transport_latency"52 def get_module_name(self):53 return "transport_bench"54 def process_data(self, testdir, rawiter):55 results = RawResults('iteration')56 times = []57 iteration = None58 for line in rawiter:59 m = re.match("transport_bench: iteration (\d+)", line)60 if m:61 if times:62 results.add_group(iteration, times)63 iteration = int(m.group(1))64 times = []65 continue66 m = re.match("page \d+ took (\d+)", line)67 if m:68 assert(iteration is not None)69 times.append(int(m.group(1)))70 if len(times) != 0:71 results.add_group(iteration, times)72 return results73@tests.add_test74class FlounderStubsEmptyTest(RpcTestCommon):75 ''' Empty flounder stubs '''76 name = "flounder_stubs_empty"77 def get_module_name(self):78 return "flounder_stubs_empty_bench"79@tests.add_test80class FlounderStubsPayloadGlueTest(RpcTestCommon):81 ''' Payload_Glue flounder stubs '''82 name = "flounder_stubs_payload_glue"83 benchmark_type = 084 def get_module_name(self):85 return "flounder_stubs_payload_glue_bench"86@tests.add_test87class FlounderStubsPayload1GlueTest(RpcTestCommon):88 ''' Payload_Glue1 flounder stubs '''89 name = "flounder_stubs_payload1_glue"90 benchmark_type = 291 def get_module_name(self):92 return "flounder_stubs_payload_glue_bench"93@tests.add_test94class FlounderStubsPayload2GlueTest(RpcTestCommon):95 ''' Payload_Glue2 flounder stubs '''96 name = "flounder_stubs_payload2_glue"97 benchmark_type = 398 def get_module_name(self):99 return "flounder_stubs_payload_glue_bench"100@tests.add_test101class FlounderStubsPayload8GlueTest(RpcTestCommon):102 ''' Payload_Glue8 flounder stubs '''103 name = "flounder_stubs_payload8_glue"104 benchmark_type = 4105 def get_module_name(self):106 return "flounder_stubs_payload_glue_bench"107@tests.add_test108class FlounderStubsPayload16GlueTest(RpcTestCommon):109 ''' Payload_Glue16 flounder stubs '''110 name = "flounder_stubs_payload16_glue"111 benchmark_type = 5112 def get_module_name(self):113 return "flounder_stubs_payload_glue_bench"114@tests.add_test115class FlounderStubsEmptyGlueTest(RpcTestCommon):116 ''' Empty_Glue flounder stubs '''117 name = "flounder_stubs_empty_glue"118 benchmark_type = 1119 def get_module_name(self):120 return "flounder_stubs_payload_glue_bench"121@tests.add_test122class FlounderStubsPayloadTest(RpcTestCommon):123 ''' Payload flounder stubs '''124 name = "flounder_stubs_payload"125 def get_module_name(self):126 return "flounder_stubs_payload_bench"127@tests.add_test128class FlounderStubsPayload1Test(RpcTestCommon):129 ''' Payload1 flounder stubs '''130 name = "flounder_stubs_payload1"131 benchmark_type = 2132 def get_module_name(self):133 return "flounder_stubs_payload_bench"134@tests.add_test135class FlounderStubsPayload2Test(RpcTestCommon):136 ''' Payload2 flounder stubs '''137 name = "flounder_stubs_payload2"138 benchmark_type = 3139 def get_module_name(self):140 return "flounder_stubs_payload_bench"141@tests.add_test142class FlounderStubsPayload8Test(RpcTestCommon):143 ''' Payload8 flounder stubs '''144 name = "flounder_stubs_payload8"145 benchmark_type = 4146 def get_module_name(self):147 return "flounder_stubs_payload_bench"148@tests.add_test149class FlounderStubsPayload16Test(RpcTestCommon):150 ''' Payload16 flounder stubs '''151 name = "flounder_stubs_payload16"152 benchmark_type = 5153 def get_module_name(self):154 return "flounder_stubs_payload_bench"155@tests.add_test156class FlounderStubsBufferTest(RpcTestCommon):157 ''' Buffer flounder stubs '''158 name = "flounder_stubs_buffer"159 def get_module_name(self):160 return "flounder_stubs_buffer_bench"161@tests.add_test162class RouteTest(RpcTestCommon):163 ''' routing microbenchmark '''164 name = "route_bench"165 def get_module_name(self):166 return "route_bench"167@tests.add_test168class UmpLatencyTest(RpcTestCommon):169 ''' UMP latency microbenchmark '''170 name = "ump_latency"171 def get_module_name(self):172 return "ump_latency"173@tests.add_test174class UmpExchangeTest(RpcTestCommon):175 ''' UMP exchange microbenchmark '''176 name = "ump_exchange"177 def get_module_name(self):178 return "ump_exchange"179@tests.add_test180class UmpThroughputTest(RpcTestCommon):181 ''' UMP throughput microbenchmark '''182 name = "ump_throughput"183 def get_module_name(self):184 return "ump_throughput"185@tests.add_test186class UmpSendTest(RpcTestCommon):187 ''' UMP send microbenchmark '''188 name = "ump_send"189 def get_module_name(self):190 return "ump_send"191@tests.add_test192class UmpReceiveTest(RpcTestCommon):193 ''' UMP receive microbenchmark '''194 name = "ump_receive"195 def get_module_name(self):196 return "ump_receive"197@tests.add_test198class UmpCacheTest(RpcTestCommon):199 ''' UMP cache utilisation microbenchmark '''200 name = "ump_latency_cache"201 def get_module_name(self):202 return "ump_latency_cache"203 # XXX: don't attempt to run on an intel machine, as perfmon is unsupported204 def run(self, build, machine, testdir):205 perftype = machine.get_perfcount_type()206 if perftype.startswith('intel'):207 return ["Test skipped: %s performance monitoring unsupported\n" % perftype]208 else:209 return super(RpcTestCommon, self).run(build, machine, testdir)210 def process_data(self, testdir, raw_iter):211 # check the first line to see if the test actually ran212 try:213 firstline = raw_iter.next()214 if firstline.startswith('Test skipped'):215 return PassFailResult(True)216 except StopIteration: # empty results file217 return PassFailResult(False)218 # process the rest of the file219 results = {}220 data = {}221 index = {}222 iteration = {}223 results['Data'] = RawResults('iter', name='dcache')224 results['Instruction'] = RawResults('iter', name='icache')225 for cache in ['Data', 'Instruction']:226 data[cache] = []227 index[cache] = None228 iteration[cache] = 1229 for line in raw_iter:230 m = re.match("(Data|Instruction) cache miss\s+(\d+)\s+(\d+)\s+(\d+)",231 line)232 if m:233 cache = m.group(1)234 i = int(m.group(2))235 value = int(m.group(4)) - int(m.group(3))236 if index[cache] is not None and i < index[cache]: # new iteration237 results[cache].add_group(iteration[cache], data[cache])238 iteration[cache] += 1239 data[cache] = [value]240 else:241 data[cache].append(value)242 index[cache] = i243 ret = []244 for cache in ['Data', 'Instruction']:245 results[cache].add_group(iteration[cache], data[cache])246 ret.append(results[cache])247 return ret248@tests.add_test249class LrpcTest(RpcTestCommon):250 ''' LRPC microbenchmark '''251 name = "lrpc"252 def get_module_name(self):253 return "lrpc_bench"254 def get_modules(self, build, machine):255 modules = super(RpcTestCommon, self).get_modules(build, machine)256 modules.add_module("lrpc_bench", ["server"])257 modules.add_module("lrpc_bench", ["client"])258 return modules259 def get_finish_string(self):260 return "End of benchmarks."261 def run(self, *args):262 # XXX: want to use the base single-boot implementation263 return TestCommon.run(self, *args)264 def process_data(self, testdir, raw_iter):265 results = RawResults('iter')266 data = []...
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!