Method run_for

bessctl/test_utils.py:216–237

(self, module, igates, duration, pkt_update_fields=[])

        self.bess.reset_all()

    def run_for(self, module, igates, duration, pkt_update_fields=[]):
        self.bess.pause_all()

        fields = pkt_update_fields
        if len(fields) == 0:
            fields.append({'offset': 26, 'size': 4,
                           'min': 1, 'max': pow(2, 32) - 1})
            fields.append({'offset': 30, 'size': 4,
                           'min': 1, 'max': pow(2, 32) - 1})

        # source and associate sockets
        for igate in igates:
            src = self.bess.create_module('Source')
            random = self.bess.create_module('RandomUpdate', 'RandomUpdateArg',
                                             {'fields': fields})
            self.bess.connect_modules(src.name, random.name)
            self.bess.connect_modules(random.name, module.name, 0, igate)

        # running tests
        self.bess.resume_all()
        time.sleep(duration)
        self.bess.pause_all()

    def run_pipeline(self, src_module, dst_module, igate, input_pkts,
                     ogates, time_out=3, proto=scapy.Ether):