<commit_before>#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
::Timeout

atx-agent:ReverseProxy use http.DefaultTransport. Default Timeout: 30s

|-- Dial --|-- TLS handshake --|-- Request --|-- Resp.headers --|-- Respose.body --|
|------------------------------ http.Client.Timeout -------------------------------|

Refs:
    - https://golang.org/pkg/net/http/#RoundTripper
    - http://colobu.com/2016/07/01/the-complete-guide-to-golang-net-http-timeouts
"""

from __future__ import absolute_import, print_function

import base64
import contextlib
import functools
import hashlib
import io
import json
import logging
import os
import re
import shutil
import subprocess
import sys
import threading
import time
import warnings
import xml.dom.minidom
from collections import namedtuple
from datetime import datetime
from typing import List, Optional, Tuple, Union

# import progress.bar
import adbutils
import packaging
import requests
import six
import six.moves.urllib.parse as urlparse
from cached_property import cached_property
from deprecated import deprecated
from logzero import setup_logger
from PIL import Image
from retry import retry
from urllib3.util.retry import Retry

from . import xpath
from ._selector import Selector, UiObject
from .exceptions import (BaseError, ConnectError, GatewayError, JSONRPCError,
                         NullObjectExceptionError, NullPointerExceptionError,
                         RetryError, ServerError, SessionBrokenError,
                         StaleObjectExceptionError,
                         UiAutomationNotConnectedError, UiObjectNotFoundError)
from .init import Initer
# from .session import Session  # noqa: F401
from .settings import Settings
from .swipe import SwipeExt
from .utils import list2cmdline
from .version import __atx_agent_version__, __apk_version__
from .watcher import Watcher

if six.PY2:
    FileNotFoundError = OSError

DEBUG = False
HTTP_TIMEOUT = 60
WAIT_FOR_DEVICE_TIMEOUT = int(os.getenv("WAIT_FOR_DEVICE_TIMEOUT", 70))

# logger = logging.getLogger("uiautomator2")

logger = setup_logger("uiautomator2", level=logging.DEBUG)
_mswindows = (os.name == "nt")


class TimeoutRequestsSession(requests.Session):
    def __init__(self):
        super(TimeoutRequestsSession, self).__init__()
        # refs: https://stackoverflow.com/questions/33895739/python-requests-cant-load-any-url-remote-end-closed-connection-without-respo
        # refs: https://stackoverflow.com/questions/15431044/can-i-set-max-retries-for-requests-request
        
        # Is retry necessary, maybe not, so I closed it at 2020/05/29
        # retries = Retry(total=3, connect=3, backoff_factor=0.5)
        # adapter = requests.adapters.HTTPAdapter(max_retries=retries)
        # self.mount("http://", adapter)
        # self.mount("https://", adapter)

    def request(self, method, url, **kwargs):
        # Init timeout and set connect-timeout to 3s
        if 'timeout' not in kwargs:
            kwargs['timeout'] = (3, HTTP_TIMEOUT)
        if isinstance(kwargs['timeout'], (int, float)):
            kwargs['timeout'] = (3, kwargs['timeout'])

        verbose = hasattr(self, 'debug') and self.debug
        if verbose:
            data = kwargs.get('data') or '""'
            if isinstance(data, dict):
                data = json.dumps(data)
            time_start = time.time()
            print(datetime.now().strftime("%H:%M:%S.%f")[:-3],
                  "$ curl -X {method} -d '{data}' '{url}'".format(
                      method=method, url=url, data=data))  # yaml: disable
        try:
            resp = super(TimeoutRequestsSession,
                         self).request(method, url, **kwargs)
        except requests.ConnectionError as e:
            # High possibly atx-agent is down
            raise
        else:
            if verbose:
                print(
                    datetime.now().strftime("%H:%M:%S.%f")[:-3],
                    "Response (%d ms) >>>\n" %
                    ((time.time() - time_start) * 1000) + resp.text.rstrip() +
                    "\n<<< END")

            from types import MethodType

            def raise_for_status(_self):
                if _self.status_code != 200:
                    raise requests.HTTPError(_self.status_code, _self.text)

            resp.raise_for_status = MethodType(raise_for_status, resp)
            return resp


ShellResponse = namedtuple("ShellResponse", ("output", "exit_code"))
_tmq_production = (os.environ.get("TMQ") == "true")


class _Service(object):
    def __init__(self, name, u2obj: "Device"):
        self.name = name
        # FIXME(ssx): support other service: minicap, minitouch
        assert name == 'uiautomator'
        self.u2obj = u2obj
        self.service_url = self.u2obj.path2url("/services/" + name)

    def _raise_for_status(self, res: requests.Response):
        if res.status_code != 200:
            if res.headers['content-type'].startswith("application/json"):
                raise RuntimeError(res.json()["description"])
            warnings.warn(res.text)
            res.raise_for_status()

    def start(self):
        """
        Manually run with the following command:
            adb shell am instrument -w -r -e debug false -e class com.github.uiautomator.stub.Stub \
                com.github.uiautomator.test/android.support.test.runner.AndroidJUnitRunner
        """
        # kill uiautomator
        res = self.u2obj.http.post(self.service_url)
        self._raise_for_status(res)

    def stop(self):
        """
        1. stop command which launched with uiautomator 1.0
            Eg: adb shell uiautomator runtest androidUiAutomator.jar
        """
        res = self.u2obj.http.delete(self.service_url)
        self._raise_for_status(res)

    def running(self) -> bool:
        res = self.u2obj.http.get(self.service_url)
        self._raise_for_status(res)
        return res.json().get("running")


class _AgentRequestSession(TimeoutRequestsSession):
    def __init__(self, clnt: "_BaseClient"):
        super().__init__()
        self.__client = clnt

    def request(self, method, url, **kwargs):
        retry = kwargs.pop("retry", True)
        try:
            url = self.__client.path2url(url) # may raise adbutils.AdbError when device offline
            return super().request(method, url, **kwargs)
        except (requests.ConnectionError, requests.ReadTimeout, adbutils.AdbError) as e:
            if not retry:
                raise
            
            # if atx-agent is already running, just raise error
            if isinstance(e, requests.RequestException) and \
                self.__client._is_agent_alive(): 
                raise

        logger.warning("atx-agent has something wrong, auto recovering")
        # ReadTimeout: sometime means atx-agent is running but not responsing
        # one reason is futex_wait_queue: https://stackoverflow.com/questions/9801256/app-hangs-on-futex-wait-queue-me-every-a-couple-of-minutes
        
        # fix atx-agent and request again
        self.__client._prepare_atx_agent()
        url = self.__client.path2url(url)
        return super().request(method, url, **kwargs)


class _BaseClient(object):
    """
    提供最基础的控制类，这个类暂时先不公开吧
    """
    def __init__(self, serial_or_url: Optional[str] = None):
        """
        Args:
            serial_or_url: device serialno or atx-agent base url<commit_msg>raise EnvironmentError directly when connected with wifi, but atx-agent is down<commit_after>#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
::Timeout

atx-agent:ReverseProxy use http.DefaultTransport. Default Timeout: 30s

|-- Dial --|-- TLS handshake --|-- Request --|-- Resp.headers --|-- Respose.body --|
|------------------------------ http.Client.Timeout -------------------------------|

Refs:
    - https://golang.org/pkg/net/http/#RoundTripper
    - http://colobu.com/2016/07/01/the-complete-guide-to-golang-net-http-timeouts
"""

from __future__ import absolute_import, print_function

import base64
import contextlib
import functools
import hashlib
import io
import json
import logging
import os
import re
import shutil
import subprocess
import sys
import threading
import time
import warnings
import xml.dom.minidom
from collections import namedtuple
from datetime import datetime
from typing import List, Optional, Tuple, Union

# import progress.bar
import adbutils
import packaging
import requests
import six
import six.moves.urllib.parse as urlparse
from cached_property import cached_property
from deprecated import deprecated
from logzero import setup_logger
from PIL import Image
from retry import retry
from urllib3.util.retry import Retry

from . import xpath
from ._selector import Selector, UiObject
from .exceptions import (BaseError, ConnectError, GatewayError, JSONRPCError,
                         NullObjectExceptionError, NullPointerExceptionError,
                         RetryError, ServerError, SessionBrokenError,
                         StaleObjectExceptionError,
                         UiAutomationNotConnectedError, UiObjectNotFoundError)
from .init import Initer
# from .session import Session  # noqa: F401
from .settings import Settings
from .swipe import SwipeExt
from .utils import list2cmdline
from .version import __atx_agent_version__, __apk_version__
from .watcher import Watcher

if six.PY2:
    FileNotFoundError = OSError

DEBUG = False
HTTP_TIMEOUT = 60
WAIT_FOR_DEVICE_TIMEOUT = int(os.getenv("WAIT_FOR_DEVICE_TIMEOUT", 70))

# logger = logging.getLogger("uiautomator2")

logger = setup_logger("uiautomator2", level=logging.DEBUG)
_mswindows = (os.name == "nt")


class TimeoutRequestsSession(requests.Session):
    def __init__(self):
        super(TimeoutRequestsSession, self).__init__()
        # refs: https://stackoverflow.com/questions/33895739/python-requests-cant-load-any-url-remote-end-closed-connection-without-respo
        # refs: https://stackoverflow.com/questions/15431044/can-i-set-max-retries-for-requests-request
        
        # Is retry necessary, maybe not, so I closed it at 2020/05/29
        # retries = Retry(total=3, connect=3, backoff_factor=0.5)
        # adapter = requests.adapters.HTTPAdapter(max_retries=retries)
        # self.mount("http://", adapter)
        # self.mount("https://", adapter)

    def request(self, method, url, **kwargs):
        # Init timeout and set connect-timeout to 3s
        if 'timeout' not in kwargs:
            kwargs['timeout'] = (3, HTTP_TIMEOUT)
        if isinstance(kwargs['timeout'], (int, float)):
            kwargs['timeout'] = (3, kwargs['timeout'])

        verbose = hasattr(self, 'debug') and self.debug
        if verbose:
            data = kwargs.get('data') or '""'
            if isinstance(data, dict):
                data = json.dumps(data)
            time_start = time.time()
            print(datetime.now().strftime("%H:%M:%S.%f")[:-3],
                  "$ curl -X {method} -d '{data}' '{url}'".format(
                      method=method, url=url, data=data))  # yaml: disable
        try:
            resp = super(TimeoutRequestsSession,
                         self).request(method, url, **kwargs)
        except requests.ConnectionError as e:
            # High possibly atx-agent is down
            raise
        else:
            if verbose:
                print(
                    datetime.now().strftime("%H:%M:%S.%f")[:-3],
                    "Response (%d ms) >>>\n" %
                    ((time.time() - time_start) * 1000) + resp.text.rstrip() +
                    "\n<<< END")

            from types import MethodType

            def raise_for_status(_self):
                if _self.status_code != 200:
                    raise requests.HTTPError(_self.status_code, _self.text)

            resp.raise_for_status = MethodType(raise_for_status, resp)
            return resp


ShellResponse = namedtuple("ShellResponse", ("output", "exit_code"))
_tmq_production = (os.environ.get("TMQ") == "true")


class _Service(object):
    def __init__(self, name, u2obj: "Device"):
        self.name = name
        # FIXME(ssx): support other service: minicap, minitouch
        assert name == 'uiautomator'
        self.u2obj = u2obj
        self.service_url = self.u2obj.path2url("/services/" + name)

    def _raise_for_status(self, res: requests.Response):
        if res.status_code != 200:
            if res.headers['content-type'].startswith("application/json"):
                raise RuntimeError(res.json()["description"])
            warnings.warn(res.text)
            res.raise_for_status()

    def start(self):
        """
        Manually run with the following command:
            adb shell am instrument -w -r -e debug false -e class com.github.uiautomator.stub.Stub \
                com.github.uiautomator.test/android.support.test.runner.AndroidJUnitRunner
        """
        # kill uiautomator
        res = self.u2obj.http.post(self.service_url)
        self._raise_for_status(res)

    def stop(self):
        """
        1. stop command which launched with uiautomator 1.0
            Eg: adb shell uiautomator runtest androidUiAutomator.jar
        """
        res = self.u2obj.http.delete(self.service_url)
        self._raise_for_status(res)

    def running(self) -> bool:
        res = self.u2obj.http.get(self.service_url)
        self._raise_for_status(res)
        return res.json().get("running")


class _AgentRequestSession(TimeoutRequestsSession):
    def __init__(self, clnt: "_BaseClient"):
        super().__init__()
        self.__client = clnt

    def request(self, method, url, **kwargs):
        retry = kwargs.pop("retry", True)
        try:
            url = self.__client.path2url(url) # may raise adbutils.AdbError when device offline
            return super().request(method, url, **kwargs)
        except (requests.ConnectionError, requests.ReadTimeout, adbutils.AdbError) as e:
            if not retry:
                raise
            
            # if atx-agent is already running, just raise error
            if isinstance(e, requests.RequestException) and \
                self.__client._is_agent_alive(): 
                raise

        if not self.__client._serial:
            raise EnvironmentError("http-request to atx-agent error, can only recover from USB")

        logger.warning("atx-agent has something wrong, auto recovering")
        # ReadTimeout: sometime means atx-agent is running but not responsing
        # one reason is futex_wait_queue: https://stackoverflow.com/questions/9801256/app-hangs-on-futex-wait-queue-me-every-a-couple-of-minutes
        
        # fix atx-agent and request again
        self.__client._prepare_atx_agent()
        url = self.__client.path2url(url)
        return super().request(method, url, **kwargs)


class _BaseClient(object):
    """
    提供最基础的控制类，这个类暂时先不公开吧
    """
    def __init__(self, serial_or_url: Optional[str] = None):
        """
        Args:
            serial_or_url: device serialno or atx-agent base url
{'fraction_non_alphanumeric': 0.1153487131617549, 'fraction_numerical': 0.011734897986398186, 'mean_word_length': 2.459178966789668, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 11, 'https://': 10, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2233119', 'n_tokens_mistral': 4496, 'n_tokens_neox': 4331, 'n_words': 1056}