KaneAI - World’s First E2E Software Testing Agent.Signup for Private Beta 
How to use route method in Playwright Internal
Best JavaScript code snippet using playwright-internal
extra_function.py
Source:extra_function.py 
1from user_class import *2import copy3import random4from memory_profiler import profile5import json6import argparse7def parse_arguments():8 9 parser = argparse.ArgumentParser()10 subparser = parser.add_mutually_exclusive_group(required=True)11 subparser.add_argument('-net', metavar='net.json', help='net file')12 subparser.add_argument('-b', metavar='behavior_space.json', help='behavior space')13 subparser.add_argument('-o', metavar='oss_space.json', help='behavior space from osservation')14 subparser.add_argument('-cs', metavar='closing_space.json', help='silence closing space')15 subparser.add_argument('-dgn', metavar='diagnosticator.json', help='diagnosticator')16 17 subparser2 = parser.add_mutually_exclusive_group(required=False)18 subparser2.add_argument("-all", help='run all program', action='store_const', const='all', dest='step')19 subparser2.add_argument('-gb', help='generate behavior space', action='store_const', const='gb', dest='step')20 21 subparser2.add_argument("-go", help='generate behavior space from osservation', action='store_const', const='gbo', dest='step')22 23 subparser2.add_argument("-d", help='diagnosis behavior space', action='store_const', const='d', dest='step')24 subparser2.add_argument("-gcs", help='generate closing space', action='store_const', const='gcs', dest='step')25 26 #subparser2.add_argument("-dc", help='diagnosis closing space', action='store_const', const='dc', dest='step')27 subparser2.add_argument("-gd", help='generate diagnosticator', action='store_const', const='gd', dest='step')28 subparser2.add_argument("-do", help='linear_diagnostic', action='store_const', const='do', dest='step')29 30 parser.add_argument("-ol2", help='osservation list for diagnosticator', required=False, nargs='*')31 parser.add_argument("-ol", help='osservation list', required=False, nargs='*')32 33 parser.add_argument("-view", help='view file', required=False)34 parser.add_argument("-bk", help='view file', action='store_true', required=False) 35 36 return parser.parse_args()37def json_to_obj(path):38 39 with open(path) as f:40 data = json.load(f)41 list_FA = []42 list_link = []43 for link in data["list_link"]:44 LINK = Link(link["start_FA"], link["finish_FA"], link["alias"])45 list_link.append(LINK)46 for fa in data["list_fa"]:47 48 list_states = []49 for state in fa["list_states"]:50 state_alias = state["state_alias"]51 state_obj = State(alias=state_alias)52 list_states.append(state_obj)53 54 list_trans = []55 for trans in fa["list_trans"]:56 alias = trans["alias"]57 state_in_alias = trans["state_in"]58 state_out_alias = trans["state_out"]59 label_oss = trans["label_oss"] if trans["label_oss"] else '\u03b5'60 label_rel = trans["label_rel"] if trans["label_rel"] else '\u03b5'61 input_event = trans["input_event"]62 output_event = trans["output_event"]63 state_in = None64 state_out = None65 for state in list_states:66 if state_in and state_out:67 break68 if state.alias == state_in_alias:69 state_in = state70 if state.alias == state_out_alias:71 state_out = state72 73 input_event_func = None74 if input_event:75 event = Event(input_event["event"])76 link_alias = input_event["link"]77 link = None78 for link_temp in list_link:79 if link_temp.alias == link_alias:80 link = link_temp81 break82 83 input_event_func = EventFunction(event,link)84 85 output_events_func = []86 for out_ev in output_event:87 event = Event(out_ev["event"])88 link_alias = out_ev["link"]89 link = None90 for link_temp in list_link:91 if link_temp.alias == link_alias:92 link = link_temp93 break94 95 out_temp = EventFunction(event,link)96 output_events_func.append(out_temp)97 trans_obj = Transition(state_in, state_out, input_event_func = input_event_func, output_events_func = output_events_func, label_oss = label_oss, label_rel = label_rel, alias = alias)98 list_trans.append(trans_obj)99 100 101 initial_state_alias = fa["initial_state"]102 actual_state_alias = fa["actual_state"]103 initial_state = None104 actual_state = None105 for state in list_states:106 if initial_state and actual_state:107 break108 if state.alias == initial_state_alias:109 initial_state = state110 if state.alias == actual_state_alias:111 actual_state = state112 alias = fa["FA_alias"]113 FA_temp = FA(list_states, list_trans, initial_state, actual_state, alias = alias)114 list_FA.append(FA_temp)115 net = Net(list_FA, list_link)116 '''for f in net.list_FA:117 print(f.alias)118 for state in f.list_states:119 print(state.alias)120 for trans in f.list_trans:121 print(trans.alias, trans.start_state.alias, trans.finish_state.alias, trans.label_oss)122 print("input", trans.input_event_func)'''123 124 '''with open("net.json", "w") as f:125 f.write(net)'''126 127 return net128def format_net_to_text(doc, text_to_file):129 text_to_file += "NET with " + str(len(doc.list_FA)) + " FAs and " + str(len(doc.list_link)) + " LINKS\n"130 text_to_file += ""131 for f in doc.list_FA:132 text_to_file += "\tFA: " + f.alias + " with " + str(len(f.list_states)) + " states and " + str(len(f.list_trans))+" trans\n"133 134 text_to_file += "\t\t STATES: "135 136 for state in f.list_states:137 text_to_file += state.alias + " "138 text_to_file += "\n"139 text_to_file += "\t\t TRANSITION: "140 for trans in f.list_trans:141 text_to_file += trans.alias + " "142 text_to_file += "\n"143 return text_to_file144def BFS(list_routes, list_space_nodes, final_node):145 grey = []146 good_nodes = set([])147 grey.append(final_node)148 q = []149 q.append(final_node)150 while q:151 node = q.pop(0)152 for route in list_routes:153 if route.finish_node == node:154 if route.start_node not in grey and route.start_node not in good_nodes:155 q.append(route.start_node)156 grey.append(route.start_node)157 good_nodes.add(node)158 grey.remove(node)159 return list_space_nodes - good_nodes160def change_state(transition, index_FA, state_alias, list_link, node, osservation = None):161 new_list_values_link = node.list_values_link.copy()162 new_list_states_FA = node.list_states_FA.copy()163 index_oss = node.index_oss164 if osservation != None:165 if transition.label_oss in osservation and index_oss < len(osservation):166 if transition.label_oss != osservation[index_oss]:167 return None168 else:169 index_oss += 1170 else:171 if transition.label_oss != '\u03b5':172 return None173 func = transition.input_event_func174 if func is not None:175 func_link = func.link176 func_event = func.event177 index_link = list_link.index(func_link)178 if new_list_values_link[index_link] != func_event.alias:179 return None180 else:181 new_list_values_link[index_link] = ''182 funcs = transition.output_events_func183 for func in funcs:184 func_link = func.link185 func_event = func.event186 index_link = list_link.index(func_link)187 if new_list_values_link[index_link] != '':188 return None189 else:190 new_list_values_link[index_link] = func_event.alias191 new_state_alias = transition.finish_state.alias192 new_list_states_FA[index_FA] = new_state_alias193 return Node(list_states_FA = new_list_states_FA, list_values_link = new_list_values_link, index_oss = index_oss)194from operator import attrgetter195from itertools import groupby196def reg_expr(oss_space):197 list_routes = oss_space.list_routes198 list_nodes = oss_space.list_nodes199 initial_node = oss_space.initial_node200 if len(list_nodes) == 0:201 print("No nodes")202 return ''203 list_nodes.remove(initial_node)204 if any(route for route in list_routes if route.finish_node == initial_node):205 place_holder_node = Node([],[])206 new_route = Route(place_holder_node, initial_node)207 list_routes.append(new_route)208 initial_node = place_holder_node209 210 final_node = Node([],[])211 for node in list_nodes:212 if node.final:213 new_route = Route(node, final_node)214 list_routes.append(new_route)215 while len(list_nodes) > 0:216 list_route_node_auto = []217 list_route_node_input = []218 list_route_node_output = []219 node = random.choice(list_nodes)220 #print(node.id)221 222 for route in list_routes:223 if route.start_node == node and route.finish_node == node:224 list_route_node_auto.append(route)225 if route.start_node == node and route.finish_node != node:226 list_route_node_output.append(route) 227 if route.finish_node == node and route.start_node != node:228 list_route_node_input.append(route)229 230 if len(list_route_node_output) > 1:231 232 groups = groupby(list_route_node_output, attrgetter('finish_node')) #o(n)233 for (key, data) in groups:234 routes = list(data)235 if len(routes) > 1: 236 new_labels_rel = set()237 for route in routes:238 if len(route.label_rel) > 2:239 if route.label_rel[0] == '(' and route.label_rel[-1] == ')' and route.label_rel[-2]!='*':240 route.label_rel = route.label_rel[1:-1]241 #print(route.label_rel)242 for label_t in route.set_label_rel:243 new_labels_rel.add(label_t)244 list_routes.remove(route)245 list_route_node_output.remove(route)246 247 new_label_rel = '|'.join(new_labels_rel)248 #print(new_label_rel)249 new_route = Route(node, key, label_rel = new_label_rel)250 new_route.set_label_rel = new_labels_rel251 list_routes.append(new_route)252 list_route_node_output.append(new_route)253 if len(list_route_node_input) == 1 and len(list_route_node_output) == 1 and len(list_route_node_auto) == 0:254 route_in = list_route_node_input[0]255 route_out = list_route_node_output[0]256 new_label, set_label = clear_label(route_in.label_rel, route_out.label_rel, set_label1=route_in.set_label_rel, set_label2=route_out.set_label_rel)257 258 new_route = Route(route_in.start_node, route_out.finish_node , label_rel = new_label)259 new_route.set_label_rel = set_label260 261 list_routes.append(new_route)262 263 list_routes.remove(route_in)264 list_routes.remove(route_out)265 list_route_node_input.remove(route_in)266 list_route_node_output.remove(route_out)267 268 269 if len(list_route_node_auto) > 0:270 for auto_route in list_route_node_auto:271 272 for input_route in list_route_node_input:273 for output_route in list_route_node_output:274 new_label, set_label = clear_label(input_route.label_rel, output_route.label_rel, autolabel=auto_route.label_rel, set_label1=input_route.set_label_rel, set_label2=output_route.set_label_rel, set_auto=auto_route.set_label_rel)275 new_route = Route(input_route.start_node, output_route.finish_node, label_rel = new_label) 276 #print("entrato", auto_route.label_rel)277 new_route.set_label_rel = set_label278 #print("uscito", new_route.label_rel)279 list_routes.append(new_route)280 list_routes.remove(input_route) 281 list_routes.remove(auto_route)282 for output_route in list_route_node_output:283 list_routes.remove(output_route)284 else:285 for input_route in list_route_node_input:286 for output_route in list_route_node_output:287 new_label, set_label = clear_label(input_route.label_rel, output_route.label_rel, set_label1=input_route.set_label_rel, set_label2=output_route.set_label_rel)288 new_route = Route(input_route.start_node, output_route.finish_node, label_rel = new_label)289 new_route.set_label_rel = set_label290 list_routes.append(new_route)291 list_routes.remove(input_route)292 for output_route in list_route_node_output:293 list_routes.remove(output_route)294 list_nodes.remove(node)295 '''for route in list_routes:296 print("ROUTE",route.start_node.id, route.finish_node.id)297 '''298 final_labels = set()299 for route in list_routes:300 #new_labels_rel = set()301 if len(route.label_rel) > 2:302 if route.label_rel[0] == '(' and route.label_rel[-1] == ')' and route.label_rel[-2]!='*':303 route.label_rel = route.label_rel[1:-1]304 305 if len(route.set_label_rel) > 1:306 list_label = route.label_rel.split("|")307 for label in list_label:308 if label[0] == '(' and label[-1] == ')' and label[-2]!='*':309 label = label[1:-1]310 final_labels.add(label)311 '''for label in route.set_label_rel:312 final_labels.add(label)'''313 else:314 final_labels.add(route.label_rel)315 final_label = '|'.join(final_labels)316 #new_route = Route(node, key, label_rel = final_label)317 318 return final_label #return regular expression319def reg_expr_closing(closing_space):320 321 list_routes = closing_space.list_routes322 list_nodes = [node.node for node in closing_space.list_nodes]323 initial_node = closing_space.initial_node324 list_output_routes = closing_space.list_output_routes325 list_final_output_nodes = []326 list_output_nodes = []327 #list_final_routes = []328 list_nodes_not_final = []329 ex_initial_node_id = -2330 if len(list_nodes) == 1:331 new_route = Route(initial_node.node, initial_node.node)332 new_route.label_rel = '\u03b5'333 new_route.rif_node = initial_node.node 334 return [new_route] 335 336 if any(route for route in list_routes if route.finish_node.id == initial_node.node.id):337 338 place_holder_node = Node([],[])339 new_route = Route(place_holder_node, initial_node.node)340 list_routes.append(new_route)341 list_nodes_not_final.append(place_holder_node)342 ex_initial_node_id = initial_node.node.id343 initial_node.node = place_holder_node344 place_holder_node.id = -1345 346 final_node = Node([],[])347 final_node.id = -99348 349 for node in list_nodes:350 if node.final:351 list_final_output_nodes.append(node)352 new_route = Route(node, final_node)353 list_routes.append(new_route)354 else:355 list_nodes_not_final.append(node)356 for route in list_output_routes:357 if not route.start_node.final:358 if not route.start_node in list_output_nodes:359 list_output_nodes.append(route.start_node)360 list_final_output_nodes.append(route.start_node)361 new_route = Route(route.start_node, final_node)362 list_routes.append(new_route)363 list_nodes_not_final.remove(route.start_node)364 number_expr_nodes = len(list_final_output_nodes)365 366 if initial_node.node in list_nodes_not_final:367 list_nodes_not_final.remove(initial_node.node)368 while len(list_nodes_not_final) > 0:369 list_route_node_auto = []370 list_route_node_input = []371 list_route_node_output = []372 node = random.choice(list_nodes_not_final)373 374 for route in list_routes:375 if route.start_node == node and route.finish_node == node:376 list_route_node_auto.append(route)377 if route.start_node == node and route.finish_node != node:378 list_route_node_output.append(route) 379 if route.finish_node == node and route.start_node != node:380 list_route_node_input.append(route)381 382 if len(list_route_node_output) > 1: 383 groups = groupby(list_route_node_output, attrgetter('finish_node')) #o(n)384 for (key, data) in groups:385 routes = list(data)386 if len(routes) > 1:387 new_labels_rel = set()388 for route in routes:389 if len(route.label_rel) > 2:390 if route.label_rel[0] == '(' and route.label_rel[-1] == ')':391 route.label_rel = route.label_rel[1:-1]392 for label_t in route.set_label_rel:393 new_labels_rel.add(label_t)394 list_routes.remove(route)395 list_route_node_output.remove(route)396 397 new_label_rel = '|'.join(new_labels_rel)398 if len(new_labels_rel) > 1:399 new_route = Route(node, key, label_rel = new_label_rel)400 else:401 new_route = Route(node, key, label_rel = new_label_rel)402 #print(new_route.label_rel)403 new_route.set_label_rel = new_labels_rel404 405 list_routes.append(new_route)406 list_route_node_output.append(new_route)407 if len(list_route_node_input) == 1 and len(list_route_node_output) == 1 and len(list_route_node_auto) == 0:408 route_in = list_route_node_input[0]409 route_out = list_route_node_output[0]410 new_label, set_label = clear_label(route_in.label_rel, route_out.label_rel, set_label1=route_in.set_label_rel, set_label2=route_out.set_label_rel)411 412 new_route = Route(route_in.start_node, route_out.finish_node , label_rel = new_label)413 new_route.set_label_rel = set_label414 415 list_routes.append(new_route)416 list_routes.remove(route_in)417 list_routes.remove(route_out)418 list_route_node_input.remove(route_in)419 list_route_node_output.remove(route_out)420 421 422 if len(list_route_node_auto) > 0:423 for auto_route in list_route_node_auto:424 for input_route in list_route_node_input:425 for output_route in list_route_node_output:426 new_label, set_label = clear_label(input_route.label_rel, output_route.label_rel, autolabel=auto_route.label_rel, set_label1=input_route.set_label_rel, set_label2=output_route.set_label_rel, set_auto=auto_route.set_label_rel)427 new_route = Route(input_route.start_node, output_route.finish_node, label_rel = new_label) 428 429 new_route.set_label_rel = set_label430 list_routes.append(new_route)431 list_routes.remove(input_route) 432 list_routes.remove(auto_route)433 434 for output_route in list_route_node_output:435 list_routes.remove(output_route)436 else:437 for input_route in list_route_node_input:438 for output_route in list_route_node_output:439 new_label, set_label = clear_label(input_route.label_rel, output_route.label_rel, set_label1=input_route.set_label_rel, set_label2=output_route.set_label_rel)440 new_route = Route(input_route.start_node, output_route.finish_node, label_rel = new_label)441 new_route.set_label_rel = set_label442 list_routes.append(new_route)443 list_routes.remove(input_route)444 for output_route in list_route_node_output:445 list_routes.remove(output_route)446 list_nodes_not_final.remove(node)447 448 #### QUA####449 while len(list_final_output_nodes) > 0:450 list_route_node_auto = []451 list_route_node_input = []452 list_route_node_output = []453 node = random.choice(list_final_output_nodes)454 #print("node id", node.id)455 '''for route in list_routes:456 print("route", route.start_node.id, route.finish_node.id, route.label_rel)457 '''458 for route in list_routes:459 if route.start_node == node and route.finish_node == node:460 list_route_node_auto.append(route)461 if route.start_node == node and route.finish_node != node:462 list_route_node_output.append(route) 463 if route.finish_node == node and route.start_node != node:464 list_route_node_input.append(route)465 466 if len(list_route_node_input) == 1 and len(list_route_node_output) == 1 and len(list_route_node_auto) == 0:467 route_in = list_route_node_input[0]468 route_out = list_route_node_output[0]469 new_label, set_label = clear_label(route_in.label_rel, route_out.label_rel, set_label1=route_in.set_label_rel, set_label2=route_out.set_label_rel)470 471 new_route = Route(route_in.start_node, route_out.finish_node , label_rel = new_label)472 new_route.set_label_rel = set_label473 474 if route_out.rif_node != None:475 new_route.rif_node = route_out.rif_node476 elif route_out.finish_node == final_node:477 new_route.rif_node = node478 479 list_routes.append(new_route)480 481 list_routes.remove(route_in)482 list_routes.remove(route_out)483 list_route_node_input.remove(route_in)484 list_route_node_output.remove(route_out)485 486 if len(list_route_node_auto) > 0:487 remove = False488 for auto_route in list_route_node_auto:489 for input_route in list_route_node_input:490 remove = True491 for output_route in list_route_node_output:492 new_label, set_label = clear_label(input_route.label_rel, output_route.label_rel, autolabel=auto_route.label_rel, set_label1=input_route.set_label_rel, set_label2=output_route.set_label_rel, set_auto=auto_route.set_label_rel)493 new_route = Route(input_route.start_node, output_route.finish_node, label_rel = new_label) 494 495 new_route.set_label_rel = set_label496 list_routes.append(new_route)497 if output_route.rif_node != None:498 new_route.rif_node = output_route.rif_node499 elif output_route.finish_node == final_node:500 new_route.rif_node = node501 list_routes.remove(input_route) 502 list_routes.remove(auto_route)503 504 if remove:505 for output_route in list_route_node_output:506 list_routes.remove(output_route)507 if len(list_route_node_input) == 0:508 for output_route in list_route_node_output: 509 if output_route.rif_node != None:510 output_route.rif_node = output_route.rif_node511 elif output_route.finish_node == final_node:512 output_route.rif_node = node 513 else:514 remove = False515 for input_route in list_route_node_input:516 remove = True517 #print("qua", input_route.label_rel)518 for output_route in list_route_node_output:519 if input_route.start_node.id == -1 and output_route.finish_node.id == -99:520 #print("rimosso")521 continue522 new_label, set_label = clear_label(input_route.label_rel, output_route.label_rel, set_label1=input_route.set_label_rel, set_label2=output_route.set_label_rel)523 new_route = Route(input_route.start_node, output_route.finish_node, label_rel = new_label)524 new_route.set_label_rel = set_label525 #print("nuova rotta", new_route.label_rel, new_route.start_node.id, new_route.finish_node.id)526 list_routes.append(new_route)527 528 if output_route.rif_node != None:529 new_route.rif_node = output_route.rif_node530 elif output_route.finish_node == final_node:531 new_route.rif_node = node532 list_routes.remove(input_route)533 if remove:534 for output_route in list_route_node_output:535 list_routes.remove(output_route)536 if len(list_route_node_input) == 0:537 for output_route in list_route_node_output: 538 if output_route.rif_node != None:539 output_route.rif_node = output_route.rif_node540 elif output_route.finish_node == final_node:541 output_route.rif_node = node542 list_final_output_nodes.remove(node)543 '''for route in list_routes:544 print(closing_space.id, route.label_rel, route.rif_node.id)'''545 already_done = []546 routes = []547 for route in list_routes:548 if not route.rif_node.id in already_done:549 labels = set()550 gen = (r for r in list_routes if r.rif_node.id == route.rif_node.id)551 for r in gen:552 labels.add(r.label_rel)553 554 route.label_rel = "|".join(labels)555 route.set_label_rel = labels556 routes.append(route)557 already_done.append(route.rif_node.id)558 return routes559def clear_label(label1, label2, autolabel = None, set_label1 = None, set_label2 = None, set_auto = None):560 new_label = ''561 new_label_set = {new_label}562 #print("clear_label", label1, label2, set_label1, set_label2, autolabel, set_auto)563 if autolabel and autolabel != '\u03b5':564 autolabel = "(" + autolabel + ")"565 autolabel, set_auto = clear_label(autolabel, '*', set_label1={autolabel}, set_label2={'*'})566 if label1 == '\u03b5' and label2 == '\u03b5':567 if autolabel:568 if autolabel == '\u03b5':569 new_label = '\u03b5'570 else:571 new_label = autolabel572 new_label_set = set_auto573 else:574 new_label = '\u03b5'575 new_label_set = {new_label}576 elif label1 == '\u03b5':577 if autolabel:578 if autolabel == '\u03b5':579 new_label = label2580 new_label_set = set_label2581 else:582 new_label, new_label_set = clear_label(autolabel, label2, set_label1=set_auto, set_label2=set_label2)583 else:584 new_label = label2585 new_label_set = set_label2586 elif label2 == '\u03b5':587 if autolabel:588 if autolabel == '\u03b5':589 new_label = label1590 new_label_set = set_label1591 else:592 new_label, new_label_set = clear_label(label1, autolabel, set_label1=set_label1, set_label2=set_auto)593 else:594 new_label = label1595 new_label_set = set_label1596 else:597 if autolabel:598 if autolabel == '\u03b5':599 new_label, new_label_set = clear_label(label1, label2, set_label1=set_label1, set_label2=set_label2)600 else:601 new_label, new_label_set = clear_label(label1, autolabel, set_label1=set_label1, set_label2=set_auto)602 new_label, new_label_set = clear_label(new_label, label2, set_label1=new_label_set, set_label2=set_label2)603 elif set_label1:604 labels = set()605 #new_label_set = set()606 for label_temp in set_label1:607 for label_temp2 in set_label2:608 new_label_temp, new_label_set_temp = clear_label(label_temp, label_temp2)609 labels.add(new_label_temp)610 611 new_label = '|'.join(labels)612 new_label_set = labels613 614 else:615 616 new_label = label1 + label2617 new_label_set = {label1, label2}618 619 #print("sono qua",new_label, new_label_set )620 return new_label, new_label_set621def linear_diagnostic(diagnosticator_space, osservation):622 X = []623 X.append((diagnosticator_space.initial_state.id,'\u03b5'))624 X_final = []625 X_final += X626 list_states = diagnosticator_space.list_states627 for o in osservation:628 Xnew = []629 #print("osservazione", o)630 for (state_id, p) in X:631 groups = groupby(list_states[state_id].list_routes, attrgetter('finish_state')) #o(n)632 #print(state_id)633 #print("STATO", state_id, list_states[state_id].delta)634 for (key, data) in groups:635 #print(key.id)636 routes = list(data)637 for route in routes:638 if route.label_oss == o:639 p_label_set = set(p.split("|"))640 route_label_set = set(route.label_rel.split("|"))641 new_p, temp = clear_label(p, route.label_rel, set_label1 = p_label_set, set_label2 = route_label_set)642 #print("\t", new_p, p, route.label_rel, route.set_label_rel, )643 add = True644 for e_tuple_temp in Xnew:645 if e_tuple_temp[0] == route.finish_state.id and sorted(e_tuple_temp[1]) != sorted(new_p):646 #c'è un errore647 lst = list(e_tuple_temp)648 add = False649 label_set = set()650 label_set.add(e_tuple_temp[1])651 label_set.add(new_p)652 lst[1] = '|'.join(label_set)653 e_tuple_temp = tuple(lst)654 #print("\tdio", lst[1])655 break656 if add:657 #print("stato aggiunto", route.finish_state.id)658 Xnew.append((route.finish_state.id, new_p))659 #print("\t\t", add)660 X = copy.deepcopy(Xnew)661 X_final += Xnew662 #print(X_final)663 final_diagnosis = []664 labels_to_return = set()665 for e in Xnew:666 state = list_states[e[0]]667 if state.delta != '':668 list_delta = set(state.delta.split('|'))669 p_label_set = set(e[1].split("|"))670 p = "|".join(p_label_set)671 delta = "|".join(list_delta)672 label, temp = clear_label(p, delta, set_label1=p_label_set, set_label2=list_delta)673 labels_to_return.add(frozenset(temp))674 for frozen in labels_to_return:675 final_diagnosis.append('|'.join(list(frozen)))676 #print(final_diagnosis)677 return final_diagnosis678 ...
test_v_s_route.py
Source:test_v_s_route.py
...135 assert_event_and_count(vsr_1_event_text, initial_count_vsr_1 + 4, events_ns_1)136 assert_event_with_full_equality_and_count(vs_event_text, initial_count_vs + 4, events_ns_1)137 assert_event_and_count(vsr_2_event_text, initial_count_vsr_2 + 4, events_ns_2)138 print("\nStep 6: remove VSRoute and check")139 delete_v_s_route(kube_apis.custom_objects, v_s_route_setup.route_m.name, v_s_route_setup.namespace)140 wait_before_test(1)141 new_config = get_vs_nginx_template_conf(kube_apis.v1,142 v_s_route_setup.namespace,143 v_s_route_setup.vs_name,144 ic_pod_name,145 ingress_controller_prerequisites.namespace)146 resp_1 = requests.get(f"{req_url}{v_s_route_setup.route_m.paths[0]}",147 headers={"host": v_s_route_setup.vs_host})148 resp_2 = requests.get(f"{req_url}{v_s_route_setup.route_m.paths[1]}",149 headers={"host": v_s_route_setup.vs_host})150 resp_3 = requests.get(f"{req_url}{v_s_route_setup.route_s.paths[0]}",151 headers={"host": v_s_route_setup.vs_host})152 assert resp_1.status_code == 404153 assert resp_2.status_code == 404154 assert resp_3.status_code == 200155 events_ns_1 = get_events(kube_apis.v1, v_s_route_setup.route_m.namespace)156 events_ns_2 = get_events(kube_apis.v1, v_s_route_setup.route_s.namespace)157 assert_locations_not_in_config(new_config, v_s_route_setup.route_m.paths)158 assert_event_and_count(vsr_1_event_text, initial_count_vsr_1 + 4, events_ns_1)159 assert_event_with_full_equality_and_count(vs_event_text, initial_count_vs + 4, events_ns_1)160 # a warning event because the VS references a non-existing VSR161 assert_event_with_full_equality_and_count(vs_warning_event_text, 1, events_ns_1)162 assert_event_and_count(vsr_2_event_text, initial_count_vsr_2 + 5, events_ns_2)163 print("\nStep 7: restore VSRoute and check")164 create_v_s_route_from_yaml(kube_apis.custom_objects,165 f"{TEST_DATA}/virtual-server-route/route-multiple.yaml",166 v_s_route_setup.namespace)167 wait_before_test(1)168 new_config = get_vs_nginx_template_conf(kube_apis.v1,169 v_s_route_setup.namespace,170 v_s_route_setup.vs_name,171 ic_pod_name,172 ingress_controller_prerequisites.namespace)173 resp_1 = requests.get(f"{req_url}{v_s_route_setup.route_m.paths[0]}",174 headers={"host": v_s_route_setup.vs_host})175 resp_2 = requests.get(f"{req_url}{v_s_route_setup.route_m.paths[1]}",176 headers={"host": v_s_route_setup.vs_host})177 resp_3 = requests.get(f"{req_url}{v_s_route_setup.route_s.paths[0]}",178 headers={"host": v_s_route_setup.vs_host})179 assert_responses_and_server_name(resp_1, resp_2, resp_3)180 events_ns_1 = get_events(kube_apis.v1, v_s_route_setup.route_m.namespace)181 events_ns_2 = get_events(kube_apis.v1, v_s_route_setup.route_s.namespace)182 assert_locations_in_config(new_config, v_s_route_setup.route_m.paths)183 assert_event_and_count(vsr_1_event_text, 1, events_ns_1)184 assert_event_with_full_equality_and_count(vs_event_text, initial_count_vs + 5, events_ns_1)185 assert_event_and_count(vsr_2_event_text, initial_count_vsr_2 + 6, events_ns_2)186 print("\nStep 8: remove one backend service and check")187 delete_service(kube_apis.v1, "backend1-svc", v_s_route_setup.route_m.namespace)188 wait_before_test(1)189 resp_1 = requests.get(f"{req_url}{v_s_route_setup.route_m.paths[0]}",190 headers={"host": v_s_route_setup.vs_host})191 resp_2 = requests.get(f"{req_url}{v_s_route_setup.route_m.paths[1]}",192 headers={"host": v_s_route_setup.vs_host})193 resp_3 = requests.get(f"{req_url}{v_s_route_setup.route_s.paths[0]}",194 headers={"host": v_s_route_setup.vs_host})195 assert resp_1.status_code == 502196 assert resp_2.status_code == 200197 assert resp_3.status_code == 200198 events_ns_1 = get_events(kube_apis.v1, v_s_route_setup.route_m.namespace)199 events_ns_2 = get_events(kube_apis.v1, v_s_route_setup.route_s.namespace)200 assert_event_and_count(vsr_1_event_text, 2, events_ns_1)201 assert_event_with_full_equality_and_count(vs_event_text, initial_count_vs + 6, events_ns_1)202 assert_event_and_count(vsr_2_event_text, initial_count_vsr_2 + 7, events_ns_2)203 print("\nStep 9: restore backend service and check")204 create_service_with_name(kube_apis.v1, v_s_route_setup.route_m.namespace, "backend1-svc")205 wait_before_test(1)206 resp_1 = requests.get(f"{req_url}{v_s_route_setup.route_m.paths[0]}",207 headers={"host": v_s_route_setup.vs_host})208 resp_2 = requests.get(f"{req_url}{v_s_route_setup.route_m.paths[1]}",209 headers={"host": v_s_route_setup.vs_host})210 resp_3 = requests.get(f"{req_url}{v_s_route_setup.route_s.paths[0]}",211 headers={"host": v_s_route_setup.vs_host})212 assert_responses_and_server_name(resp_1, resp_2, resp_3)213 events_ns_1 = get_events(kube_apis.v1, v_s_route_setup.route_m.namespace)214 events_ns_2 = get_events(kube_apis.v1, v_s_route_setup.route_s.namespace)215 assert_event_and_count(vsr_1_event_text, 3, events_ns_1)216 assert_event_with_full_equality_and_count(vs_event_text, initial_count_vs + 7, events_ns_1)217 assert_event_and_count(vsr_2_event_text, initial_count_vsr_2 + 8, events_ns_2)218 print("\nStep 10: remove VS and check")219 delete_virtual_server(kube_apis.custom_objects, v_s_route_setup.vs_name, v_s_route_setup.namespace)220 wait_before_test(1)221 resp_1 = requests.get(f"{req_url}{v_s_route_setup.route_m.paths[0]}",222 headers={"host": v_s_route_setup.vs_host})223 resp_2 = requests.get(f"{req_url}{v_s_route_setup.route_m.paths[1]}",224 headers={"host": v_s_route_setup.vs_host})225 resp_3 = requests.get(f"{req_url}{v_s_route_setup.route_s.paths[0]}",226 headers={"host": v_s_route_setup.vs_host})227 assert resp_1.status_code == 404228 assert resp_2.status_code == 404229 assert resp_3.status_code == 404230 list0_list_ns_1 = get_events(kube_apis.v1, v_s_route_setup.route_m.namespace)231 list0_list_ns_2 = get_events(kube_apis.v1, v_s_route_setup.route_s.namespace)232 assert_event_and_count(vsr_1_event_text, 3, list0_list_ns_1)233 assert_event_with_full_equality_and_count(vs_event_text, initial_count_vs + 7, list0_list_ns_1)234 assert_event_and_count(vsr_2_event_text, initial_count_vsr_2 + 8, list0_list_ns_2)235 print("\nStep 11: restore VS and check")236 create_virtual_server_from_yaml(kube_apis.custom_objects,237 f"{TEST_DATA}/virtual-server-route/standard/virtual-server.yaml",238 v_s_route_setup.namespace)239 wait_before_test(1)240 resp_1 = requests.get(f"{req_url}{v_s_route_setup.route_m.paths[0]}",241 headers={"host": v_s_route_setup.vs_host})242 resp_2 = requests.get(f"{req_url}{v_s_route_setup.route_m.paths[1]}",243 headers={"host": v_s_route_setup.vs_host})244 resp_3 = requests.get(f"{req_url}{v_s_route_setup.route_s.paths[0]}",245 headers={"host": v_s_route_setup.vs_host})246 assert_responses_and_server_name(resp_1, resp_2, resp_3)247 list1_list_ns_1 = get_events(kube_apis.v1, v_s_route_setup.route_m.namespace)248 list1_list_ns_2 = get_events(kube_apis.v1, v_s_route_setup.route_s.namespace)249 assert_event_and_count(vsr_1_event_text, 4, list1_list_ns_1)250 assert_event_with_full_equality_and_count(vs_event_text, 1, list1_list_ns_1)251 assert_event_and_count(vsr_2_event_text, initial_count_vsr_2 + 9, list1_list_ns_2)252@pytest.mark.vsr253@pytest.mark.parametrize('crd_ingress_controller, v_s_route_setup',254 [({"type": "complete", "extra_args": [f"-enable-custom-resources"]},255 {"example": "virtual-server-route"})],256 indirect=True)257class TestVirtualServerRouteValidation:258 def test_vsr_without_vs(self, kube_apis,259 ingress_controller_prerequisites,260 crd_ingress_controller,261 v_s_route_setup,262 test_namespace):263 ic_pod_name = get_first_pod_name(kube_apis.v1, ingress_controller_prerequisites.namespace)264 vsr_name = create_v_s_route_from_yaml(kube_apis.custom_objects,265 f"{TEST_DATA}/virtual-server-route/route-orphan.yaml",266 test_namespace)267 vsr_paths = get_paths_from_vsr_yaml(f"{TEST_DATA}/virtual-server-route/route-orphan.yaml")268 wait_before_test(1)269 new_config = get_vs_nginx_template_conf(kube_apis.v1,270 v_s_route_setup.namespace,271 v_s_route_setup.vs_name,272 ic_pod_name,273 ingress_controller_prerequisites.namespace)274 new_list_ns_3 = get_events(kube_apis.v1, test_namespace)275 assert_locations_not_in_config(new_config, vsr_paths)276 assert_event_and_count(f"VirtualServer {v_s_route_setup.namespace}/{v_s_route_setup.vs_name} ignores VirtualServerRoute", 1, new_list_ns_3)277 @pytest.mark.parametrize("route_yaml", [f"{TEST_DATA}/virtual-server-route/route-single-invalid-host.yaml",278 f"{TEST_DATA}/virtual-server-route/route-single-duplicate-path.yaml"])279 def test_make_existing_vsr_invalid(self, kube_apis,280 ingress_controller_prerequisites,281 crd_ingress_controller,282 v_s_route_setup,283 route_yaml):284 ic_pod_name = get_first_pod_name(kube_apis.v1, ingress_controller_prerequisites.namespace)285 patch_v_s_route_from_yaml(kube_apis.custom_objects,286 v_s_route_setup.route_s.name,287 route_yaml,288 v_s_route_setup.route_s.namespace)289 wait_before_test(1)290 new_config = get_vs_nginx_template_conf(kube_apis.v1,291 v_s_route_setup.namespace,292 v_s_route_setup.vs_name,293 ic_pod_name,294 ingress_controller_prerequisites.namespace)295 new_vs_events = get_events(kube_apis.v1, v_s_route_setup.namespace)296 new_vsr_events = get_events(kube_apis.v1, v_s_route_setup.route_s.namespace)297 assert_locations_not_in_config(new_config, v_s_route_setup.route_s.paths)298 text = f"{v_s_route_setup.route_s.namespace}/{v_s_route_setup.route_s.name}"299 assert_event_and_count(f"Configuration for {v_s_route_setup.namespace}/{v_s_route_setup.vs_name} was added or updated with warning(s)",300 1,301 new_vs_events)302 if route_yaml == f"{TEST_DATA}/virtual-server-route/route-single-invalid-host.yaml":303 assert_event_and_count(f"VirtualServer is invalid or doesn't exist",304 1,305 new_vsr_events)306 else:307 assert_event_and_count(f"VirtualServerRoute {text} was rejected with error",308 1,309 new_vsr_events)310 def test_openapi_validation_flow(self, kube_apis, ingress_controller_prerequisites,311 crd_ingress_controller, v_s_route_setup):312 ic_pod_name = get_first_pod_name(kube_apis.v1, ingress_controller_prerequisites.namespace)313 config_old = get_vs_nginx_template_conf(kube_apis.v1,314 v_s_route_setup.namespace,315 v_s_route_setup.vs_name,316 ic_pod_name,317 ingress_controller_prerequisites.namespace)318 route_yaml = f"{TEST_DATA}/virtual-server-route/route-single-invalid-openapi.yaml"319 try:320 patch_v_s_route_from_yaml(kube_apis.custom_objects,321 v_s_route_setup.route_s.name,322 route_yaml,323 v_s_route_setup.route_s.namespace)324 except ApiException as ex:325 assert ex.status == 422 and "spec.subroutes.action.pass" in ex.body326 except Exception as ex:327 pytest.fail(f"An unexpected exception is raised: {ex}")328 else:329 pytest.fail("Expected an exception but there was none")330 wait_before_test(1)331 config_new = get_vs_nginx_template_conf(kube_apis.v1,332 v_s_route_setup.namespace,333 v_s_route_setup.vs_name,334 ic_pod_name,335 ingress_controller_prerequisites.namespace)336 assert config_old == config_new, "Expected: config doesn't change"337@pytest.mark.vsr338@pytest.mark.parametrize('crd_ingress_controller, v_s_route_setup',339 [({"type": "complete", "extra_args": [f"-enable-custom-resources"]},340 {"example": "virtual-server-route"})],341 indirect=True)342class TestCreateInvalidVirtualServerRoute:343 def test_create_invalid_vsr(self, kube_apis,344 ingress_controller_prerequisites,345 crd_ingress_controller,346 v_s_route_setup):347 route_yaml = f"{TEST_DATA}/virtual-server-route/route-single-duplicate-path.yaml"348 ic_pod_name = get_first_pod_name(kube_apis.v1, ingress_controller_prerequisites.namespace)349 text = f"{v_s_route_setup.route_s.namespace}/{v_s_route_setup.route_s.name}"350 vs_event_text = f"Configuration for {v_s_route_setup.namespace}/{v_s_route_setup.vs_name} was added or updated with warning(s)"351 vsr_event_text = f"VirtualServerRoute {text} was rejected with error: spec.subroutes[1].path: Duplicate value: \"/backend2\""352 delete_v_s_route(kube_apis.custom_objects,353 v_s_route_setup.route_s.name,354 v_s_route_setup.route_s.namespace)355 create_v_s_route_from_yaml(kube_apis.custom_objects,356 route_yaml,357 v_s_route_setup.route_s.namespace)358 wait_before_test(1)359 new_config = get_vs_nginx_template_conf(kube_apis.v1,360 v_s_route_setup.namespace,361 v_s_route_setup.vs_name,362 ic_pod_name,363 ingress_controller_prerequisites.namespace)364 new_vs_events = get_events(kube_apis.v1, v_s_route_setup.namespace)365 new_vsr_events = get_events(kube_apis.v1, v_s_route_setup.route_s.namespace)366 assert_locations_not_in_config(new_config, v_s_route_setup.route_s.paths)...
test_bgpspeaker.py
Source:test_bgpspeaker.py
...287 @mock.patch(288 'ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',289 mock.MagicMock(return_value=None))290 @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')291 def test_evpn_prefix_add_ip_prefix_route(self, mock_call):292 # Prepare test data293 route_type = bgpspeaker.EVPN_IP_PREFIX_ROUTE294 route_dist = '65000:100'295 esi = 0 # denotes single-homed296 ethernet_tag_id = 200297 ip_prefix = '192.168.0.0/24'298 gw_ip_addr = '172.16.0.1'299 next_hop = '0.0.0.0'300 expected_kwargs = {301 'route_type': route_type,302 'route_dist': route_dist,303 'esi': esi,304 'ethernet_tag_id': ethernet_tag_id,305 'ip_prefix': ip_prefix,306 'gw_ip_addr': gw_ip_addr,307 'next_hop': next_hop,308 }309 # Test310 speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')311 speaker.evpn_prefix_add(312 route_type=route_type,313 route_dist=route_dist,314 esi=esi,315 ethernet_tag_id=ethernet_tag_id,316 ip_prefix=ip_prefix,317 gw_ip_addr=gw_ip_addr,318 )319 # Check320 mock_call.assert_called_with(321 'evpn_prefix.add_local', **expected_kwargs)322 @mock.patch(323 'ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',324 mock.MagicMock(return_value=None))325 @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')326 def test_evpn_prefix_add_ip_prefix_route_vni(self, mock_call):327 # Prepare test data328 route_type = bgpspeaker.EVPN_IP_PREFIX_ROUTE329 route_dist = '65000:100'330 esi = 0 # denotes single-homed331 ethernet_tag_id = 200332 ip_prefix = '192.168.0.0/24'333 gw_ip_addr = '172.16.0.1'334 vni = 500335 tunnel_type = bgpspeaker.TUNNEL_TYPE_VXLAN336 next_hop = '0.0.0.0'337 expected_kwargs = {338 'route_type': route_type,339 'route_dist': route_dist,340 'esi': esi,341 'ethernet_tag_id': ethernet_tag_id,342 'ip_prefix': ip_prefix,343 'gw_ip_addr': gw_ip_addr,344 'tunnel_type': tunnel_type,345 'vni': vni,346 'next_hop': next_hop,347 }348 # Test349 speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')350 speaker.evpn_prefix_add(351 route_type=route_type,352 route_dist=route_dist,353 esi=esi,354 ethernet_tag_id=ethernet_tag_id,355 ip_prefix=ip_prefix,356 gw_ip_addr=gw_ip_addr,357 tunnel_type=tunnel_type,358 vni=vni,359 )360 # Check361 mock_call.assert_called_with(362 'evpn_prefix.add_local', **expected_kwargs)363 @raises(ValueError)364 @mock.patch('ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',365 mock.MagicMock(return_value=None))366 @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')367 def test_evpn_prefix_add_invalid_route_type(self, mock_call):368 # Prepare test data369 route_type = 'foobar' # Invalid EVPN route type370 route_dist = '65000:100'371 esi = 0 # denotes single-homed372 ethernet_tag_id = 200373 mac_addr = 'aa:bb:cc:dd:ee:ff'374 ip_addr = '192.168.0.1'375 next_hop = '10.0.0.1'376 # Test377 speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')378 speaker.evpn_prefix_add(379 route_type=route_type,380 route_dist=route_dist,381 esi=esi,382 ethernet_tag_id=ethernet_tag_id,383 mac_addr=mac_addr,384 ip_addr=ip_addr,385 next_hop=next_hop,386 )387 # Check388 mock_call.assert_called_with(389 'evpn_prefix.add_local', 'Invalid arguments detected')390 @mock.patch(391 'ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',392 mock.MagicMock(return_value=None))393 @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')394 def test_evpn_prefix_del_auto_discovery(self, mock_call):395 # Prepare test data396 route_type = bgpspeaker.EVPN_ETH_AUTO_DISCOVERY397 route_dist = '65000:100'398 esi = {399 'type': ESI_TYPE_LACP,400 'mac_addr': 'aa:bb:cc:dd:ee:ff',401 'port_key': 100,402 }403 ethernet_tag_id = EVPN_MAX_ET404 expected_kwargs = {405 'route_type': route_type,406 'route_dist': route_dist,407 'esi': esi,408 'ethernet_tag_id': ethernet_tag_id,409 }410 # Test411 speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')412 speaker.evpn_prefix_del(413 route_type=route_type,414 route_dist=route_dist,415 esi=esi,416 ethernet_tag_id=ethernet_tag_id,417 )418 # Check419 mock_call.assert_called_with(420 'evpn_prefix.delete_local', **expected_kwargs)421 @mock.patch('ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',422 mock.MagicMock(return_value=None))423 @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')424 def test_evpn_prefix_del_mac_ip_adv(self, mock_call):425 # Prepare test data426 route_type = bgpspeaker.EVPN_MAC_IP_ADV_ROUTE427 route_dist = '65000:100'428 ethernet_tag_id = 200429 mac_addr = 'aa:bb:cc:dd:ee:ff'430 ip_addr = '192.168.0.1'431 expected_kwargs = {432 'route_type': route_type,433 'route_dist': route_dist,434 'ethernet_tag_id': ethernet_tag_id,435 'mac_addr': mac_addr,436 'ip_addr': ip_addr,437 }438 # Test439 speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')440 speaker.evpn_prefix_del(441 route_type=route_type,442 route_dist=route_dist,443 ethernet_tag_id=ethernet_tag_id,444 mac_addr=mac_addr,445 ip_addr=ip_addr,446 )447 # Check448 mock_call.assert_called_with(449 'evpn_prefix.delete_local', **expected_kwargs)450 @mock.patch('ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',451 mock.MagicMock(return_value=None))452 @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')453 def test_evpn_prefix_del_multicast_etag(self, mock_call):454 # Prepare test data455 route_type = bgpspeaker.EVPN_MULTICAST_ETAG_ROUTE456 route_dist = '65000:100'457 esi = 0 # denotes single-homed458 ethernet_tag_id = 200459 mac_addr = 'aa:bb:cc:dd:ee:ff'460 ip_addr = '192.168.0.1'461 expected_kwargs = {462 'route_type': route_type,463 'route_dist': route_dist,464 # 'esi': esi, # should be ignored465 'ethernet_tag_id': ethernet_tag_id,466 # 'mac_addr': mac_addr, # should be ignored467 'ip_addr': ip_addr,468 }469 # Test470 speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')471 speaker.evpn_prefix_del(472 route_type=route_type,473 route_dist=route_dist,474 esi=esi,475 ethernet_tag_id=ethernet_tag_id,476 mac_addr=mac_addr,477 ip_addr=ip_addr,478 )479 # Check480 mock_call.assert_called_with(481 'evpn_prefix.delete_local', **expected_kwargs)482 @raises(ValueError)483 @mock.patch('ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',484 mock.MagicMock(return_value=None))485 @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')486 def test_evpn_prefix_del_invalid_route_type(self, mock_call):487 # Prepare test data488 route_type = 'foobar' # Invalid EVPN route type489 route_dist = '65000:100'490 esi = 0 # denotes single-homed491 ethernet_tag_id = 200492 mac_addr = 'aa:bb:cc:dd:ee:ff'493 ip_addr = '192.168.0.1'494 # Test495 speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')496 speaker.evpn_prefix_del(497 route_type=route_type,498 route_dist=route_dist,499 esi=esi,500 ethernet_tag_id=ethernet_tag_id,501 mac_addr=mac_addr,502 ip_addr=ip_addr,503 )504 # Check505 mock_call.assert_called_with(506 'evpn_prefix.delete_local', 'Invalid arguments detected')507 @mock.patch(508 'ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',509 mock.MagicMock(return_value=None))510 @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')511 def test_evpn_prefix_del_eth_segment(self, mock_call):512 # Prepare test data513 route_type = bgpspeaker.EVPN_ETH_SEGMENT514 route_dist = '65000:100'515 esi = {516 'esi_type': ESI_TYPE_MAC_BASED,517 'mac_addr': 'aa:bb:cc:dd:ee:ff',518 'local_disc': 100,519 }520 ip_addr = '192.168.0.1'521 expected_kwargs = {522 'route_type': route_type,523 'route_dist': route_dist,524 'esi': esi,525 'ip_addr': ip_addr,526 }527 # Test528 speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')529 speaker.evpn_prefix_del(530 route_type=route_type,531 route_dist=route_dist,532 esi=esi,533 ip_addr=ip_addr,534 )535 # Check536 mock_call.assert_called_with(537 'evpn_prefix.delete_local', **expected_kwargs)538 @mock.patch(539 'ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',540 mock.MagicMock(return_value=None))541 @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')542 def test_evpn_prefix_del_ip_prefix_route(self, mock_call):543 # Prepare test data544 route_type = bgpspeaker.EVPN_IP_PREFIX_ROUTE545 route_dist = '65000:100'546 ethernet_tag_id = 200547 ip_prefix = '192.168.0.0/24'548 expected_kwargs = {549 'route_type': route_type,550 'route_dist': route_dist,551 'ethernet_tag_id': ethernet_tag_id,552 'ip_prefix': ip_prefix,553 }554 # Test555 speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')556 speaker.evpn_prefix_del(...
OverlapGraph.py
Source:OverlapGraph.py
1import os2import numpy as np3from tqdm import tqdm4import matplotlib5matplotlib.use('Agg')6import matplotlib.pyplot as plt7from PIL import Image8class OverlapGraph():9 # æå®ã®å½¢å¼ã«ãã©ããã³ã°çµæãæ´å½¢ãã¦åºåããã¯ã©ã¹10 def __init__(self, data_list, route_list):11 # ã³ã³ã¹ãã©ã¯ã¿12 # args:13 self.data_list = data_list14 self.route_list = route_list15 # maxrange : ã°ã©ãã®æç»ç¯å²(±[μm])16 self.maxrange = 317 # ãã®ä»18 self.all_node_list = []19 self.all_node_id = []20 self.devided_route = []21 self.colors = []22 def setMaxRange(self, value):23 self.maxrange = value24 def route_extractor(self, threshold):25 # ãã©ããã³ã°çµæããé¾å¤ã®ãã¬ã¼ã é£ç¶ãã¦ãã©ããã³ã°ããã¦ããçµè·¯ãæ½åº26 # æå¹çµè·¯ã®æ½åº27 effective_route = [28 route for route in self.route_list29 if route.route_length > threshold30 ]31 # å
¨Nodeã¯ã©ã¹ã®é£ç¶é
åãä½ã32 self.all_node_list = [node for data in self.data_list for node in data]33 self.all_node_id = [node.ID for node in self.all_node_list]34 # çµè·¯å§ç¹ããé¾å¤ã¾ã§ã®ãã¼ãIDã¨TIMEã®ãªã¹ããä½ã35 self.devided_route = [route.route[:threshold] for route in effective_route]36 self.devided_route_time = [37 [t for t in range(route.t, route.t+10)]38 for route in effective_route39 ]40 # ã©ã³ãã ã«è²ãæå¹çµè·¯ã®åæ°åçæï¼å¤0ï½255ã®ç¯å²ã§nè¡3åã®ã©ã³ãã ãªndarrayãçæï¼41 self.colors = np.random.rand(len(effective_route), 3)42 def figout(self, out_path, *, pix2m=1):43 # æ½åºããå
¨çµè·¯ãéããããã¦è¡¨ç¤ºããï¼44 print('\nRoute overlapping graph making now ...')45 # åºåå
ä½æ46 print('> Check output dirctory: \"' + out_path + '\"')47 if os.path.isdir(out_path) is False:48 print('> Output directory are not found.')49 os.makedirs(out_path, exist_ok=True)50 print('> New directory created!')51 else:52 print('> Already exist.')53 print('> Calculating ...')54 # æç»ç¨ã®åº§æ¨ãªã¹ããä½ã55 point_list = [56 [57 [self.all_node_list[ID].x, self.all_node_list[ID].y]58 for ID in route59 ] for route in self.devided_route60 ]61 # å§ç¹ã®åº§æ¨ã(0, 0)ã«ãã¦å
¨ä½ãå¹³è¡ç§»åãã62 moved_point_list = [63 [64 np.array(np.array(node) - np.array(route[0])).tolist()65 for node in route66 ] for route in point_list67 ]68 # æ°å¤ãå®éã®è·é¢ã«åããã69 real_scale_list = np.array(moved_point_list) * pix2m70 # ã°ã©ãã®çæ71 fig = plt.figure(figsize=(5, 5))72 ax = fig.add_subplot(1, 1, 1)73 for r_idx, route in enumerate(real_scale_list):74 for i in range(1, len(route)):75 parent_x = route[i-1][0]76 parent_y = -route[i-1][1]77 child_x = route[i][0]78 child_y = -route[i][1]79 # 注æï¼ç»åã®è¦ãæ¹ã¨å®éã®å¤ã®é½é½¬ã解æ¶ããããã«ï¼y軸ã¯ç¬¦å·å転ãã80 ax.plot(81 [parent_x, child_x], [parent_y, child_y],82 c=self.colors[r_idx].tolist())83 # å
¨çµè·¯ã®æ大å¤(+ä½ç½)ãæ±ãã¦ï¼ã°ã©ãæ ã®ãµã¤ãºã決ãã84 # max_range = np.absolute(real_scale_list).max() + pix2m85 # plt.xlim([-max_range, max_range])86 # plt.ylim([-max_range, max_range])87 # ã°ã©ãæ ã®ãµã¤ãºã3umã«è¨å®ãã88 w = self.maxrange89 plt.xlim([-w, w])90 plt.ylim([-w, w])91 # ã°ã©ãã®ä¿å92 filename = out_path + '/overlapped_route.png'93 plt.savefig(filename, dpi=100, transparent=True)94 plt.close()95 print('> Done.')96 return moved_point_list, real_scale_list97 def trajout(self, out_path, *, pix2m=1):98 # æ½åºããå
¨çµè·¯ãåå¥ã«ä¸¦ã¹ã¦åºåããï¼99 print('\nRoute trajectory graph making now ...')100 # åºåå
ä½æ101 out_path = out_path + 'traj_fig/'102 print('> Check output dirctory: \"'+ out_path + '\"')103 if os.path.isdir(out_path) is False:104 print('> Output directory are not found.')105 os.makedirs(out_path, exist_ok=True)106 print('> New directory created!')107 else:108 print('> Already exist.')109 print('> Calculating ...')110 # æç»ç¨ã®åº§æ¨ãªã¹ããä½ã111 point_list = [112 [113 [self.all_node_list[ID].x, self.all_node_list[ID].y]114 for ID in route115 ] for route in self.devided_route116 ]117 # å§ç¹ã®åº§æ¨ã(0, 0)ã«ãã¦å
¨ä½ãå¹³è¡ç§»åãã118 moved_point_list = [119 [120 np.array(np.array(node) - np.array(route[0])).tolist()121 for node in route122 ] for route in point_list123 ]124 # æ°å¤ãå®éã®è·é¢ã«åããã125 real_scale_list = np.array(moved_point_list) * pix2m126 # å
¨çµè·¯ã®æ大å¤(+ä½ç½)ãæ±ãã¦ï¼ã°ã©ãæ ã®ãµã¤ãºã決ãã127 max_range = np.absolute(real_scale_list).max() + pix2m128 # ããã°ã¬ã¹ãã¼ã®çæ129 with tqdm(total=len(real_scale_list), ascii=True, ncols=100) as pbar:130 for r_idx, route in enumerate(real_scale_list):131 # ããã°ã¬ã¹ãã¼ãé²ãã132 pbar.update(1)133 # ã°ã©ãã®çæ134 fig = plt.figure(figsize=(2, 2))135 ax = fig.add_subplot(1, 1, 1)136 for i in range(1, len(route)):137 parent_x = route[i-1][0]138 parent_y = -route[i-1][1]139 child_x = route[i][0]140 child_y = -route[i][1]141 # 注æï¼ç»åã®è¦ãæ¹ã¨å®éã®å¤ã®é½é½¬ã解æ¶ããããã«ï¼y軸ã¯ç¬¦å·å転ãã142 ax.plot(143 [parent_x, child_x], [parent_y, child_y],144 c=self.colors[r_idx].tolist())145 # ã°ã©ãã®æç»ç¯å²ã®è¨å®146 plt.xlim([-max_range, max_range])147 plt.ylim([-max_range, max_range])148 # ã°ã©ãæ ã®ãµã¤ãºã3umã«è¨å®ãã149 # w = self.maxrange150 # plt.xlim([-w, w])151 # plt.ylim([-w, w])152 # ã°ã©ãã®ä¿å153 filename = out_path + '/route_%05d.png' % r_idx154 plt.savefig(filename, dpi=100, transparent=True)155 plt.close()156 print('> Done.')157 def movout(self, out_path, bg_path):158 # ãã©ããã³ã°çµæããªã¼ãã¼ã¬ã¤æç»ãã¦åºåãã159 print('\nRoute trajectory movie making now ...')160 # åºåå
ä½æ161 out_path = out_path + 'mov_fig/'162 print('> Check output dirctory: \"' + out_path + '\"')163 if os.path.isdir(out_path) is False:164 print('> Output directory are not found.')165 os.makedirs(out_path, exist_ok=True)166 print('> New directory created!')167 else:168 print('> Already exist.')169 print('> Calculating ...')170 # ããã°ã¬ã¹ãã¼ã®çæ171 with tqdm(total=len(self.data_list), ascii=True, ncols=100) as pbar:172 # ãã¬ã¼ã ãã¨ã«ã«ã¼ãå¦çãã173 for idx, nodes in enumerate(self.data_list):174 # ããã°ã¬ã¹ãã¼ãé²ãã175 pbar.update(1)176 # èæ¯ã®èªã¿è¾¼ã¿177 img = Image.open(bg_path % idx)178 img = np.array(img)179 h = img.shape[0]180 w = img.shape[1]181 # ã°ã©ãã®çæ182 fig = plt.figure(figsize=(w/10, h/10))183 ax = fig.add_subplot(1, 1, 1)184 # ç»åã®ã«ã©ã¼/ã°ã¬ã¼ã¹ã±ã¼ã«å¤å®185 if len(img.shape) is 2:186 ax.imshow(img, 'gray')187 print('> input image is gray scale.')188 else:189 ax.imshow(img)190 # print('> input image is RGB color')191 # æç»ç¨ã®åº§æ¨ãªã¹ããä½ã192 for r_idx, (T, route) in enumerate(zip(self.devided_route_time, self.devided_route)):193 if idx in T:194 index = T.index(idx)195 for i in reversed(range(index+1)):196 ID = route[i]197 node = self.all_node_list[ID]198 if node.parent_ID is not None:199 parent = self.all_node_list[node.parent_ID]200 child_x = node.x201 child_y = node.y202 parent_x = parent.x203 parent_y = parent.y204 ax.plot(205 [parent_x, child_x], [parent_y, child_y],206 c=self.colors[r_idx].tolist()207 )208 # ãã¼ãã®æç»209 for i, node in enumerate(nodes):210 ax.scatter(node.x, node.y, c='red', s=200, marker='*', edgecolors='black', linewidths=1)211 # ç¯å²è¨å®212 plt.xlim([0, w-1])213 plt.ylim([h-1, 0])214 filename = out_path + '%05d.tif' % idx215 plt.savefig(filename, dpi=100)216 plt.close()...
test_v_s_route_upstream_tls.py
Source:test_v_s_route_upstream_tls.py
1import requests2import pytest3from kubernetes.client.rest import ApiException4from settings import TEST_DATA5from suite.custom_assertions import assert_event_and_get_count, assert_event_count_increased, assert_response_codes, \6 assert_event, assert_no_new_events7from suite.custom_resources_utils import get_vs_nginx_template_conf, patch_v_s_route_from_yaml8from suite.resources_utils import create_items_from_yaml, get_first_pod_name, \9 delete_items_from_yaml, wait_until_all_pods_are_ready, wait_before_test, get_events10@pytest.fixture(scope="class")11def v_s_route_secure_app_setup(request, kube_apis, v_s_route_setup) -> None:12 """13 Prepare a secure example app for Virtual Server Route.14 1st namespace with backend1-svc and backend3-svc and deployment15 and 2nd namespace with https backend2-svc and deployment.16 :param request: internal pytest fixture17 :param kube_apis: client apis18 :param v_s_route_setup:19 :return:20 """21 print("---------------------- Deploy a VS Route Example Application ----------------------------")22 create_items_from_yaml(kube_apis,23 f"{TEST_DATA}/common/app/vsr/secure/multiple.yaml", v_s_route_setup.route_m.namespace)24 create_items_from_yaml(kube_apis,25 f"{TEST_DATA}/common/app/vsr/secure/single.yaml", v_s_route_setup.route_s.namespace)26 wait_until_all_pods_are_ready(kube_apis.v1, v_s_route_setup.route_m.namespace)27 wait_until_all_pods_are_ready(kube_apis.v1, v_s_route_setup.route_s.namespace)28 def fin():29 print("Clean up the Application:")30 delete_items_from_yaml(kube_apis,31 f"{TEST_DATA}/common/app/vsr/secure/multiple.yaml",32 v_s_route_setup.route_m.namespace)33 delete_items_from_yaml(kube_apis,34 f"{TEST_DATA}/common/app/vsr/secure/single.yaml",35 v_s_route_setup.route_s.namespace)36 request.addfinalizer(fin)37@pytest.mark.vsr38@pytest.mark.parametrize('crd_ingress_controller, v_s_route_setup',39 [({"type": "complete", "extra_args": [f"-enable-custom-resources"]},40 {"example": "virtual-server-route-upstream-tls"})],41 indirect=True)42class TestVSRouteUpstreamTls:43 def test_responses_and_config_after_setup(self, kube_apis, ingress_controller_prerequisites,44 crd_ingress_controller, v_s_route_setup, v_s_route_secure_app_setup):45 ic_pod_name = get_first_pod_name(kube_apis.v1, ingress_controller_prerequisites.namespace)46 config = get_vs_nginx_template_conf(kube_apis.v1,47 v_s_route_setup.namespace,48 v_s_route_setup.vs_name,49 ic_pod_name,50 ingress_controller_prerequisites.namespace)51 req_url = f"http://{v_s_route_setup.public_endpoint.public_ip}:{v_s_route_setup.public_endpoint.port}"52 resp_1 = requests.get(f"{req_url}{v_s_route_setup.route_m.paths[0]}",53 headers={"host": v_s_route_setup.vs_host})54 resp_2 = requests.get(f"{req_url}{v_s_route_setup.route_s.paths[0]}",55 headers={"host": v_s_route_setup.vs_host})56 vs_line = f"vs_{v_s_route_setup.namespace}_{v_s_route_setup.vs_name}"57 proxy_host_s = f"{vs_line}_vsr_{v_s_route_setup.route_s.namespace}_{v_s_route_setup.route_s.name}"58 proxy_host_m = f"{vs_line}_vsr_{v_s_route_setup.route_m.namespace}_{v_s_route_setup.route_m.name}"59 assert f'proxy_pass https://{proxy_host_m}' not in config60 assert f'proxy_pass https://{proxy_host_s}' in config61 assert_response_codes(resp_1, resp_2)62 def test_events_after_setup(self, kube_apis, ingress_controller_prerequisites,63 crd_ingress_controller, v_s_route_setup, v_s_route_secure_app_setup):64 text_s = f"{v_s_route_setup.route_s.namespace}/{v_s_route_setup.route_s.name}"65 text_m = f"{v_s_route_setup.route_m.namespace}/{v_s_route_setup.route_m.name}"66 text_vs = f"{v_s_route_setup.namespace}/{v_s_route_setup.vs_name}"67 vsr_s_event_text = f"Configuration for {text_s} was added or updated"68 vsr_m_event_text = f"Configuration for {text_m} was added or updated"69 vs_event_text = f"Configuration for {text_vs} was added or updated"70 events_ns_m = get_events(kube_apis.v1, v_s_route_setup.route_m.namespace)71 events_ns_s = get_events(kube_apis.v1, v_s_route_setup.route_s.namespace)72 assert_event(vsr_s_event_text, events_ns_s)73 assert_event(vsr_m_event_text, events_ns_m)74 assert_event(vs_event_text, events_ns_m)75 def test_validation_flow(self, kube_apis, ingress_controller_prerequisites,76 crd_ingress_controller,77 v_s_route_setup, v_s_route_secure_app_setup):78 ic_pod_name = get_first_pod_name(kube_apis.v1, ingress_controller_prerequisites.namespace)79 initial_events_ns_m = get_events(kube_apis.v1, v_s_route_setup.route_m.namespace)80 initial_events_ns_s = get_events(kube_apis.v1, v_s_route_setup.route_s.namespace)81 try:82 patch_v_s_route_from_yaml(kube_apis.custom_objects,83 v_s_route_setup.route_s.name,84 f"{TEST_DATA}/virtual-server-route-upstream-tls/route-single-invalid.yaml",85 v_s_route_setup.route_s.namespace)86 except ApiException as ex:87 assert ex.status == 422 and "spec.upstreams.tls.enable" in ex.body88 except Exception as ex:89 pytest.fail(f"An unexpected exception is raised: {ex}")90 else:91 pytest.fail("Expected an exception but there was none")92 wait_before_test(1)93 config = get_vs_nginx_template_conf(kube_apis.v1,94 v_s_route_setup.namespace,95 v_s_route_setup.vs_name,96 ic_pod_name,97 ingress_controller_prerequisites.namespace)98 req_url = f"http://{v_s_route_setup.public_endpoint.public_ip}:{v_s_route_setup.public_endpoint.port}"99 resp_1 = requests.get(f"{req_url}{v_s_route_setup.route_m.paths[0]}",100 headers={"host": v_s_route_setup.vs_host})101 resp_2 = requests.get(f"{req_url}{v_s_route_setup.route_s.paths[0]}",102 headers={"host": v_s_route_setup.vs_host})103 new_events_ns_m = get_events(kube_apis.v1, v_s_route_setup.route_m.namespace)104 new_events_ns_s = get_events(kube_apis.v1, v_s_route_setup.route_s.namespace)105 vs_line = f"vs_{v_s_route_setup.namespace}_{v_s_route_setup.vs_name}"106 proxy_host_s = f"{vs_line}_vsr_{v_s_route_setup.route_s.namespace}_{v_s_route_setup.route_s.name}"107 proxy_host_m = f"{vs_line}_vsr_{v_s_route_setup.route_m.namespace}_{v_s_route_setup.route_m.name}"108 assert f'proxy_pass https://{proxy_host_m}' not in config109 assert f'proxy_pass https://{proxy_host_s}' in config110 assert_response_codes(resp_1, resp_2)111 assert_no_new_events(initial_events_ns_m, new_events_ns_m)112 assert_no_new_events(initial_events_ns_s, new_events_ns_s)113 def test_responses_and_config_after_disable_tls(self, kube_apis, ingress_controller_prerequisites,114 crd_ingress_controller,115 v_s_route_setup, v_s_route_secure_app_setup):116 ic_pod_name = get_first_pod_name(kube_apis.v1, ingress_controller_prerequisites.namespace)117 text_s = f"{v_s_route_setup.route_s.namespace}/{v_s_route_setup.route_s.name}"118 text_m = f"{v_s_route_setup.route_m.namespace}/{v_s_route_setup.route_m.name}"119 text_vs = f"{v_s_route_setup.namespace}/{v_s_route_setup.vs_name}"120 vsr_s_event_text = f"Configuration for {text_s} was added or updated"121 vsr_m_event_text = f"Configuration for {text_m} was added or updated"122 vs_event_text = f"Configuration for {text_vs} was added or updated"123 initial_events_ns_m = get_events(kube_apis.v1, v_s_route_setup.route_m.namespace)124 initial_events_ns_s = get_events(kube_apis.v1, v_s_route_setup.route_s.namespace)125 initial_count_vsr_m = assert_event_and_get_count(vsr_m_event_text, initial_events_ns_m)126 initial_count_vsr_s = assert_event_and_get_count(vsr_s_event_text, initial_events_ns_s)127 initial_count_vs = assert_event_and_get_count(vs_event_text, initial_events_ns_m)128 patch_v_s_route_from_yaml(kube_apis.custom_objects,129 v_s_route_setup.route_s.name,130 f"{TEST_DATA}/virtual-server-route-upstream-tls/route-single-disable-tls.yaml",131 v_s_route_setup.route_s.namespace)132 wait_before_test(1)133 config = get_vs_nginx_template_conf(kube_apis.v1,134 v_s_route_setup.namespace,135 v_s_route_setup.vs_name,136 ic_pod_name,137 ingress_controller_prerequisites.namespace)138 req_url = f"http://{v_s_route_setup.public_endpoint.public_ip}:{v_s_route_setup.public_endpoint.port}"139 resp_1 = requests.get(f"{req_url}{v_s_route_setup.route_m.paths[0]}",140 headers={"host": v_s_route_setup.vs_host})141 resp_2 = requests.get(f"{req_url}{v_s_route_setup.route_s.paths[0]}",142 headers={"host": v_s_route_setup.vs_host})143 new_events_ns_m = get_events(kube_apis.v1, v_s_route_setup.route_m.namespace)144 new_events_ns_s = get_events(kube_apis.v1, v_s_route_setup.route_s.namespace)145 assert 'proxy_pass https://' not in config146 assert_response_codes(resp_1, resp_2, 200, 400)147 assert_event_count_increased(vsr_m_event_text, initial_count_vsr_m, new_events_ns_m)148 assert_event_count_increased(vs_event_text, initial_count_vs, new_events_ns_m)...
test_v_s_route_status.py
Source:test_v_s_route_status.py
1import pytest2from kubernetes.client.rest import ApiException3from suite.resources_utils import wait_before_test4from suite.custom_resources_utils import (5 read_crd,6 patch_virtual_server_from_yaml,7 patch_v_s_route_from_yaml,8 delete_virtual_server,9 create_virtual_server_from_yaml,10)11from settings import TEST_DATA12@pytest.mark.vsr13@pytest.mark.parametrize(14 "crd_ingress_controller, v_s_route_setup",15 [16 (17 {"type": "complete", "extra_args": [f"-enable-custom-resources", f"-enable-leader-election=false"],},18 {"example": "virtual-server-route-status"},19 )20 ],21 indirect=True,22)23class TestVirtualServerRouteStatus:24 def patch_valid_vsr(self, kube_apis, v_s_route_setup) -> None:25 """26 Function to revert vsr deployments to valid state27 """28 patch_src_m = f"{TEST_DATA}/virtual-server-route-status/route-multiple.yaml"29 patch_v_s_route_from_yaml(30 kube_apis.custom_objects,31 v_s_route_setup.route_m.name,32 patch_src_m,33 v_s_route_setup.route_m.namespace,34 )35 wait_before_test()36 patch_src_s = f"{TEST_DATA}/virtual-server-route-status/route-single.yaml"37 patch_v_s_route_from_yaml(38 kube_apis.custom_objects,39 v_s_route_setup.route_s.name,40 patch_src_s,41 v_s_route_setup.route_s.namespace,42 )43 wait_before_test()44 def patch_valid_vs(self, kube_apis, v_s_route_setup) -> None:45 """46 Function to revert vs deployment to valid state47 """48 patch_src = f"{TEST_DATA}/virtual-server-route-status/standard/virtual-server.yaml"49 patch_virtual_server_from_yaml(50 kube_apis.custom_objects, v_s_route_setup.vs_name, patch_src, v_s_route_setup.namespace,51 )52 wait_before_test()53 @pytest.mark.smoke54 def test_status_valid(55 self, kube_apis, crd_ingress_controller, v_s_route_setup, v_s_route_app_setup56 ):57 """58 Test VirtualServerRoute status with a valid fields in yaml59 """60 response_m = read_crd(61 kube_apis.custom_objects,62 v_s_route_setup.route_m.namespace,63 "virtualserverroutes",64 v_s_route_setup.route_m.name,65 )66 response_s = read_crd(67 kube_apis.custom_objects,68 v_s_route_setup.route_s.namespace,69 "virtualserverroutes",70 v_s_route_setup.route_s.name,71 )72 assert (73 response_m["status"]74 and response_m["status"]["reason"] == "AddedOrUpdated"75 and response_m["status"]["referencedBy"]76 and response_m["status"]["state"] == "Valid"77 )78 assert (79 response_s["status"]80 and response_s["status"]["reason"] == "AddedOrUpdated"81 and response_s["status"]["referencedBy"]82 and response_s["status"]["state"] == "Valid"83 )84 def test_status_invalid(85 self, kube_apis, crd_ingress_controller, v_s_route_setup, v_s_route_app_setup86 ):87 """88 Test VirtualServerRoute status with a invalid paths in vsr yaml89 """90 patch_src_m = f"{TEST_DATA}/virtual-server-route-status/route-multiple-invalid.yaml"91 patch_v_s_route_from_yaml(92 kube_apis.custom_objects,93 v_s_route_setup.route_m.name,94 patch_src_m,95 v_s_route_setup.route_m.namespace,96 )97 wait_before_test()98 patch_src_s = f"{TEST_DATA}/virtual-server-route-status/route-single-invalid.yaml"99 patch_v_s_route_from_yaml(100 kube_apis.custom_objects,101 v_s_route_setup.route_s.name,102 patch_src_s,103 v_s_route_setup.route_s.namespace,104 )105 wait_before_test()106 response_m = read_crd(107 kube_apis.custom_objects,108 v_s_route_setup.route_m.namespace,109 "virtualserverroutes",110 v_s_route_setup.route_m.name,111 )112 response_s = read_crd(113 kube_apis.custom_objects,114 v_s_route_setup.route_s.namespace,115 "virtualserverroutes",116 v_s_route_setup.route_s.name,117 )118 self.patch_valid_vsr(kube_apis, v_s_route_setup)119 assert (120 response_m["status"]121 and response_m["status"]["reason"] == "Rejected"122 and not response_m["status"]["referencedBy"]123 and response_m["status"]["state"] == "Invalid"124 )125 assert (126 response_s["status"]127 and response_s["status"]["reason"] == "Rejected"128 and not response_s["status"]["referencedBy"]129 and response_s["status"]["state"] == "Invalid"130 )131 132 def test_status_invalid_prefix(133 self, kube_apis, crd_ingress_controller, v_s_route_setup, v_s_route_app_setup134 ):135 """136 Test VirtualServerRoute status with a invalid path /prefix in vsr yaml137 i.e. referring to non-existing path138 """139 patch_src_m = f"{TEST_DATA}/virtual-server-route-status/route-multiple-invalid-prefixed-path.yaml"140 patch_v_s_route_from_yaml(141 kube_apis.custom_objects,142 v_s_route_setup.route_m.name,143 patch_src_m,144 v_s_route_setup.route_m.namespace,145 )146 wait_before_test()147 patch_src_s = f"{TEST_DATA}/virtual-server-route-status/route-single-invalid-prefixed-path.yaml"148 patch_v_s_route_from_yaml(149 kube_apis.custom_objects,150 v_s_route_setup.route_s.name,151 patch_src_s,152 v_s_route_setup.route_s.namespace,153 )154 wait_before_test()155 response_m = read_crd(156 kube_apis.custom_objects,157 v_s_route_setup.route_m.namespace,158 "virtualserverroutes",159 v_s_route_setup.route_m.name,160 )161 response_s = read_crd(162 kube_apis.custom_objects,163 v_s_route_setup.route_s.namespace,164 "virtualserverroutes",165 v_s_route_setup.route_s.name,166 )167 self.patch_valid_vsr(kube_apis, v_s_route_setup)168 assert (169 response_m["status"]170 and response_m["status"]["reason"] == "AddedOrUpdated"171 and response_m["status"]["referencedBy"]172 and response_m["status"]["state"] == "Valid"173 )174 assert (175 response_s["status"]176 and response_s["status"]["reason"] == "Ignored"177 and not response_s["status"]["referencedBy"]178 and response_s["status"]["state"] == "Warning"179 )180 def test_status_invalid_vsr_in_vs(181 self, kube_apis, crd_ingress_controller, v_s_route_setup, v_s_route_app_setup182 ):183 """184 Test VirtualServerRoute status with invalid vsr reference in vs yaml185 """186 patch_src = f"{TEST_DATA}/virtual-server-route-status/virtual-server-invalid.yaml"187 patch_virtual_server_from_yaml(188 kube_apis.custom_objects, v_s_route_setup.vs_name, patch_src, v_s_route_setup.namespace,189 )190 wait_before_test()191 response_m = read_crd(192 kube_apis.custom_objects,193 v_s_route_setup.route_m.namespace,194 "virtualserverroutes",195 v_s_route_setup.route_m.name,196 )197 response_s = read_crd(198 kube_apis.custom_objects,199 v_s_route_setup.route_s.namespace,200 "virtualserverroutes",201 v_s_route_setup.route_s.name,202 )203 self.patch_valid_vs(kube_apis, v_s_route_setup)204 assert (205 response_m["status"]206 and response_m["status"]["reason"] == "Ignored"207 and not response_m["status"]["referencedBy"]208 and response_m["status"]["state"] == "Warning"209 )210 assert (211 response_s["status"]212 and response_s["status"]["reason"] == "Ignored"213 and not response_s["status"]["referencedBy"]214 and response_s["status"]["state"] == "Warning"215 )216 def test_status_remove_vs(217 self, kube_apis, crd_ingress_controller, v_s_route_setup, v_s_route_app_setup218 ):219 """220 Test VirtualServerRoute status after deleting referenced VirtualServer221 """222 delete_virtual_server(223 kube_apis.custom_objects, v_s_route_setup.vs_name, v_s_route_setup.namespace,224 )225 response_m = read_crd(226 kube_apis.custom_objects,227 v_s_route_setup.route_m.namespace,228 "virtualserverroutes",229 v_s_route_setup.route_m.name,230 )231 response_s = read_crd(232 kube_apis.custom_objects,233 v_s_route_setup.route_s.namespace,234 "virtualserverroutes",235 v_s_route_setup.route_s.name,236 )237 vs_src = f"{TEST_DATA}/virtual-server-route-status/standard/virtual-server.yaml"238 create_virtual_server_from_yaml(kube_apis.custom_objects, vs_src, v_s_route_setup.namespace)239 assert (240 response_m["status"]241 and response_m["status"]["reason"] == "NoVirtualServerFound"242 and not response_m["status"]["referencedBy"]243 and response_m["status"]["state"] == "Warning"244 )245 assert (246 response_s["status"]247 and response_s["status"]["reason"] == "NoVirtualServerFound"248 and not response_s["status"]["referencedBy"]249 and response_s["status"]["state"] == "Warning"...
main.py
Source:main.py
1from RouteLocalization import *2from PublicValue import *3from DBControl import DBControl4from RouteInfo import RouteInfo5from RouteStation import RouteStation6import operator7##ì ì í ì¤ê° ê²½ì ì§ ì ì í¨ì8def selectMidStation(stationDict):9 priorityKeyword={"(ê²½ì )":-100, "리": 1, "ë": 1, "íµ": 1, "íêµ": 2, "ì¬ë ¹ë¶":2, "경찰ì":2, "ì¬ë¬´":3, "주민":3}10 seqIndexs = list(stationDict.keys())#í¤ ê° ë¦¬ì¤í¸11 midIndex = int((int(seqIndexs[-1]) + int(seqIndexs[0]))/2) #í¤ ê° ì ëì ì¤ê° ê°12 endIndex = midIndex*213 if stationDict[seqIndexs[-1]][-1]=='N': # ë§ì§ë§ í목 ì ë¥ì¥ì´ ë°íì ì´ ìëë©´ 1/4ì§ì ì¼ë¡14 midIndex /= 215 endIndex /= 216 midIndex = round(midIndex)17 endIndex = round(endIndex)18 fivePerIndex = round(endIndex * 0.05)19 '''20 five > 0.521 end * 0.05 > 0.522 end * 23 '''24 if endIndex<=10:25 fivePerIndex = int(midIndex/2)26 print("[fivePerIndex]"+str(fivePerIndex), "[midIndex]"+str(midIndex), "[endIndex]"+str(endIndex))27 propertyIndex = {}28 for i in range(midIndex-fivePerIndex, midIndex+fivePerIndex+1):29 propertyIndex[i]=030 for keyword in priorityKeyword.keys():31 if keyword in stationDict[str(i)][5]:32 propertyIndex[i]+=priorityKeyword[keyword]33 propertyIndex = sorted(propertyIndex.items(), key=operator.itemgetter(1))34 answerIndex = len(propertyIndex)-1 # ((a, b), (c, d))35 for i in range(answerIndex-1, -1, -1):36 if propertyIndex[i][1] < propertyIndex[answerIndex][1] :37 answerIndex = i+138 break39 for i in propertyIndex:40 print("%s ,seq: %d, property: %d" % (stationDict[str(i[0])][5], i[0], i[1]))41 return (stationDict[str(propertyIndex[answerIndex][0])][5], stationDict[seqIndexs[-1]][-1])42 '''43 inc=144 while True:45 print(midStaiontionName)46 # if "(ê²½ì )" in midStaiontionName:47 48 if (not "(ê²½ì )" in midStaiontionName) and ( "리" in midStaiontionName or "ë" in midStaiontionName or "íµ" in midStaiontionName or "íêµ" in midStaiontionName or "ë§ì" in midStaiontionName or "ì" in midStaiontionName) :49 break50 else:51 incdec = int(round(inc/2)*pow(-1, inc))52 midStaiontionName = stationDict[str(int(midIndex)+incdec)][5]53 inc+=154 '''55 return midStaiontionName56errorCount = 057#ìë²ì°ê²°58##DB59server = "13.125.133.124"60serverPort = 3306 61dbc = DBControl(server, "root", "005410", "joambusdb", serverPort)62#dbc = DBControl("localhost", "root", "ë¹ë²", "busarrivaldb")63#ë
¸ì 목ë¡64print("ë
¸ì 목ë¡")65routeList = routeLocalization()66if dbc.isThisTable("routelist"):67 dbc.emptyTable("routelist")68else:69 dbc.createTable("routelist", ROUTE_LIST_IN_JOAM_TAG, ROUTE_LIST_IN_JOAM_TAG_BYTE)70for i in range(len(routeList)):71 dbc.addData("routelist", ROUTE_LIST_IN_JOAM_TAG, routeList[i], i)72#ë
¸ì ì ë¥ì¥73print("ë
¸ì ì ë¥ì¥")74if dbc.isThisTable("routestation"):75 dbc.emptyTable("routestation")76else:77 dbc.createTable("routestation", ROUTE_STATION_DB_TAG, ROUTE_STATION_DB_TAG_BYTE)78id = 079routeMidStations = {}80isOneWays = {}81for i in range(len(routeList)):82 print(routeList[i]+"ì ë¥ì¥")83 routeStation = RouteStation(routeList[i]).getStationInfos()84 85 seqIndexs = list(routeStation.keys())86 '''87 midIndex = str(int((int(seqIndexs[-1]) + int(seqIndexs[0]))/2))88 if routeStation[seqIndexs[-1]][-1]=='N':89 midIndex = str(int(int(midIndex)/2))90 midStaiontionName=routeStation[midIndex][5]91 inc=192 while True:93 print(midStaiontionName)94 # if "(ê²½ì )" in midStaiontionName:95 96 if (not "(ê²½ì )" in midStaiontionName) and ( "리" in midStaiontionName or "ë" in midStaiontionName or "íµ" in midStaiontionName or "íêµ" in midStaiontionName or "ë§ì" in midStaiontionName or "ì" in midStaiontionName) :97 break98 else:99 incdec = int(round(inc/2)*pow(-1, inc))100 midStaiontionName = routeStation[str(int(midIndex)+incdec)][5]101 inc+=1102 '''103 routeMidStations[routeList[i]]=selectMidStation(routeStation)104 isOneWays[routeList[i]]=routeMidStations[routeList[i]][1]105 routeMidStations[routeList[i]]=routeMidStations[routeList[i]][0]106 107 108 for seq in seqIndexs:109 stationInfo = [routeList[i], seq]110 stationInfo+=list(routeStation[seq])111 #routeStation.insert(0, routeList[i])112 dbc.addData("routestation", ROUTE_STATION_DB_TAG, tuple(stationInfo), id)113 id+=1114#ë
¸ì ì ë³´115print("ë
¸ì ì ë³´")116if dbc.isThisTable("routeInfo"):117 dbc.emptyTable("routeInfo")118else:119 dbc.createTable("routeInfo", ROUTE_INFO_DB_TAG, ROUTE_INFO_DB_TAG_BYTE)120for i in range(len(routeList)):121 routeInfo = RouteInfo(routeList[i]).getInfoItems()+[routeMidStations[routeList[i]], isOneWays[routeList[i]]]122 print(routeInfo)123 dbc.addData("routeInfo", ROUTE_INFO_DB_TAG, tuple(routeInfo), i)...
route_cache.py
Source:route_cache.py
...28 def __init__(self):29 self.routing_obj = Routing()30 self.routing_obj.set_family("inet6")31 self.route_cache = []32 def remove_route(self, route):33 """Remove a route from the route cache"""34 if route not in self.route_cache:35 return False36 target, nexthop, nexthop_iface = route.to_tuple()37 logger.debug("Remove route to %s through %s on iface %s" % (target, nexthop, nexthop_iface))38 self.routing_obj.remove(target, (nexthop, nexthop_iface), table="local")39 self.route_cache.remove(route)40 return True41 def remove_routes(self, routes):42 """Remove a list of routes from the route cache"""43 route_update = False44 for route in routes:45 route_update += self.remove_route(route)46 return bool(route_update)47 def lookup_nexthop(self, nexthop, target=None):48 """Lookup a next hop in the route cache.49 The point is to make it easier to remove routes going through a specific router"""50 if target:51 return [route for route in self.route_cache if route.nexthop == nexthop and route.target == target]52 else:53 return [route for route in self.route_cache if route.nexthop == nexthop]54 def remove_nexthop(self, nexthop, target=None):55 route_update = False56 for route in self.lookup_nexthop(nexthop, target):57 route_update += self.remove_route(route)58 return bool(route_update)59 def add_route(self, route):60 """add a route to a target through a next hop"""61 if route in self.route_cache:62 return False63 (target, nexthop, nexthop_iface) = route.to_tuple()64 logger.debug("Add route to %s through %s on iface %s" % (target, nexthop, nexthop_iface))65 # sanity check: do not add a route to an address that is assigned on66 # the node67 assert target == "default" or not gv.address_cache.is_assigned(target.split("/")[0])68 try:69 self.routing_obj.add(target, (nexthop, nexthop_iface), table="local")70 except: pass71 self.route_cache.append(route)72 return True73 def add_routes(self, routes):74 """Add a list of routes to the route cache"""75 route_update = False76 for route in routes:77 route_update += self.add_route(route)78 return bool(route_update)79 def empty_cache(self):80 """Empty the route cache"""81 for route in copy(self.route_cache):82 self.remove_route(route)83 assert self.route_cache == []84 def __str__(self):85 """Print the complete route cache"""86 return self.routing_obj.__str__()87class Route(object):88 def __init__(self, target, nexthop, nexthop_iface, onehop=False):89 """Store route information:90 - target is the route target address (it can be a prefix e.g "2000::/3"))91 - nexthop is the next hop to reach target (e.g. "fe80::a3)92 - nexthop_iface is the interface name where nexthop can be reached (e.g. "eth0")93 - onehop indicate if the route is for a direct neighbor"""94 self.target = target95 self.nexthop = nexthop96 self.nexthop_iface = nexthop_iface...
Using AI Code Generation
1const { chromium, webkit, firefox } = require('playwright');2(async () => {3 const browser = await chromium.launch({ headless: false });4 const context = await browser.newContext();5 const page = await context.newPage();6 await page.route('**/*', route => {7 if (route.request().resourceType() === 'image')8 route.abort();9 route.continue();10 });11 await browser.close();12})();
Using AI Code Generation
1const { chromium } = require("playwright");2const { route } = require("playwright/lib/server/network");3(async () => {4 const browser = await chromium.launch();5 const context = await browser.newContext();6 const page = await context.newPage();7 await route("**/json", (route, request) => {8 route.fulfill({9 body: JSON.stringify({ message: "hello world" }),10 });11 });12 await browser.close();13})();
Using AI Code Generation
1const { route } = require('@playwright/test');2module.exports = {3 route('**/*', route => {4 route.fulfill({5 });6 }),7};8const { test } = require('@playwright/test');9test('test', async ({ page }) => {10});11const { test, beforeEach, afterEach } = require('@playwright/test');12beforeEach(async ({ page }) => {13});14afterEach(async ({ page }) => {15 await page.screenshot({ path: 'test-results/test.png' });16});17test('test', async ({ page }) => {18});19const { test } = require('@playwright/test');20test('test', async ({ page }) => {21});22const { test, beforeAll, afterAll } = require('@playwright/test');23beforeAll(async () => {24});25afterAll(async () => {26});27test('test', async ({ page }) => {28});29const { test } = require('@playwright/test');30test('test', async ({ page }) => {31});
Using AI Code Generation
1const { route } = require('@playwright/test');2route('**/*', route => {3 route.fulfill({4 });5});6route('**/*', route => {7 route.fulfill({8 });9});10route('**/*', route => {11 route.fulfill({12 });13});
Using AI Code Generation
1const { route } = require('@playwright/test');2const { test, expect } = require('@playwright/test');3test('My Test', async ({ page }) => {4 await page.route('**/*.{png,jpg,jpeg}', route => route.abort());5 await page.route('**/logo*.png', route => route.fulfill({6 headers: {7 }8 }));9});10const { route } = require('@playwright/test');11const { test, expect } = require('@playwright/test');12test('My Test', async ({ page }) => {13 await page.route('**/*.{png,jpg,jpeg}', route => route.abort());14 await page.route('**/logo*.png', route => route.fulfill({15 headers: {16 }17 }));18});19const { route } = require('@playwright/test');20const { test, expect } = require('@playwright/test');21test('My Test', async ({ page }) => {22 await page.route('**/*.{png,jpg,jpeg}', route => route.abort());23 await page.route('**/logo*.png', route => route.fulfill({24 headers: {25 }26 }));27});28const { route } = require('@playwright/test');29const { test, expect } = require('@playwright/test');30test('My Test', async ({ page }) => {31 await page.route('**/*.{png,jpg,jpeg}',
Using AI Code Generation
1const {test} = require('@playwright/test');2const {route} = require('@playwright/test');3test('test', async ({ page }) => {4 await page.route('**/api/v1/playwright', route => route.fulfill({5 body: JSON.stringify({6 })7 }));8 const response = await page.evaluate(() => fetch('/api/v1/playwright').then(r => r.json()));9 expect(response).toEqual({ success: true });10});11const {test} = require('@playwright/test');12const {route} = require('@playwright/test');13test('test', async ({ page }) => {14 await page.route('**/api/v1/playwright', route => route.fulfill({15 body: JSON.stringify({16 })17 }));18 const response = await page.evaluate(() => fetch('/api/v1/playwright').then(r => r.json()));19 expect(response).toEqual({ success: true });20});21const {test} = require('@playwright/test');22const {route} = require('@playwright/test');23test('test', async ({ page }) => {24 await page.route('**/api/v1/playwright', route => route.fulfill({25 body: JSON.stringify({26 })27 }));28 const response = await page.evaluate(() => fetch('/api/v1/playwright').then(r => r.json()));29 expect(response).toEqual({ success: true });30});31const {test} = require('@playwright/test');32const {route} = require('@playwright/test');33test('test', async ({ page }) => {34 await page.route('**/api/v1/playwright', route => route.fulfill({
Using AI Code Generation
1const { route } = require('@playwright/test');2const route = route('**/api/v1/users', route => route.fulfill({3 headers: {4 },5 body: JSON.stringify({ name: 'John Doe' }),6}));7const { route } = require('@playwright/test');8const route = route('**/api/v1/users', route => route.fulfill({9 headers: {10 },11 body: JSON.stringify({ name: 'John Doe' }),12}));13const { route } = require('@playwright/test');14const route = route('**/api/v1/users', route => route.fulfill({15 headers: {16 },17 body: JSON.stringify({ name: 'John Doe' }),18}));19const { route } = require('@playwright/test');20const route = route('**/api/v1/users', route => route.fulfill({21 headers: {22 },23 body: JSON.stringify({ name: 'John Doe' }),24}));25const { route } = require('@playwright/test');26const route = route('**/api/v1/users', route => route.fulfill({27 headers: {28 },29 body: JSON.stringify({ name: 'John Doe' }),30}));31const { route } = require('@playwright/test');32const route = route('**/api/v1/users', route => route.fulfill({33 headers: {
Using AI Code Generation
1const { route } = require('@playwright/test');2route('**/api/**', (route, request) => {3 route.fulfill({4 body: JSON.stringify({ name: 'John Doe' }),5 });6});7const { route } = require('@playwright/test');8route('**/api/**', (route, request) => {9 route.fulfill({10 body: JSON.stringify({ name: 'John Doe' }),11 });12});13const { route } = require('@playwright/test');14route('**/api/**', (route, request) => {15 route.fulfill({16 body: JSON.stringify({ name: 'John Doe' }),17 });18});19const { route } = require('@playwright/test');20route('**/api/**', (route, request) => {21 route.fulfill({22 body: JSON.stringify({ name: 'John Doe' }),23 });24});25const { route } = require('@playwright/test');26route('**/api/**', (route, request) => {27 route.fulfill({28 body: JSON.stringify({ name: 'John Doe' }),29 });30});31const { route } = require('@play
Playwright tutorial
LambdaTest’s Playwright tutorial will give you a broader idea about the Playwright automation framework, its unique features, and use cases with examples to exceed your understanding of Playwright testing. This tutorial will give A to Z guidance, from installing the Playwright framework to some best practices and advanced concepts.
Chapters:
- What is Playwright : Playwright is comparatively new but has gained good popularity. Get to know some history of the Playwright with some interesting facts connected with it.
- How To Install Playwright : Learn in detail about what basic configuration and dependencies are required for installing Playwright and run a test. Get a step-by-step direction for installing the Playwright automation framework.
- Playwright Futuristic Features: Launched in 2020, Playwright gained huge popularity quickly because of some obliging features such as Playwright Test Generator and Inspector, Playwright Reporter, Playwright auto-waiting mechanism and etc. Read up on those features to master Playwright testing.
- What is Component Testing: Component testing in Playwright is a unique feature that allows a tester to test a single component of a web application without integrating them with other elements. Learn how to perform Component testing on the Playwright automation framework.
- Inputs And Buttons In Playwright: Every website has Input boxes and buttons; learn about testing inputs and buttons with different scenarios and examples.
- Functions and Selectors in Playwright: Learn how to launch the Chromium browser with Playwright. Also, gain a better understanding of some important functions like “BrowserContext,” which allows you to run multiple browser sessions, and “newPage” which interacts with a page.
- Handling Alerts and Dropdowns in Playwright : Playwright interact with different types of alerts and pop-ups, such as simple, confirmation, and prompt, and different types of dropdowns, such as single selector and multi-selector get your hands-on with handling alerts and dropdown in Playright testing.
- Playwright vs Puppeteer: Get to know about the difference between two testing frameworks and how they are different than one another, which browsers they support, and what features they provide.
- Run Playwright Tests on LambdaTest: Playwright testing with LambdaTest leverages test performance to the utmost. You can run multiple Playwright tests in Parallel with the LammbdaTest test cloud. Get a step-by-step guide to run your Playwright test on the LambdaTest platform.
- Playwright Python Tutorial: Playwright automation framework support all major languages such as Python, JavaScript, TypeScript, .NET and etc. However, there are various advantages to Python end-to-end testing with Playwright because of its versatile utility. Get the hang of Playwright python testing with this chapter.
- Playwright End To End Testing Tutorial: Get your hands on with Playwright end-to-end testing and learn to use some exciting features such as TraceViewer, Debugging, Networking, Component testing, Visual testing, and many more.
- Playwright Video Tutorial: Watch the video tutorials on Playwright testing from experts and get a consecutive in-depth explanation of Playwright automation testing.
Run Playwright Internal automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
