TestMu Conference is Back! Future of Testing + AI LambdaTest G2 Announcement

How to use rpc_uri method in uiautomator

Best Python code snippet using uiautomator

FlaskExtRpcClient.py

Source:FlaskExtRpcClient.py

1#! /usr/bin/env python2# -*- coding: utf-8 -*-3import threading4import zerorpc5import traceback6try:7    from flask import _app_ctx_stack as stack8except ImportError:9    from flask import _request_ctx_stack as stack10global_data = threading.local()11RPC_URI = ['tcp://127.0.0.1:5007']12class RpcClient(zerorpc.Client):13    def __init__(self, app=None, auto_close=False):14        self.app = app15        if app is not None and auto_close:16            self.init_app(app)17    def init_app(self, app):18        app.config.setdefault('RPC_CLI', ':memory:')19        # Use the newstyle teardown_appcontext if it's available,20        # otherwise fall back to the request context21        if hasattr(app, 'teardown_appcontext'):22            app.teardown_appcontext(self.teardown)23        else:24            app.teardown_request(self.teardown)25    def connect(self):26        if not hasattr(global_data, "cli"):27            global_data.cli = zerorpc.Client(connect_to=RPC_URI)28            print ' global_data.cli  zerorpc.Client connect_to ', RPC_URI29        else:30            test = global_data.cli.is_alive()31            if not test:32                try:33                    global_data.cli.close()34                except Exception, e:35                    traceback.print_exc()36                global_data.cli = zerorpc.Client(connect_to=RPC_URI)37                print ' global_data.cli  zerorpc.Client connect_to ', RPC_URI38            else:39                pass40        return global_data.cli41    def connect2(self):42        return zerorpc.Client(connect_to=RPC_URI,timeout=60, heartbeat=30)43    def __del__(self):44        if hasattr(global_data, "cli"):45            try:46                print ' global_data.cli  zerorpc.Client __del__  close'47                global_data.cli.close()48            except Exception, e:49                traceback.print_exc()50    def teardown(self, exception):51        '''52        æ¯æ¬¡è¿æ¥å®æ¯é½ä¼å³écliï¼æ¶èä¼æ¯è¾å¤§53        :param exception:54        :return:55        '''56        ctx = stack.top57        if hasattr(ctx, 'rpc_cli'):58            print "DEL RPC CLIENT"59            ctx.rpc_cli.close()60    @property61    def connection(self):62        ctx = stack.top63        if ctx is not None:64            if not hasattr(ctx, 'rpc_cli'):65                # ctx.rpc_cli = self.connect2()66                ctx.rpc_cli = self.connect()...

client.py

Source:client.py

1import base642import requests3class RPCClient(object):4    def __init__(5            self,6            host,7            auth_uri,8            username,9            password,10            rpc_uri,11            auth_host=None12    ):13        self.host = host14        self.auth_uri = auth_uri15        self.username = username16        self.password = password17        self.rpc_uri = rpc_uri18        self.auth_host = auth_host19        self.token = self.get_token()20    def prepare_headers(self):21        return {22            'Authorization': f'Bearer {self.token}',23            'Content-Type': 'application/json'24        }25    def get_token(self):26        url = self.auth_host if self.auth_host else self.host + self.auth_uri27        key = self.username28        secret = self.password29        auth_raw_string = "{}:{}".format(key, secret)30        auth_string = base64.b64encode(auth_raw_string.encode('utf-8')).decode()31        headers = {32            'Authorization': 'Basic {}'.format(auth_string)33        }34        response = requests.post(url, headers=headers)35        resj = response.json()36        access_token = resj.get('access_token')37        return access_token38    def post(self, headers, payload):39        default_headers = self.prepare_headers()40        default_headers.update(headers)41        response = requests.post(self.host + self.rpc_uri, headers=default_headers, json=payload)42        return response.json()43    def ping(self):44        response = requests.get(self.host + self.rpc_uri + '/ping')45        return response.json()46    def model(self, model, method, *args, **kwargs):47        data = {48            "type": "model",49            "model": model,50            "name": method,51            "args": args,52            "kwargs": kwargs53        }54        res = self.post({}, data)55        return res56    def request(self, name, *args, **kwargs):57        data = {58            "type": "request",59            "name": name,60            "args": args,61            "kwargs": kwargs62        }63        res = self.post({}, data)64        return res65if __name__ == '__main__':66    client = RPCClient(67        "http://localhost:5000",68        "/api/auth",69        "admin",70        "admin",71        "/jsonrpc"72    )73    client.model("demo.model", 'insert_data')74    client.request("test_api")75    res = client.ping()...

configconsole.py

Source:configconsole.py

1"""Console tool for Contour configuration."""2import click3from contour.localdata import config4@click.group()5def cli():6    """Utilities for Contour authority operators."""7@cli.command()8@click.argument('rpc_uri')9def btcrpc(rpc_uri):10    """Set the URI for the Bitcoin RPC interface."""11    config['btc_rpc_uri'] = rpc_uri12    config.write()...

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.