_____________________ TestLagpattern.test_valid_lagpattern _____________________

self = <tests.api_tests.TestLagpattern testMethod=test_valid_lagpattern>

    def test_valid_lagpattern(self):

        """Test whether api.lagpattern is valid."""

        mysite = self.get_site()

        if mysite.siteinfo['dbrepllag'][0]['lag'] == -1:

            raise unittest.SkipTest(

                '{0} is not running on a replicated database cluster.'

                .format(mysite)

            )

        mythrottle = DummyThrottle(mysite)

        mysite._throttle = mythrottle

        params = {'action': 'query',

                  'titles': self.get_mainpage().title(),

                  'maxlag': -1}

        req = api.Request(site=mysite, parameters=params)

        try:

            req.submit()

        except SystemExit:

            pass  # expected exception from DummyThrottle instance

        except api.APIError as e:

            pywikibot.warning(

                'Wrong api.lagpattern regex, cannot retrieve lag value')

>           raise e

tests/api_tests.py:1066: 

_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

tests/api_tests.py:1060: in test_valid_lagpattern

    req.submit()

_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = pywikibot.data.api.Request<wsbeta:en->'/w/api.php?meta=userinfo&titles=Main+Page&maxlag=-1&action=query&rawcontinue=&format=json&uiprop=blockinfo|hasmsg'>

    def submit(self):

        """

            Submit a query and parse the response.

    

            @return: a dict containing data retrieved from api.php

            @rtype: dict

            """

        self._add_defaults()

        if (not config.enable_GET_without_SSL and

                self.site.protocol() != 'https' or

                self.site.is_oauth_token_available()):  # work around T108182

            use_get = False

        elif self.use_get is None:

            if self.action == 'query':

                # for queries check the query module

                modules = set()

                for mod_type_name in ('list', 'prop', 'generator'):

                    modules.update(self._params.get(mod_type_name, []))

            else:

                modules = set([self.action])

            if modules:

                self.site._paraminfo.fetch(modules)

                use_get = all('mustbeposted' not in self.site._paraminfo[mod]

                              for mod in modules)

            else:

                # If modules is empty, just 'meta' was given, which doesn't

                # require POSTs, and is required for ParamInfo

                use_get = True

        else:

            use_get = self.use_get

        while True:

            paramstring = self._http_param_string()

            simulate = self._simulate(self.action)

            if simulate:

                return simulate

            if self.throttle:

                self.site.throttle(write=self.write)

            else:

                pywikibot.log(

                    "Submitting unthrottled action '{0}'.".format(self.action))

            uri = self.site.scriptpath() + "/api.php"

            try:

                if self.mime:

                    (headers, body) = Request._build_mime_request(

                        self._encoded_items(), self.mime_params)

                    use_get = False  # MIME requests require HTTP POST

                else:

                    headers = {'Content-Type': 'application/x-www-form-urlencoded'}

                    if (not self.site.maximum_GET_length() or

                            self.site.maximum_GET_length() < len(paramstring)):

                        use_get = False

                    if use_get:

                        uri = '{0}?{1}'.format(uri, paramstring)

                        body = None

                    else:

                        body = paramstring

    

                pywikibot.debug('API request to {0} (uses get: {1}):\n'

                                'Headers: {2!r}\nURI: {3!r}\n'

                                'Body: {4!r}'.format(self.site, use_get,

                                                     headers, uri, body),

                                _logger)

    

                rawdata = http.request(

                    site=self.site, uri=uri, method='GET' if use_get else 'POST',

                    body=body, headers=headers)

            except Server504Error:

                pywikibot.log(u"Caught HTTP 504 error; retrying")

                self.wait()

                continue

            except Server414Error:

                if use_get:

                    pywikibot.log('Caught HTTP 414 error; retrying')

                    use_get = False

                    self.wait()

                    continue

                else:

                    pywikibot.warning('Caught HTTP 414 error, although not '

                                      'using GET.')

                    raise

            except FatalServerError:

                # This error is not going to be fixed by just waiting

                pywikibot.error(traceback.format_exc())

                raise

            # TODO: what other exceptions can occur here?

            except Exception:

                # for any other error on the http request, wait and retry

                pywikibot.error(traceback.format_exc())

                pywikibot.log(u"%s, %s" % (uri, paramstring))

                self.wait()

                continue

            if not isinstance(rawdata, unicode):

                rawdata = rawdata.decode(self.site.encoding())

            pywikibot.debug((u"API response received from %s:\n" % self.site) +

                            rawdata, _logger)

            if rawdata.startswith(u"unknown_action"):

                raise APIError(rawdata[:14], rawdata[16:])

            try:

                result = json.loads(rawdata)

            except ValueError:

                # if the result isn't valid JSON, there must be a server

                # problem. Wait a few seconds and try again

                pywikibot.warning(

                    "Non-JSON response received from server %s; the server may be down."

                    % self.site)

                # there might also be an overflow, so try a smaller limit

                for param in self._params:

                    if param.endswith("limit"):

                        # param values are stored a list of str

                        value = self._params[param][0]

                        try:

                            self._params[param] = [str(int(value) // 2)]

                            pywikibot.output(u"Set %s = %s"

                                             % (param, self._params[param]))

                        except:

                            pass

                self.wait()

                continue

            if not result:

                result = {}

            if not isinstance(result, dict):

                raise APIError("Unknown",

                               "Unable to process query response of type %s."

                               % type(result),

                               data=result)

    

            if self.action == 'query' and 'userinfo' in result.get('query', ()):

                # if we get passed userinfo in the query result, we can confirm

                # that we are logged in as the correct user. If this is not the

                # case, force a re-login.

                username = result['query']['userinfo']['name']

                if self.site.user() is not None and self.site.user() != username:

                    pywikibot.error(

                        "Logged in as '{actual}' instead of '{expected}'. "

                        "Forcing re-login.".format(

                            actual=username,

                            expected=self.site.user()

                        )

                    )

                    self.site._relogin()

                    continue

    

            self._handle_warnings(result)

    

            if "error" not in result:

                return result

    

            error = result['error'].copy()

            for key in result:

                if key in ('error', 'warnings'):

                    continue

                assert key not in error

                assert isinstance(result[key], basestring), \

                    'Unexpected %s: %r' % (key, result[key])

                error[key] = result[key]

    

            if "*" in result["error"]:

                # help text returned

                result['error']['help'] = result['error'].pop("*")

            code = result['error'].setdefault('code', 'Unknown')

            info = result['error'].setdefault('info', None)

    

            # Older wikis returned an error instead of a warning when

            # the request asked for too many values. If we get this

            # error, assume we are not logged in (we can't check this

            # because the userinfo data is not present) and force

            # a re-login

            if code.endswith('limit'):

                pywikibot.error("Received API limit error. Forcing re-login")

                self.site._relogin()

                continue

    

            # If the user assertion failed, we're probably logged out as well.

            if code == 'assertuserfailed':

                pywikibot.error("User assertion failed. Forcing re-login.")

                self.site._relogin()

                continue

    

            # Lastly, the purge module require a POST if used as anonymous user,

            # but we normally send a GET request. If the API tells us the request

            # has to be POSTed, we're probably logged out.

            if code == 'mustbeposted' and self.action == 'purge':

                pywikibot.error("Received unexpected 'mustbeposted' error. "

                                "Forcing re-login.")

                self.site._relogin()

                continue

    

            if code == "maxlag":

                lag = lagpattern.search(info)

                if lag:

                    pywikibot.log(

                        u"Pausing due to database lag: " + info)

                    self.site.throttle.lag(int(lag.group("lag")))

                    continue

            elif code == 'help' and self.action == 'help':

                # The help module returns an error result with the complete

                # API information. As this data was requested, return the

                # data instead of raising an exception.

                return {'help': {'mime': 'text/plain',

                                 'help': result['error']['help']}}

    

            pywikibot.warning('API error %s: %s' % (code, info))

    

            if code.startswith('internal_api_error_') or code == 'readonly':

                if code == 'readonly':  # T154011

                    class_name = code

                else:

                    class_name = code[len(u'internal_api_error_'):]

    

                del error['code']  # is added via class_name

                e = APIMWException(class_name, **error)

    

                # If the error key is in this table, it is probably a temporary problem,

                # so we will retry the edit.

                # TODO: T154011: 'ReadOnlyError' seems replaced by 'readonly'

                retry = class_name in ['DBConnectionError',  # T64974

                                       'DBQueryError',  # T60158

                                       'ReadOnlyError',  # T61227

                                       'readonly',  # T154011

                                       ]

    

                pywikibot.error("Detected MediaWiki API exception %s%s"

                                % (e,

                                   "; retrying" if retry else "; raising"))

                # Due to bug T66958, Page's repr may return non ASCII bytes

                # Get as bytes in PY2 and decode with the console encoding as

                # the rest should be ASCII anyway.

                param_repr = str(self._params)

                if PY2:

                    param_repr = param_repr.decode(config.console_encoding)

                pywikibot.log(u"MediaWiki exception %s details:\n"

                              u"          query=\n%s\n"

                              u"          response=\n%s"

                              % (class_name,

                                 pprint.pformat(param_repr),

                                 result))

    

                if retry:

                    self.wait()

                    continue

    

                raise e

    

            # Phab. tickets T48535, T64126, T68494, T68619

            if code == "failed-save" and \

               self.action == 'wbeditentity' and \

               self._is_wikibase_error_retryable(result["error"]):

                self.wait()

                continue

            # If readapidenied is returned try to login

            if code == 'readapidenied' and self.site._loginstatus in (-3, -1):

                self.site.login()

                continue

            if code == 'badtoken':

                user_tokens = self.site.tokens._tokens[self.site.user()]

                # all token values mapped to their type

                tokens = dict((token, t_type)

                              for t_type, token in user_tokens.items())

                # determine which tokens are bad

                invalid_param = {}

                for name, param in self._params.items():

                    if len(param) == 1 and param[0] in tokens:

                        invalid_param[name] = tokens[param[0]]

                # doesn't care about the cache so can directly load them

                if invalid_param:

                    pywikibot.log(

                        u'Bad token error for {0}. Tokens for "{1}" used in '

                        u'request; invalidated them.'.format(

                            self.site.user(),

                            '", "'.join(sorted(set(invalid_param.values())))))

                    self.site.tokens.load_tokens(set(invalid_param.values()))

                    # fix parameters; lets hope that it doesn't mistake actual

                    # parameters as tokens

                    for name, t_type in invalid_param.items():

                        self[name] = self.site.tokens[t_type]

                    continue

                else:

                    # otherwise couldn't find any … weird there is nothing what

                    # can be done here because it doesn't know which parameters

                    # to fix

                    pywikibot.log(

                        u'Bad token error for {0} but no parameter is using a '

                        u'token. Current tokens: {1}'.format(

                            self.site.user(),

                            ', '.join('{0}: {1}'.format(*e)

                                      for e in user_tokens.items())))

            if 'mwoauth-invalid-authorization' in code:

                if 'Nonce already used' in info:

                    pywikibot.error(

                        'Retrying failed OAuth authentication for {0}: {1}'

                        .format(self.site, info))

                    continue

                raise NoUsername('Failed OAuth authentication for %s: %s'

                                 % (self.site, info))

            if code == 'cirrussearch-too-busy-error':  # T170647

                self.wait()

                continue

            # raise error

            try:

                # Due to bug T66958, Page's repr may return non ASCII bytes

                # Get as bytes in PY2 and decode with the console encoding as

                # the rest should be ASCII anyway.

                param_repr = str(self._params)

                if PY2:

                    param_repr = param_repr.decode(config.console_encoding)

                pywikibot.log(u"API Error: query=\n%s"

                              % pprint.pformat(param_repr))

                pywikibot.log(u"           response=\n%s"

                              % result)

    

>               raise APIError(**result['error'])

E               pywikibot.data.api.APIError: maxlag: Waiting for 10.68.18.35:3306: 0 seconds lagged. [host:10.68.18.35:3306; lag:0; type:db; help:See https://en.wikisource.beta.wmflabs.org/w/api.php for API usage. Subscribe to the mediawiki-api-announce mailing list at &lt;https://lists.wikimedia.org/mailman/listinfo/mediawiki-api-announce&gt; for notice of API deprecations and breaking changes.]

pywikibot/data/api.py:2197: APIError