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

How to use run_safe method in localstack

Best Python code snippet using localstack_python

adb.py

Source:adb.py

...19ADB_BRIDGE_DIR = realpath(INPUTS_DIR + '/adb_bridge')20ADB_BIN_DIR = realpath(INPUTS_DIR + '/external/adb')21ANDROID_TMP_DIR = '/data/local/tmp'22# Print adb output to stdout when "-v" is passed to QCSuper23def run_safe(args, **kwargs):24    debug('[>] Running adb command: ' + list2cmdline(args))25    result = run(args, **kwargs)26    result_string = ((result.stdout or b'') + (result.stderr or b''))27    if result and result_string:28        debug('[<] Obtained result for running "%s": %s' % (list2cmdline(args), result_string))29    return result30"""31    This class implements reading Qualcomm DIAG data from a the /dev/diag32    character device, on a remote Android device connected through ADB.33    34    For this, it uploads the C program located in "./adb_bridge/" which35    creates a TCP socket acting as a proxy to /dev/diag.36"""37QCSUPER_TCP_PORT = 4355538if platform in ('cygwin', 'win32'):39    adb_exe = ADB_BIN_DIR + '/adb_windows.exe'40elif platform == 'darwin':41    adb_exe = which('adb') or ADB_BIN_DIR + '/adb_macos'42else:43    adb_exe = which('adb') or ADB_BIN_DIR + '/adb_linux'44class AdbConnector(HdlcMixin, BaseInput):45    46    def __init__(self):47        48        self.su_command = '%s'49        50        # Whether we can use "adb exec-out" instead51        # of "adb shell" (we should do it if the52        # remote phone supports it)53        self.can_use_exec_out = None # (boolean, None = unknown)54        55        self.ADB_TIMEOUT = 1056        57        # Send batch commands to check for the writability of /dev/diag through58        # adb, and for the availability of the "su" command59        60        bash_output = self.adb_shell(61            'test -w /dev/diag; echo DIAG_NOT_WRITEABLE=$?; ' +62            'test -e /dev/diag; echo DIAG_NOT_EXISTS=$?; ' +63            'test -r /dev; echo DEV_NOT_READABLE=$?; ' +64            'su -c id'65        )66        67        # Check for the presence of /dev/diag68        69        if not search('DIAG_NOT_WRITEABLE=[01]', bash_output):70            71            print('Could not run a bash command your phone, is adb functional?')72            73            exit(bash_output)74        75        # If writable, continue76        77        elif 'DIAG_NOT_WRITEABLE=0' in bash_output:78            79            pass80        # If not present, raise an error81        82        elif 'DEV_NOT_READABLE=0' in bash_output and 'DIAG_NOT_EXISTS=1' in bash_output:83            84            exit('Could not find /dev/diag, does your phone have a Qualcomm chip?')85        # If maybe present but not writable, check for root86        87        elif 'uid=0' in bash_output:88            89            self.su_command = 'su -c "%s"'90        91        elif 'uid=0' in self.adb_shell('su 0,0 sh -c "id"'):92            93            self.su_command = 'su 0,0 sh -c "%s"'94    95        else:96            97            # "adb shell su" didn't work, try "adb root"98            99            adb = run_safe([adb_exe, 'root'], stdout = PIPE, stderr = STDOUT, stdin = DEVNULL)100            101            if b'cannot run as root' in adb.stdout:102                103                exit('Could not get root to adb, is your phone rooted?')104        105            run_safe([adb_exe, 'wait-for-device'], stdin = DEVNULL, check = True)106        107        # Once root has been obtained, send batch commands to check108        # for the presence of /dev/diag through adb109        110        bash_output = self.adb_shell(111            'test -e /dev/diag; echo DIAG_NOT_EXISTS=$?'112        )113        114        # If not present, raise an error115        116        if 'DIAG_NOT_EXISTS=1' in bash_output:117            118            exit('Could not find /dev/diag, does your phone have a Qualcomm chip?')119        120        # Upload the adb_bridge121        122        adb = run_safe([adb_exe, 'push', ADB_BRIDGE_DIR + '/adb_bridge', ANDROID_TMP_DIR],123            124            stdin = DEVNULL, stdout = PIPE, stderr = STDOUT125        )126    127        if b'error' in adb.stdout or adb.returncode != 0:128            129            exit(adb.stdout.decode('utf8'))130        131        # Launch the adb_bridge132        133        self._relaunch_adb_bridge()134        self.packet_buffer = b''135        136        super().__init__()137    138    def _relaunch_adb_bridge(self):139        140        if hasattr(self, 'adb_proc'):141            self.adb_proc.terminate()142        143        self.adb_shell(144            'killall -q adb_bridge; ' +145            'chmod 755 ' + ANDROID_TMP_DIR + '/adb_bridge'146        )147        148        run_safe([adb_exe, 'forward', 'tcp:' + str(QCSUPER_TCP_PORT), 'tcp:' + str(QCSUPER_TCP_PORT)], check = True, stdin = DEVNULL)149        150        self.adb_proc = Popen([adb_exe, 'exec-out' if self.can_use_exec_out else 'shell', self.su_command % (ANDROID_TMP_DIR + '/adb_bridge')],151            152            stdin = DEVNULL, stdout = PIPE, stderr = STDOUT,153            preexec_fn = setpgrp,154            bufsize = 0, universal_newlines = True155        )156    157        for line in self.adb_proc.stdout:158            159            if 'Connection to Diag established' in line:160                161                break162            163            else:164                165                stderr.write(line)166                stderr.flush()167        self.socket = socket(AF_INET, SOCK_STREAM)168        169        try:170            171            self.socket.connect(('localhost', QCSUPER_TCP_PORT))172        173        except Exception:174            175            self.adb_proc.terminate()176            177            exit('Could not communicate with the adb_bridge through TCP')178        179        self.received_first_packet = False180    181    """182        This utility function tries to run a command to adb,183        raising an exception when it is unreachable.184        185        :param command: A shell command (string)186        187        :returns The combined stderr and stdout from "adb shell" (string)188    """189    190    def adb_shell(self, command):191        192        try:193            194            # Can we use "adb exec-out"?195            196            if self.can_use_exec_out is None:197                        198                adb = run_safe([adb_exe, 'exec-out', 'id'],199                    200                    stdin = DEVNULL, stdout = PIPE, stderr = STDOUT, timeout = self.ADB_TIMEOUT201                )202            203                self.can_use_exec_out = (adb.returncode == 0)204            205            # Can we execute commands?206            207            adb = run_safe([adb_exe, 'exec-out' if self.can_use_exec_out else 'shell', self.su_command % command],208                209                stdin = DEVNULL, stdout = PIPE, stderr = STDOUT, timeout = self.ADB_TIMEOUT210            )211        212        except TimeoutExpired:213            214            exit('Communication with adb timed out, is your phone displaying ' +215                'a confirmation dialog?')216        if b'error' in adb.stdout or adb.returncode != 0:217    218            if b'device not found' in adb.stdout or b'no devices' in adb.stdout:219            220                exit('Could not connect to the adb, is your phone plugged with USB '221                    + 'debugging enabled?')222            223            elif b'confirmation dialog on your device' in adb.stdout:224            225                exit('Could not connect to the adb, is your phone displaying ' +226                    'a confirmation dialog?')227            228            else:229                230                print('Could not connect to your device through adb')231            232            exit(adb.stdout.decode('utf8'))233        234        return adb.stdout.decode('utf8').strip()235    236    def __del__(self):237        238        try:239            240            if hasattr(self, 'adb_proc'):241                self.adb_proc.terminate()242        243        except Exception:244            245            pass246    247    def send_request(self, packet_type, packet_payload):248        249        raw_payload = self.hdlc_encapsulate(bytes([packet_type]) + packet_payload)250        251        self.socket.send(raw_payload)252    253    def get_gps_location(self):254        255        lat = None256        lng = None257        258        gps_info = run_safe([adb_exe, 'exec-out' if self.can_use_exec_out else 'shell', 'dumpsys', 'location'], stdout = PIPE)259        gps_info = gps_info.stdout.decode('utf8')260        261        gps_info = search('(\d+\.\d+),(\d+\.\d+)', gps_info)262        if gps_info:263            lat, lng = map(float, gps_info.groups())264        265        return lat, lng266    267    def read_loop(self):268        269        while True:270            271            while self.TRAILER_CHAR not in self.packet_buffer:272                ...

1step_match.py

Source:1step_match.py

...34    if log_file is not None:35        log_file.write(log_line)36        log_file.write("\n")37        log_file.flush()38def run_safe(command, output_fn=None, output_fo=None, err_msg=None, thr_exc=True, silent=False):39    """Run a shell command safely.40    Args:41        command (list of str): Command to execute.42        output_fn (str): Name of a file for storing the output.43        output_fo (fileobject): Output file object. If both params are None, the standard output is used.44        err_msg (str): Error message if the command fails.45        thr_exc (bool): Through exception if the command fails. error_msg or thr_exc must be set.46        silent (bool): Silent mode (print messages only if the command fails).47    Raises:48        RuntimeError: Command exited with a non-zero code.49    """50    assert output_fn is None or output_fo is None51    assert err_msg is not None or thr_exc52    assert len(command) > 053    command_safe = []54    for part in command:55        part = str(part)56        if " " in part:57            part = '"{}"'.format(part)58        command_safe.append(part)59    command_str = " ".join(command_safe)60    if not silent:61        message("Shell command:", command_str)62    if output_fn is None:63        if output_fo is None:64            out_fo = sys.stdout65        else:66            out_fo = output_fo67    else:68        out_fo = open(output_fn, "w+")69    if out_fo == sys.stdout:70        p = subprocess.Popen("/bin/bash -e -o pipefail -c '{}'".format(command_str), shell=True)71    else:72        p = subprocess.Popen("/bin/bash -e -o pipefail -c '{}'".format(command_str), shell=True, stdout=out_fo)73    stdout, stderr = p.communicate()74    error_code = p.returncode75    if output_fn is not None:76        out_fo.close()77    if error_code == 0 or error_code == 141:78        if not silent:79            message("Finished")80    else:81        message("Unfinished, an error occurred (error code {}): {}".format(error_code, command_str))82        if err_msg is not None:83            print('Error: {}'.format(err_msg), file=sys.stderr)84        if thr_exc:85            raise RuntimeError("A command failed, see messages above.")86        sys.exit(1)87def create_bwa_index(fa):88    # cmd('"{bwa}" index "{fa}"'.format(bwa=bwa,fa=fa))89    run_safe([bwa, 'fa2pac', fa, fa])90    run_safe([bwa, 'pac2bwtgen', fa + ".pac", fa + ".bwt", ">", "/dev/null"])91    run_safe([bwa, 'bwtupdate', fa + ".bwt"])92    run_safe([bwa, 'bwt2sa', fa + ".bwt", fa + ".sa"])93def create_klcp(fa, k):94    run_safe([prophyle_index, 'klcp', '-k', k, fa, ">", "/dev/null"])95def query(fa, fq, k, u=False, v=False, t=1):96    params = ""97    cmd = [prophyle_index, 'query', "-v" if v else "", "-u" if u else "", '-k', k, '-t', t, fa, fq]98    run_safe(cmd)99def main():100    parser = argparse.ArgumentParser(description='Single-command ProPhex matching.')101    parser.add_argument(102        '-k',103        type=int,104        metavar='int',105        dest='k',106        required=True,107        help='k-mer length',108    )109    parser.add_argument(110        '-t',111        type=int,112        default=1,...

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.