KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use del_br method in autotest

Best Python code snippet using autotest_python

BridgeController.py

Source:BridgeController.py

...45    def del_port(self, port_name):46        pass4748    @abstractmethod49    def del_br(self):50        pass5152    @property53    @abstractmethod54    def brctlexe(self,):55        pass565758class OvsBridge(BridgeABC):59    brctlexe = spawn.find_executable("ovs-vsctl")6061    def __init__(self, name, ip_addr, prefix_len, sdn_ctrl_cfg=dict()):62        """ Initialize a bridge object. """63        super(OvsBridge, self).__init__(name, ip_addr, prefix_len)64        ipoplib.runshell_su([OvsBridge.brctlexe,65                             "--may-exist", "add-br", self.name])6667        '''net = "{0}/{1}".format(ip_addr, prefix_len)6869        p = ipoplib.runshell_su([IPEXE, "addr", "show", self.name])70        ip_addr_info = p.stdout.decode()71        if net not in ip_addr_info:72            ipoplib.runshell_su([IPEXE, "addr", "add", net, "dev", self.name])7374        self.stp(False)75        ipoplib.runshell_su([IPEXE, "link", "set", "dev", self.name, "up"])'''7677        self.add_sdn_ctrl(sdn_ctrl_cfg)7879    def add_sdn_ctrl(self, sdn_ctrl_cfg):80        if sdn_ctrl_cfg:81            if sdn_ctrl_cfg["ConnectionType"] == "tcp":82                ctrl_conn_str = ":".join([sdn_ctrl_cfg["ConnectionType"],83                                         sdn_ctrl_cfg["HostName"],84                                         sdn_ctrl_cfg["Port"]])8586                ipoplib.runshell_su([OvsBridge.brctlexe,87                                     "set-controller",88                                     self.name,89                                     ctrl_conn_str])9091    def del_sdn_ctrl(self):92        ipoplib.runshell_su([OvsBridge.brctlexe, "del-controller", self.name])9394    def del_br(self):95        self.del_sdn_ctrl()9697        ipoplib.runshell_su([OvsBridge.brctlexe,98                             "--if-exists", "del-br", self.name])99100    def add_port(self, port_name):101        ipoplib.runshell_su([OvsBridge.brctlexe,102                             "--may-exist", "add-port", self.name, port_name])103104    def del_port(self, port_name):105        ipoplib.runshell_su([OvsBridge.brctlexe,106                             "--if-exists", "del-port", self.name, port_name])107108    def stp(self, enable):109        if enable:110            ipoplib.runshell_su([OvsBridge.brctlexe,111                                 "set", "bridge", self.name,112                                 "stp_enable=true"])113        else:114            ipoplib.runshell_su([OvsBridge.brctlexe,115                                 "set", "bridge", self.name,116                                 "stp_enable=false"])117118119class LinuxBridge(BridgeABC):120    brctlexe = spawn.find_executable("brctl")121122    def __init__(self, name, ip_addr, prefix_len, *args, **kwargs):123        """ Initialize a bridge object. """124125        super(LinuxBridge, self).__init__(name, ip_addr, prefix_len)126127        p = ipoplib.runshell_su([LinuxBridge.brctlexe, 'show'])128        wlist = map(str.split, p.stdout.decode("utf-8").splitlines()[1:])129        brwlist = filter(lambda x: len(x) != 1, wlist)130        brlist = map(lambda x: x[0], brwlist)131        for br in brlist:132            print(br)133            if br == name:134                print("deleting {}".format(br))135                self.del_br()136137        ipoplib.runshell_su([LinuxBridge.brctlexe, "addbr", self.name])138        net = "{0}/{1}".format(ip_addr, prefix_len)139        ipoplib.runshell_su([IPEXE, "addr", "add", net, "dev", name])140        self.stp(False)141        ipoplib.runshell_su([IPEXE, "link", "set", "dev", name, "up"])142143    def __str__(self):144        """ Return a string of the bridge name. """145        return self.name146147    def __repr__(self):148        """ Return a representaion of a bridge object. """149        return "<Bridge: %s>" % self.name150151    def del_br(self):152        """ Set the device down and delete the bridge. """153        ipoplib.runshell_su([IPEXE, "link", "set", "dev", self.name, "down"])154        ipoplib.runshell_su([LinuxBridge.brctlexe, "delbr", self.name])155156    def add_port(self, port):157        ipoplib.runshell_su([LinuxBridge.brctlexe, "addif", self.name, port])158159    def del_port(self, port):160        ipoplib.runshell_su([LinuxBridge.brctlexe, "delif", self.name, port])161162    def stp(self, val=True):163        """ Turn STP protocol on/off. """164165        if val:166            state = "on"167        else:168            state = "off"169        ipoplib.runshell_su([LinuxBridge.brctlexe, "stp", self.name, state])170171    def set_bridge_prio(self, prio):172        """ Set bridge priority value. """173        ipoplib.runshell_su([LinuxBridge.brctlexe,174                             "setbridgeprio", self.name, str(prio)])175176    def set_path_cost(self, port, cost):177        """ Set port path cost value for STP protocol. """178        ipoplib.runshell_su([LinuxBridge.brctlexe,179                             "setpathcost", self.name, port, str(cost)])180181    def set_port_prio(self, port, prio):182        """ Set port priority value. """183        ipoplib.runshell_su([LinuxBridge.brctlexe,184                             "setportprio", self.name, port, str(prio)])185186187class BridgeController(ControllerModule):188    def __init__(self, cfx_handle, module_config,189                 module_name, *args, **kwargs):190        super(BridgeController, self).__init__(cfx_handle, module_config,191                                               module_name)192        self._overlays = dict()193        self._lock = threading.Lock()194195    def initialize(self):196        for olid in self._cm_config["Overlays"]:197            br_cfg = self._cm_config["Overlays"][olid]198199            if self._cm_config["Overlays"][olid]["Type"] == "LXBR":200                self._overlays[olid] = LinuxBridge(br_cfg["BridgeName"],201                                                   br_cfg["IP4"],202                                                   br_cfg["PrefixLen"])203            elif self._cm_config["Overlays"][olid]["Type"] == "OVS":204                self._overlays[olid] = \205                        OvsBridge(br_cfg["BridgeName"],206                                  br_cfg["IP4"],207                                  br_cfg["PrefixLen"],208                                  sdn_ctrl_cfg=br_cfg.get("SDNController",209                                                          dict()))210            self.register_cbt("LinkManager", "LNK_ADD_IGN_INF", br_cfg["BridgeName"])211212        self._cfx_handle.start_subscription("LinkManager", "LNK_DATA_UPDATES")213        self.register_cbt("Logger", "LOG_INFO", "Module Loaded")214215    def req_handler_add_port(self, cbt):216        pass217218    def req_handler_del_port(self, cbt):219        pass220221    def req_handler_manage_bridge(self, cbt):222        try:223            olid = cbt.request.params["OverlayId"]224            port_name = cbt.request.params["TapName"]225            br = self._overlays[olid]226            if cbt.request.params["UpdateType"] == "ADDED":227                br.add_port(port_name)228                self.register_cbt(229                    "Logger", "LOG_INFO", "Port {0} added to bridge {1}"230                    .format(port_name, str(br)))231            elif cbt.request.params["UpdateType"] == "REMOVED":232                br.del_port(port_name)233                self.register_cbt(234                    "Logger", "LOG_INFO", "Port {0} removed from bridge {1}"235                    .format(port_name, str(br)))236        except RuntimeError as err:237            self.register_cbt("Logger", "LOG_WARNING", str(err))238        cbt.set_response(None, True)239        self.complete_cbt(cbt)240241    def resp_handler_(self, cbt):242        pass243244    def process_cbt(self, cbt):245        if cbt.op_type == "Request":246            if cbt.request.action == "BRG_ADD_PORT":247                self.req_handler_add_port(cbt)248            if cbt.request.action == "BRG_DEL_PORT":249                self.req_handler_del_port(cbt)250            if cbt.request.action == "LNK_DATA_UPDATES":251                self.req_handler_manage_bridge(cbt)252            else:253                self.req_handler_default(cbt)254        elif cbt.op_type == "Response":255            if cbt.request.action == "TOP_QUERY_PEER_IDS":256                self.resp_handler_query_peers(cbt)257            else:258                parent_cbt = self.get_parent_cbt(cbt)259                cbt_data = cbt.response.data260                cbt_status = cbt.response.status261                self.free_cbt(cbt)262                if (parent_cbt is not None and parent_cbt.child_count == 1):263                    parent_cbt.set_response(cbt_data, cbt_status)264                    self.complete_cbt(parent_cbt)265266    def timer_method(self):267        pass268269    def terminate(self):270        for olid in self._overlays:
...

test_impl_idl.py

Source:test_impl_idl.py

...25        super(TestOvsdbIdl, self).setUp()26        self.api = impl_idl.OvsdbIdl(ovsdb_connection)27        self.brname = utils.get_rand_device_name()28        # Destroying the bridge cleans up most things created by tests29        cleanup_cmd = self.api.del_br(self.brname)30        self.addCleanup(cleanup_cmd.execute)31    def test_br_exists_false(self):32        exists = self.api.br_exists(self.brname).execute(check_error=True)33        self.assertFalse(exists)34    def test_add_br_may_exist(self):35        self.api.add_br(self.brname).execute(check_error=True)36        with self.api.transaction(check_error=True) as txn:37            txn.add(self.api.add_br(self.brname, datapath_type="netdev"))38            exists = txn.add(self.api.br_exists(self.brname))39            dpt = txn.add(self.api.db_get("Bridge", self.brname,40                                          "datapath_type"))41        self.assertTrue(exists)42        self.assertEqual("netdev", dpt.result)43    def test_add_br_may_not_exist(self):44        self.api.add_br(self.brname).execute(check_error=True)45        cmd = self.api.add_br(self.brname, may_exist=False)46        self.assertRaises(RuntimeError, cmd.execute, check_error=True)47    def test_del_br_if_exists_false(self):48        cmd = self.api.del_br(self.brname, if_exists=False)49        self.assertRaises(RuntimeError, cmd.execute, check_error=True)50    def test_del_br_if_exists_true(self):51        self.api.del_br(self.brname).execute(check_error=True)52    def test_del_br(self):53        self.api.add_br(self.brname).execute(check_error=True)54        self.api.del_br(self.brname).execute(check_error=True)55        exists = self.api.br_exists(self.brname).execute(check_error=True)56        self.assertFalse(exists)57    def _test_add_port(self):58        pname = utils.get_rand_device_name()59        with self.api.transaction(check_error=True) as txn:60            txn.add(self.api.add_br(self.brname))61            txn.add(self.api.add_port(self.brname, pname))62        return pname63    def test_add_port(self):64        pname = self._test_add_port()65        plist_cmd = self.api.list_ports(self.brname)66        ports = plist_cmd.execute(check_error=True)67        self.assertIn(pname, ports)68    def test_add_port_may_exist_false(self):...

train_rnn.py

Source:train_rnn.py

1import numpy as np2import pickle3import math4from collections import defaultdict5def softmax(x):6    # https://stackoverflow.com/questions/34968722/softmax-function-python7    """Compute softmax values for each sets of scores in x."""8    return np.exp(x) / np.sum(np.exp(x), axis=0)9def create_one_hot(id_target, size):10    vec = np.zeros(size)11    vec[id_target] = 112    return vec13def create_ids(filename):14    x_ids = defaultdict(lambda: len(x_ids))15    y_ids = defaultdict(lambda: len(y_ids))16    for line in open(filename):17        wordtags = line.strip("\n").split(" ")18        for wordtag in wordtags:19            word, tag = wordtag.split("_")20            x_ids[word]21            y_ids[tag]22    return x_ids, y_ids23def create_feature_labels(filename, x_ids, y_ids):24    feature_labels = []25    for line in open(filename):26        wordtags = line.strip("\n").split(" ")27        x_list = []28        y_list = []29        for wordtag in wordtags:30            word, tag = wordtag.split("_")31            X = create_one_hot(x_ids[word],len(x_ids))32            x_list.append(X)33            Y = create_one_hot(y_ids[tag],len(y_ids))34            y_list.append(Y)35        feature_labels.append((x_list,y_list))36    return feature_labels37def find_max(p):38    y = 039    for i in range(len(p)):40        if p[i] > p[y]:41            y = i42    return y43def forward_rnn(wRx,wRh,br,wOh,bo,x):44    h = [0] * len(x) # hidden layers at time45    p = [0] * len(x) # output prob at time t46    y = [0] * len(x) # output value at time t47    for t in range(len(x)):48        if t > 0:49            h[t] = np.tanh(wRx.dot(x[t]) + wRh.dot(h[t-1]) + br)50        else:51            h[t] = np.tanh(wRx.dot(x[t]) + br)52        p[t] = softmax(wOh.dot(h[t]) + bo)53        #y[t] = np.argmax(p[t])54        y[t] = find_max(p[t])55    return h, p, y56def gradient_rnn(wRx,wRh,br,wOh,bo,x,h,p,y_):57    del_wRx = [0]58    del_wRh = [0]59    del_br = [0]60    del_wOh = [0]61    del_bo = [0] # initialize62    phiR_ = np.zeros(len(br)) #  Error from the following time step63    for t in range(len(x)):64        phi0_ = p[t] - y_[t]# Output error65        del_wOh += np.outer(phi0_,h[t])66        phiR = np.dot(phiR_,wRh) + np.dot(phi0_,wOh) # Backprop67        phiR_ = phiR * (1- pow(h[t],2)) # tanh gradient68        del_wRx += np.outer(phiR_,x[t])69        del_bo += phi0_ # Îb0 += Î´0' Output gradient70        del_br += phiR_ # Îbr += Î´r' Hidden gradient71        if t != 0:72            del_wRh += np.outer(phiR_,h[t-1])73    return del_wRx, del_wRh, del_br, del_wOh, del_bo74def update_weights(wRx,wRh,br,wOh,bo,del_wRx,del_wRh,del_br,del_wOh,del_bo,lamb):75    wRx += lamb * del_wRx76    wRh += lamb * del_wRh77    br += lamb * del_br78    wOh += lamb * del_wOh79    bo += lamb * del_bo80    return wRx,wRh,br,wOh,bo81if __name__ == "__main__":82    #file = "../test/05-train-input.txt"83    file = "../data/wiki-en-train.norm_pos"84    x_ids, y_ids = create_ids(file)85    feature_labels = create_feature_labels(file,x_ids,y_ids)86    wRx = (np.random.rand(2, len(x_ids)) - 0.5)/5 # (2,vocab)87    wOh =(np.random.rand(len(y_ids), 2) - 0.5) /5 # (pos_type,2)88    wRh = (np.random.rand(2,2) - 0.5) /5 #(2,2)89    bo = np.zeros(len(y_ids))90    br = np.zeros(2)91    epoch = 392    lamb = 0.01593    for _ in range(epoch):94        for x, y_correct in feature_labels:95            h,p,y_predict = forward_rnn(wRx,wRh,br,wOh,bo,x)96            #print(y_predict)97            del_wRx, del_wRh, del_br, del_wOh, del_bo = gradient_rnn(wRx,wRh,br,wOh,bo,x,h,p,y_correct)98            wRx,wRh,br,wOh,bo, = update_weights(wRx,wRh,br,wOh,bo,del_wRx,del_wRh,del_br,del_wOh,del_bo,lamb)99    net = (wRx,wRh,br,wOh,bo)100    pickle.dump(net, open("net.pickle", "wb" ) )101    with open("x_ids.txt","w") as output:102        for k,v in x_ids.items():103            print("{}\t{}".format(k,v), file=output)104    with open("y_ids.txt","w") as output:105        for k,v in y_ids.items():...

Automation Testing Tutorials

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.