(url, method='GET', data=None, timeout=None)
Do HTTP Request
Do HTTP Request
Do HTTP Request
def _unsafe_httpdo(url, method='GET', data=None, timeout=None):
    """
    Do HTTP Request
    """
    start = time.time()
    if DEBUG:
        body = json.dumps(data) if data else ''
        print("Shell$ curl -X {method} -d '{body}' '{url}'".format(
            method=method.upper(), body=body or '', url=url))

    if timeout is None:
        timeout = HTTP_TIMEOUT
    try:
        u = urlparse(url)
        request_session = _requests_session_pool_get(u.scheme, u.netloc)
        response = request_session.request(method,
                                           url,
                                           json=data,
                                           timeout=timeout)
    except (requests.ConnectionError, requests.ReadTimeout) as e:
        raise

    if response.status_code == 502:  # Bad Gateway
        raise WDABadGateway(response.status_code, response.text)

    if DEBUG:
        ms = (time.time() - start) * 1000
        response_text = response.text
        if url.endswith("/screenshot"):
            response_text = response_text[:100] + "..." # limit length of screenshot response
        print('Return ({:.0f}ms): {}'.format(ms, response_text))

    try:
        retjson = response.json()
        retjson['status'] = retjson.get('status', 0)
        r = convert(retjson)

        if isinstance(r.value, dict) and r.value.get("error"):
            status = Status.ERROR
            value = r.value.copy()
            value.pop("traceback", None)

            for errCls in (WDAInvalidSessionIdError, WDAPossiblyCrashedError, WDAKeyboardNotPresentError, WDAUnknownError, WDAStaleElementReferenceError):
                if errCls.check(value):
                    raise errCls(status, value)

            raise WDARequestError(status, value)
        return r
    except JSONDecodeError:
        if response.text == "":
            raise WDAEmptyResponseError(method, url, data)
        raise WDAError(method, url, response.text[:100] + "...") # should not too long
    except requests.ConnectionError as e:
        raise WDAError("Failed to establish connection to to WDA")
def _unsafe_httpdo ( url , method = 'GET' , data = None , timeout = None ) : start = time . time ( ) if DEBUG : body = json . dumps ( data ) if data else '' print ( "Shell$ curl -X {method} -d '{body}' '{url}'" . format ( method = method . upper ( ) , body = body or '' , url = url ) ) if timeout is None : timeout = HTTP_TIMEOUT try : u = urlparse ( url ) request_session = _requests_session_pool_get ( u . scheme , u . netloc ) response = request_session . request ( method , url , json = data , timeout = timeout ) except ( requests . ConnectionError , requests . ReadTimeout ) as e : raise if response . status_code == 502 : # Bad Gateway raise WDABadGateway ( response . status_code , response . text ) if DEBUG : ms = ( time . time ( ) - start ) * 1000 response_text = response . text if url . endswith ( "/screenshot" ) : response_text = response_text [ : 100 ] + "..." # limit length of screenshot response print ( 'Return ({:.0f}ms): {}' . format ( ms , response_text ) ) try : retjson = response . json ( ) retjson [ 'status' ] = retjson . get ( 'status' , 0 ) r = convert ( retjson ) if isinstance ( r . value , dict ) and r . value . get ( "error" ) : status = Status . ERROR value = r . value . copy ( ) value . pop ( "traceback" , None ) for errCls in ( WDAInvalidSessionIdError , WDAPossiblyCrashedError , WDAKeyboardNotPresentError , WDAUnknownError , WDAStaleElementReferenceError ) : if errCls . check ( value ) : raise errCls ( status , value ) raise WDARequestError ( status , value ) return r except JSONDecodeError : if response . text == "" : raise WDAEmptyResponseError ( method , url , data ) raise WDAError ( method , url , response . text [ : 100 ] + "..." ) # should not too long except requests . ConnectionError as e : raise WDAError ( "Failed to establish connection to to WDA" )