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

How to use wait_worker method in lisa

Best Python code snippet using lisa_python

openstack-integration.py

Source:openstack-integration.py

...74        self.worker = subprocess.Popen(self.worker_cmd,75                                       stdout=subprocess.PIPE,76                                       stderr=subprocess.PIPE,77                                       shell=True)78    def wait_worker(self):79        if not self.worker:80            return81        (stdoutdata, stderrdata) = self.worker.communicate()82        stdoutdata = stdoutdata.decode('utf-8')83        stderrdata = stderrdata.decode('utf-8')84        logging.info(self.worker_cmd + ":" +85                     " stdout " + stdoutdata +86                     " stderr " + stderrdata + " end ")87        assert self.worker.returncode == 088        self.worker = None89    def get_teuthology_log(self):90        # the archive is removed before each test, there must91        # be only one run and one job92        run = os.listdir(self.archive)[0]93        job = os.listdir(os.path.join(self.archive, run))[0]94        path = os.path.join(self.archive, run, job, 'teuthology.log')95        return open(path, 'r').read()96class TestSuite(Integration):97    def setup(self):98        self.d = tempfile.mkdtemp()99        self.setup_worker()100        logging.info("TestSuite: done worker")101    def teardown(self):102        self.wait_worker()103        shutil.rmtree(self.d)104    def test_suite_noop(self):105        cwd = os.getcwd()106        os.mkdir(self.d + '/upload', 0o755)107        upload = 'localhost:' + self.d + '/upload'108        args = ['--suite', 'noop',109                '--suite-dir', cwd + '/teuthology/openstack/test',110                '--machine-type', 'openstack',111                '--archive-upload', upload,112                '--verbose']113        logging.info("TestSuite:test_suite_noop")114        scripts.suite.main(args)115        self.wait_worker()116        log = self.get_teuthology_log()117        assert "teuthology.run:pass" in log118        assert "Well done" in log119        upload_key = teuth_config.archive_upload_key120        if upload_key:121            ssh = "RSYNC_RSH='ssh -i " + upload_key + "'"122        else:123            ssh = ''124        assert 'teuthology.log' in teuthology.misc.sh(ssh + " rsync -av " + upload)125    def test_suite_nuke(self):126        cwd = os.getcwd()127        args = ['--suite', 'nuke',128                '--suite-dir', cwd + '/teuthology/openstack/test',129                '--machine-type', 'openstack',130                '--verbose']131        logging.info("TestSuite:test_suite_nuke")132        scripts.suite.main(args)133        self.wait_worker()134        log = self.get_teuthology_log()135        assert "teuthology.run:FAIL" in log136        locks = teuthology.lock.query.list_locks(locked=True)137        assert len(locks) == 0138class TestSchedule(Integration):139    def setup(self):140        self.d = tempfile.mkdtemp()141        self.setup_worker()142    def teardown(self):143        self.wait_worker()144        shutil.rmtree(self.d)145    def test_schedule_stop_worker(self):146        job = 'teuthology/openstack/test/stop_worker.yaml'147        args = ['--name', 'fake',148                '--verbose',149                '--owner', 'test@test.com',150                '--worker', 'openstack',151                job]152        scripts.schedule.main(args)153        self.wait_worker()154    def test_schedule_noop(self):155        job = 'teuthology/openstack/test/noop.yaml'156        args = ['--name', 'fake',157                '--verbose',158                '--owner', 'test@test.com',159                '--worker', 'openstack',160                job]161        scripts.schedule.main(args)162        self.wait_worker()163        log = self.get_teuthology_log()164        assert "teuthology.run:pass" in log165        assert "Well done" in log166    def test_schedule_resources_hint(self):167        """It is tricky to test resources hint in a provider agnostic way. The168        best way seems to ask for at least 1GB of RAM and 10GB169        disk. Some providers do not offer a 1GB RAM flavor (OVH for170        instance) and the 2GB RAM will be chosen instead. It however171        seems unlikely that a 4GB RAM will be chosen because it would172        mean such a provider has nothing under that limit and it's a173        little too high.174        Since the default when installing is to ask for 7000 MB, we175        can reasonably assume that the hint has been taken into176        account if the instance has less than 4GB RAM.177        """178        try:179            teuthology.openstack.OpenStack().run("volume list")180            job = 'teuthology/openstack/test/resources_hint.yaml'181            has_cinder = True182        except subprocess.CalledProcessError:183            job = 'teuthology/openstack/test/resources_hint_no_cinder.yaml'184            has_cinder = False185        args = ['--name', 'fake',186                '--verbose',187                '--owner', 'test@test.com',188                '--worker', 'openstack',189                job]190        scripts.schedule.main(args)191        self.wait_worker()192        log = self.get_teuthology_log()193        assert "teuthology.run:pass" in log194        assert "RAM size ok" in log195        if has_cinder:196            assert "Disk size ok" in log197class TestLock(Integration):198    def setup(self):199        self.options = ['--verbose',200                        '--machine-type', 'openstack' ]201    def test_main(self):202        args = scripts.lock.parse_args(self.options + ['--lock'])203        assert teuthology.lock.cli.main(args) == 0204    def test_lock_unlock(self):205        for image in teuthology.openstack.OpenStack.image2url.keys():...

day7.py

Source:day7.py

...61    task = tasks[0]62    wait_sec = compute_wait_sec(task)63    state["tasks_free"].remove(task)64    state["tasks_inproc"].add((task, wait_sec))65def wait_worker(state):66    min_wait_time = min([wait_sec for _, wait_sec in state["tasks_inproc"]])67    state["duration"] += min_wait_time68    state["tasks_inproc"] = {(task, wait_sec - min_wait_time) for task, wait_sec in state["tasks_inproc"]}69    done = {(task, wait_sec) for task, wait_sec in state["tasks_inproc"] if wait_sec == 0}70    state["tasks_inproc"] = state["tasks_inproc"] - done71    state["tasks_done"] += sorted([task for task, _ in done])72    for (task, _) in done:73        pop_task(task, state["tasks_graph"])74def compute_wait_sec(letter):75    costs = {}76    pairs_letter_cost = zip(ascii_uppercase, range(1, len(ascii_uppercase) + 1))77    for ascii_letter, letter_cost in pairs_letter_cost:78        costs[ascii_letter] = BASELINE_SEC + letter_cost79    return costs[letter]80def partB(graph):81    state = {82        "tasks_graph": graph, # holds tasks not done yet83        "tasks_free": set(),84        "tasks_inproc": set(),85        "tasks_done": [],  # list to keep track of order86        "duration": 087    }88    step = "collect_free_tasks"89    # goto style control flow implemented with while-if-continues90    while step != "end":91        if step == "collect_free_tasks":92            collect_free_tasks(state)93            step = "all_done?"94            continue95        if step == "all_done?":96            if all_done(state):97                step = "end"98            else:99                step = "any_free_task?"100            continue101        if step == "any_free_task?":102            if any_free_task(state):103                step = "any_worker_avail?"104            else:105                step = "wait_worker"106            continue107        if step == "any_worker_avail?":108            if any_worker_avail(state):109                step = "assign_task_worker"110            else:111                step = "wait_worker"112            continue113        if step == "wait_worker":114            wait_worker(state)115            step = "collect_free_tasks"116            continue117        if step == "assign_task_worker":118            assign_task_worker(state)119            step = "any_free_task?"120            continue121    return state["duration"]122if __name__ == '__main__':123    filepath = Path(argv[1])124    # Part A125    tasks = parse(filepath)126    order = partA(tasks)127    print("part A:", "".join(order))128    # Part B...

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.