Class AuthServiceProxy

test/functional/test_framework/authproxy.py:65–180

    raise TypeError(repr(o) + " is not JSON serializable")

class AuthServiceProxy():
    __id_count = 0

    # ensure_ascii: escape unicode as \uXXXX, passed to json.dumps
    def __init__(self, service_url, service_name=None, timeout=HTTP_TIMEOUT, connection=None, ensure_ascii=True):
        self.__service_url = service_url
        self._service_name = service_name
        self.ensure_ascii = ensure_ascii  # can be toggled on the fly by tests
        self.__url = urllib.parse.urlparse(service_url)
        port = 80 if self.__url.port is None else self.__url.port
        user = None if self.__url.username is None else self.__url.username.encode('utf8')
        passwd = None if self.__url.password is None else self.__url.password.encode('utf8')
        authpair = user + b':' + passwd
        self.__auth_header = b'Basic ' + base64.b64encode(authpair)

        if connection:
            # Callables re-use the connection of the original proxy
            self.__conn = connection
        elif self.__url.scheme == 'https':
            self.__conn = http.client.HTTPSConnection(self.__url.hostname, port, timeout=timeout)
        else:
            self.__conn = http.client.HTTPConnection(self.__url.hostname, port, timeout=timeout)

    def __getattr__(self, name):
        if name.startswith('__') and name.endswith('__'):
            # Python internal stuff
            raise AttributeError
        if self._service_name is not None:
            name = "%s.%s" % (self._service_name, name)
        return AuthServiceProxy(self.__service_url, name, connection=self.__conn)

    def _request(self, method, path, postdata):
        '''
        Do a HTTP request, with retry if we get disconnected (e.g. due to a timeout).
        This is a workaround for https://bugs.python.org/issue3566 which is fixed in Python 3.5.
        '''
        headers = {'Host': self.__url.hostname,
                   'User-Agent': USER_AGENT,
                   'Authorization': self.__auth_header,
                   'Content-type': 'application/json'}
        try:
            self.__conn.request(method, path, postdata, headers)
            return self._get_response()
        except http.client.BadStatusLine as e:
            if e.line == "''":  # if connection was closed, try again
                self.__conn.close()
                self.__conn.request(method, path, postdata, headers)
                return self._get_response()
            else:
                raise
        except (BrokenPipeError, ConnectionResetError):
            # Python 3.5+ raises BrokenPipeError instead of BadStatusLine when the connection was reset
            # ConnectionResetError happens on FreeBSD with Python 3.4
            self.__conn.close()
            self.__conn.request(method, path, postdata, headers)
            return self._get_response()

    def get_request(self, *args, **argsn):