Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _get_command_line method in avocado

Best Python code snippet using avocado_python

portable-gcc-flags.py

Source:portable-gcc-flags.py

...28                            yield '-m{}{}'.format(switch_prefix, key)29                else:30                    yield '-m{}={}'.format(key, value)31    return tuple(_helper())32def _get_command_line(args):33    return ' '.join(map(shlex.quote, args))34async def _get_raw_gcc_target_flags(arch, flags):35    command_line_options = _get_gcc_command_line_options(arch, flags)36    return await _get_output(args=[*_GCC, *command_line_options, '-Q', '--help=target'])37def _parse_gcc_target_flags(content):38    flag_line_regex = re.compile(r'\s*-m([^\s]+)\s*([^\s]*)\s*')39    for line in content.splitlines():40        match = flag_line_regex.fullmatch(line)41        try:42            key, value = match.groups()43        except AttributeError:44            continue45        if key.endswith('='):46            yield key[:-1], value47        else:48            yield key, (value == '[enabled]')49async def _get_gcc_target_flags(arch, flags=None):50    return tuple(_parse_gcc_target_flags(await _get_raw_gcc_target_flags(arch, flags)))51def _make_flag_key_extractor(arch):52    def _key(flags):53        return len(flags), len(_get_command_line(_get_gcc_command_line_options(arch, flags)))54    return _key55async def _simplify_gcc_target_flags(arch, flags):56    total_flags = len(flags)57    reference_result = await _get_gcc_target_flags(arch, flags)58    tested = 059    total_tests = 2 ** len(flags)60    def _loop(index):61        if index < total_flags:62            for flags1 in _loop(index + 1):63                yield flags164                yield (flags[index], *flags1)65        else:66            yield ()67    async def _test_flags(flags1):68        nonlocal tested69        result = await _get_gcc_target_flags(arch, flags1)70        tested += 171        if tested % 32 == 0:72            print('Progress: {:.4} %'.format(tested * 100 / total_tests))73        return flags1, result == reference_result74    flags_iter = iter(_loop(0))75    result_futures = tuple(map(_test_flags, itertools.islice(flags_iter, _CONCURRENCY)))76    results = []77    while result_futures:78        done, pending = await asyncio.wait(result_futures, return_when=asyncio.FIRST_COMPLETED)79        pending.update(map(_test_flags, itertools.islice(flags_iter, _CONCURRENCY - len(pending))))80        result_futures = pending81        done_results = tuple(task.result() for task in done)82        results.extend(result for result, accept in done_results if accept)83    return min(results, key=_make_flag_key_extractor(arch))84async def _simplify_gcc_target_flags_fast(arch, flags):85    reference_result = await _get_gcc_target_flags(arch, flags)86    visited = set()87    concurrency = 088    async def _test_flags(flags1):89        if flags1 not in visited:90            visited.add(flags1)91            nonlocal concurrency92            while concurrency >= _CONCURRENCY:93                await asyncio.sleep(0.1)94            concurrency += 195            result = await _get_gcc_target_flags(arch, flags1)96            concurrency -= 197            if result == reference_result:98                return await _simplify(flags1)99            else:100                return ()101        else:102            return ()103    def _set_remove(s, value):104        result = set(s)105        result.remove(value)106        return frozenset(result)107    async def _simplify(flags1):108        print('Processing: {}'.format(_get_command_line(_get_gcc_command_line_options(arch, flags1))))109        result_futures = []110        for flag in flags1:111            result_futures.append(_test_flags(_set_remove(flags1, flag)))112        results1 = tuple(itertools.chain.from_iterable(await asyncio.gather(*result_futures)))113        if results1:114            return results1115        else:116            return [flags1]117    results = await _simplify(frozenset(flags))118    return min(results, key=_make_flag_key_extractor(arch))119async def _main_async(arch, slow):120    native_flags = await _get_gcc_target_flags('native')121    arch = arch or next(value for key, value in native_flags if key == 'arch')122    arch_flags = set(await _get_gcc_target_flags(arch))123    significant_flags = tuple(flag for flag in native_flags if flag[0] != 'arch' and (flag not in arch_flags))124    if slow:125        simplified_flags = await _simplify_gcc_target_flags(arch, significant_flags)126    else:127        simplified_flags = await _simplify_gcc_target_flags_fast(arch, significant_flags)128    print('  Original flags: {}'.format(_get_command_line(_get_gcc_command_line_options(arch, significant_flags))))129    print('Simplified flags: {}'.format(_get_command_line(_get_gcc_command_line_options(arch, simplified_flags))))130def main():131    global _GCC132    global _CONCURRENCY133    arg_parser = argparse.ArgumentParser()134    arg_parser.add_argument('--arch', type=str)135    arg_parser.add_argument('--gcc', type=str, default='gcc')136    arg_parser.add_argument('--concurrency', type=int, default=4)137    arg_parser.add_argument('--slow', action='store_true')138    args = arg_parser.parse_args()139    _GCC = re.findall(r'[^\s]+', args.gcc)140    _CONCURRENCY = args.concurrency141    if sys.platform == 'win32':142        loop = asyncio.ProactorEventLoop()143    else:...

remote_script.py

Source:remote_script.py

...33        :return:34        """35        raise NotImplementedError("Method _get_interpreter is not implemented!")36    @abstractmethod37    def _get_command_line(self):38        """39        The command line which should be called to execute the script.40        :return:41        """42        raise NotImplementedError("Method _get_command_line is not implemented!")43    @abstractmethod44    def _get_output_cleaner(self):45        """46        An optional function to clean up the output.47        :return:48        """49        return None50    def execute(self):51        with open(self.script, 'r') as f:52            contents = f.read()53            shell_exec(self._get_command_line().format(interpreter=self._get_interpreter(),54                                                       args=self.script_args,55                                                       script=contents),56                       print_output=True,57                       output_cleaner=self._get_output_cleaner())58# -----------------------------------------------------------------------------59class RunPyScript(RemoteScript):60    @staticmethod61    def regexp():62        return r"^\s*\!py($| )"63    @staticmethod64    def usage():65        write_str("Usage: !py [script on the local machine] [script arguments]\r\n", LogLevel.WARNING)66    @staticmethod67    def name():68        return "!py"69    @staticmethod70    def description():71        return "Runs a python script from the local machine in memory."72    def _get_interpreter(self):73        return "python"74    def _get_command_line(self):75        return "{interpreter} - {args} <<'__EOF__'\r\n{script}\r\n__EOF__"76    def _get_output_cleaner(self):77        # The Python interpreter displays a prompt while reading scripts from stdin. Strip it.78        return lambda s: s.lstrip(" >")79# -----------------------------------------------------------------------------80class RunShScript(RemoteScript):81    @staticmethod82    def regexp():83        return r"^\s*\!sh($| )"84    @staticmethod85    def usage():86        write_str("Usage: !sh [script on the local machine] [script arguments]\r\n", LogLevel.WARNING)87    @staticmethod88    def name():89        return "!sh"90    @staticmethod91    def description():92        return "Runs a shell script from the local machine in memory."93    def _get_interpreter(self):94        return "sh"95    def _get_command_line(self):96        return "{interpreter} -s {args} <<'__EOF__'\r\n{script}\r\n__EOF__"97# -----------------------------------------------------------------------------98register_plugin(RunPyScript)...

naoweb.py

Source:naoweb.py

...27@bottle.post('/command')28def run_command():29    """Runs naocommandline command, returns console output and sets connected cookie"""30    output = _invoke_command(bottle.request.json['command']) #pylint: disable=unsubscriptable-object31    bottle.response.set_cookie("naoconnected", '1' if _get_command_line().is_connected else '0')32    return output33@bottle.route('/camera')34def get_camera_image_top():35    return get_camera_image()36@bottle.route('/camera/<camera_name>')37def get_camera_image(camera_name='top'):38    """Takes a picture with top or bottom camera and returns the image contents in png format39        encoded in base64. 40    """41    use_bottom_camera = (camera_name == 'bottom')42    image = None43    try:44        video_controller = _get_command_line().video_controller45        image = video_controller.get_picture(use_bottom_camera)46    except (KeyError, AttributeError):47        bottle.abort(400, "Must /connect first.")48    except (ValueError, TypeError):49        bottle.abort(400, "Invalid value sent to server.")50    except Exception: # pylint: disable=broad-except51        bottle.abort(500, "Error on server.")52    if image is not None:53        output = StringIO()54        image.save(output, "JPEG", quality=90, progressive=True)55        contents = output.getvalue().encode("base64")56        output.close()57        #contents = contents.split('\n')[0] #uncommenting this may help58        return contents59    else:60        return None 61    62def _set_command_line(command_line):63    session = bottle.request.environ.get('beaker.session')64    session['commandline'] = command_line65def _get_command_line():66    return bottle.request.environ.get('beaker.session')['commandline']67def _invoke_command(command):68    """Redirects 'console' output, invokes commandline-style command and returns console output69        Note: NaoCommandLine catches most every exception and ouputs error messages to console70        which is what naoweb does as well therefore no all invoked commands will return 200 OK.71    """72    command_line = _get_command_line()73    old_stdout = _get_command_line().stdout74    command_line.stdout = new_stdout = StringIO()75    command = command_line.precmd(command) 76    command_line.onecmd(command)77    command_line.stdout = old_stdout78    return new_stdout.getvalue()...

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.