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

How to use go_online method in fMBT

Best Python code snippet using fMBT_python

dispersy_preprocessor.py

Source:dispersy_preprocessor.py

1import os2import sys3from random import expovariate, random, randint4from gumby.scenario import ScenarioRunner5class ScenarioPreProcessor(ScenarioRunner):6    def __init__(self, filename, outputfile=sys.stdout, max_tstmp=0):7        ScenarioRunner.__init__(self, filename)8        self._callables = {}9        self._callables['churn'] = self.churn10        self._callables['churn_pattern'] = self.churn_pattern11        print >> sys.stderr, "Looking for max_timestamp, max_peer... in %s" % filename,12        self.max_peer = 013        for (tstmp, lineno, clb, args) in self._parse_scenario(filename):14            max_tstmp = max(tstmp, max_tstmp)15        print >> sys.stderr, "\tfound %d and %d" % (max_tstmp, self.max_peer)16        _max_peer = self.max_peer17        print >> sys.stderr, "Preprocessing file...",18        for (tstmp, lineno, clb, args) in self._parse_scenario(filename):19            print >> outputfile, self.file_buffer[1][lineno - 1][1]20            if clb in self._callables:21                for peer in self.yes_peers:22                    for line in self._callables[clb](tstmp, max_tstmp, *args):23                        print >> outputfile, line, '{%s}' % peer24        print >> sys.stderr, "\tdone"25    def _parse_for_this_peer(self, peerspec):26        if peerspec:27            self.yes_peers, self.no_peers = self._parse_peerspec(peerspec)28            self.max_peer = max(self.max_peer, max(self.yes_peers))29        else:30            self.yes_peers = set()31            self.no_peers = set()32        if not self.yes_peers:33            self.yes_peers = set(range(1, self.max_peer + 1))34        for peer in self.no_peers:35            self.yes_peers.discard(peer)36        return True37    def churn(self, tstmp, max_tstmp, churn_type, desired_mean=300, min_online=5.0):38        desired_mean = float(desired_mean)39        min_online = float(min_online)40        def get_delay(step):41            if churn_type == 'expon':42                return min_online + expovariate(1.0 / (desired_mean - min_online))43            elif churn_type == 'fixed':44                if step == 1:45                    return randint(min_online, desired_mean)46                return desired_mean47            else:48                raise NotImplementedError('only expon churn is implemented, got %s' % churn_type)49        go_online = random() < 0.550        step = 151        while tstmp < max_tstmp:52            yield "@0:%d %s" % (tstmp, "online" if go_online else "offline")53            tstmp += get_delay(step)54            go_online = not go_online55            step += 156    def churn_pattern(self, tstmp, max_tstmp, pattern, min_online=5.0):57        pattern = [online / 100.0 for online in map(float, pattern.split(','))]58        min_online = float(min_online)59        i = 060        prev_state = None61        while tstmp < max_tstmp:62            go_online = random() < pattern[i]63            i = (i + 1) % len(pattern)64            if go_online != prev_state:65                yield "@0:%d %s" % (tstmp, "online" if go_online else "offline")66                prev_state = go_online67            tstmp += min_online68def main(inputfile, outputfile, maxtime=0):69    inputfile = os.path.abspath(inputfile)70    if os.path.exists(inputfile):71        f = open(outputfile, 'w')72        ScenarioPreProcessor(inputfile, f, maxtime)73        f.close()74    else:75        print >> sys.stderr, inputfile, "not found"76if __name__ == '__main__':77    if len(sys.argv) < 3:78        print "Usage: %s <input-file> <output-file> (<max-time>)" % (sys.argv[0])79        print >> sys.stderr, sys.argv80        exit(1)81    if len(sys.argv) == 4:82        main(sys.argv[1], sys.argv[2], float(sys.argv[3]))83    else:...

03_internet_cafe.py

Source:03_internet_cafe.py

1from time import sleep2from concurrent.futures import ThreadPoolExecutor, as_completed3from random import randint4def go_online(i):5  print('Tourist {} is online'.format(i))6  duration = randint(3, 5)7  sleep(duration)8  print('Tourist {} spend {} hours online'.format(i, duration))9  return True10def queue(i):11  print('Tourist {} is waiting to go online'.format(i))12def check_queue(tourists, already_online):13  [queue(tourist) for tourist in tourists if tourist not in already_online]14def main():15  already_online = {}16  tourists = [i for i in range(25)]17  is_queue = False18  with ThreadPoolExecutor(max_workers = 8) as executor:...

01_computer.py

Source:01_computer.py

2	def open(self, link):        3		self.current_link = link        4		print(f"Link '{link}' was opened!")5class Computer:6	def go_online(self):        7		return BrowserWindow()    8	def open_itvdn_lesson(self):        9		browser_window = self.go_online()        10		browser_window.open("itvdn.com")        11		return browser_window12my_own_computer = Computer()...

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.